Skip to main content

MCP Policy Enforcement

AxonFlow enforces policies at multiple levels during MCP connector operations:

  1. REQUEST Phase: Evaluates incoming queries before they reach the connector
  2. RESPONSE Phase: Scans connector responses before returning to the client
  3. Exfiltration Detection: Prevents large-scale data extraction via row/volume limits
  4. Dynamic Policies: Orchestrator-evaluated policies for rate limiting, budgets, and access control

This provides comprehensive security for both data input and output.

Phase-Aware Evaluation​

Request Phase​

The request phase evaluates queries before they are sent to the connector:

  • SQL Injection Detection: Blocks malicious SQL patterns (DROP, DELETE without WHERE, UNION injection)
  • Dangerous Operation Blocking: Prevents schema modifications, privilege escalation
  • Access Control: Enforces tenant-specific permissions

If a policy blocks the request, the connector is never called and a 403 response is returned.

Response Phase​

The response phase evaluates data after it's returned from the connector:

  • PII Detection: Identifies sensitive data patterns (SSN, credit cards, national IDs)
  • Data Redaction: Replaces sensitive values with redacted placeholders
  • Audit Logging: Records what data was accessed and redacted

Response Schema​

All MCP responses now include policy enforcement metadata:

{
  "success": true,
  "connector": "postgres-demo",
  "data": [
    {
      "name": "John Doe",
      "ssn": "[REDACTED]",
      "email": "[email protected]"
    }
  ],
  "row_count": 1,
  "duration_ms": 45,
  "redacted": true,
  "redacted_fields": ["data[0].ssn"],
  "policy_info": {
    "policies_evaluated": 15,
    "blocked": false,
    "redactions_applied": 1,
    "processing_time_ms": 3,
    "matched_policies": [
      {
        "policy_id": "pii-us-ssn",
        "policy_name": "US Social Security Number",
        "category": "pii-us",
        "severity": "critical",
        "action": "redact"
      }
    ],
    "exfiltration_check": {
      "rows_returned": 1,
      "row_limit": 10000,
      "bytes_returned": 256,
      "byte_limit": 10485760,
      "within_limits": true
    },
    "dynamic_policy_info": {
      "policies_evaluated": 2,
      "matched_policies": [
        {
          "policy_id": "rate-limit-1",
          "policy_name": "API Rate Limit",
          "policy_type": "rate-limit",
          "action": "allow"
        }
      ],
      "orchestrator_reachable": true,
      "processing_time_ms": 3
    }
  }
}

New Response Fields​

FieldTypeDescription
redactedbooleanWhether any fields were redacted
redacted_fieldsstring[]JSON paths of redacted fields
policy_infoobjectPolicy evaluation details

PolicyInfo Object​

FieldTypeDescription
policies_evaluatedintegerNumber of policies checked
blockedbooleanWhether request was blocked
block_reasonstringReason if blocked
redactions_appliedintegerNumber of fields redacted
processing_time_msintegerPolicy evaluation time
matched_policiesarrayPolicies that matched

PolicyMatchInfo Object​

Each entry in matched_policies contains:

FieldTypeDescription
policy_idstringUnique policy identifier
policy_namestringHuman-readable policy name
categorystringPolicy category (e.g., pii-us, security-sqli)
severitystringMatch severity (critical, high, medium, low)
actionstringAction taken (block, redact, warn, log)

ExfiltrationCheckInfo Object​

Tracks row and volume limits to prevent data exfiltration:

FieldTypeDescription
rows_returnedintegerNumber of rows in the response
row_limitintegerConfigured maximum rows per query
bytes_returnedintegerResponse data size in bytes
byte_limitintegerConfigured maximum bytes per response
within_limitsbooleanWhether response is within all limits

DynamicPolicyInfo Object​

Contains results of Orchestrator-evaluated dynamic policies:

FieldTypeDescription
policies_evaluatedintegerNumber of dynamic policies checked
matched_policiesarrayDynamic policies that matched
orchestrator_reachablebooleanWhether Orchestrator was reachable
processing_time_msintegerDynamic policy evaluation time

DynamicPolicyMatch Object​

Each entry in dynamic matched_policies contains:

FieldTypeDescription
policy_idstringUnique policy identifier
policy_namestringHuman-readable policy name
policy_typestringPolicy type (rate-limit, budget, time-access, role-access)
actionstringAction taken (allow, block, log)
reasonstringContext for the policy match

Policy Categories​

Security Policies (Request Phase)​

CategoryDescriptionAction
security-sqliSQL injection patternsBlock
security-dangerousDDL, privilege escalationBlock

PII Policies (Response Phase)​

CategoryDescriptionAction
pii-usUS SSN, Driver's LicenseRedact
pii-globalCredit cards, email, phoneRedact
pii-indiaAadhaar, PANRedact
pii-euNational IDs, IBANRedact

SDK Integration​

Go SDK​

resp, err := client.MCPQuery(ctx, axonflow.MCPQueryRequest{
    Connector: "postgres-demo",
    Statement: "SELECT name, ssn FROM customers",
})

if resp.Redacted {
    fmt.Printf("Redacted fields: %v\n", resp.RedactedFields)
}

if resp.PolicyInfo != nil {
    fmt.Printf("Evaluated %d policies in %dms\n",
        resp.PolicyInfo.PoliciesEvaluated,
        resp.PolicyInfo.ProcessingTimeMs)
}

TypeScript SDK​

const resp = await client.mcpQuery({
  connector: "postgres-demo",
  statement: "SELECT name, ssn FROM customers",
});

if (resp.redacted) {
  console.log("Redacted fields:", resp.redacted_fields);
}

if (resp.policy_info) {
  console.log(`Evaluated ${resp.policy_info.policies_evaluated} policies`);
}

Python SDK​

resp = await client.mcp_query(
    connector="postgres-demo",
    statement="SELECT name, ssn FROM customers"
)

if resp.redacted:
    print(f"Redacted fields: {resp.redacted_fields}")

if resp.policy_info:
    print(f"Evaluated {resp.policy_info.policies_evaluated} policies")

Java SDK​

ConnectorResponse resp = client.mcpQuery(
    "postgres-demo",
    "SELECT name, ssn FROM customers"
);

if (resp.isRedacted()) {
    System.out.println("Redacted fields: " + resp.getRedactedFields());
}

ConnectorPolicyInfo info = resp.getPolicyInfo();
if (info != null) {
    System.out.println("Evaluated " + info.getPoliciesEvaluated() + " policies");
}

Handling Blocked Requests​

When a request is blocked by policy, the SDK throws an exception or returns an error:

// Go
resp, err := client.MCPQuery(ctx, req)
if err != nil {
    // Check if it's a policy violation
    if policyErr, ok := err.(*axonflow.PolicyViolationError); ok {
        fmt.Printf("Blocked by policy: %s\n", policyErr.Reason)
    }
}
// TypeScript
try {
  const resp = await client.mcpQuery(req);
} catch (err) {
  if (err instanceof PolicyViolationError) {
    console.log("Blocked by policy:", err.reason);
  }
}

Exfiltration Detection​

Exfiltration detection prevents large-scale data extraction via MCP queries by enforcing row count and data volume limits.

Configuration​

VariableDefaultDescription
MCP_MAX_ROWS_PER_QUERY10000Maximum rows per MCP query
MCP_MAX_BYTES_PER_QUERY10485760Maximum bytes per response (10MB)

Behavior​

  • Responses within limits include exfiltration_check with within_limits: true
  • Responses exceeding limits return 403 with details about which limit was exceeded
  • Limits are enforced after PII redaction (redacted data counts toward limits)

Example: Checking Exfiltration Limits​

resp, err := client.MCPQuery(ctx, req)
if resp.PolicyInfo != nil && resp.PolicyInfo.ExfiltrationCheck != nil {
    info := resp.PolicyInfo.ExfiltrationCheck
    fmt.Printf("Rows: %d/%d, Bytes: %d/%d\n",
        info.RowsReturned, info.RowLimit,
        info.BytesReturned, info.ByteLimit)
    if !info.WithinLimits {
        fmt.Println("Warning: Response exceeded limits")
    }
}

Dynamic Policy Evaluation​

Dynamic policies are evaluated by the Orchestrator and support advanced use cases like rate limiting, budget controls, and access restrictions.

Configuration​

VariableDefaultDescription
MCP_DYNAMIC_POLICIES_ENABLEDfalseEnable dynamic policy evaluation
MCP_DYNAMIC_POLICIES_TIMEOUT5sTimeout for Orchestrator call
MCP_DYNAMIC_POLICIES_GRACEFULtrueContinue if Orchestrator unavailable

Supported Policy Types​

TypeDescriptionExample
rate-limitLimit requests per time windowMax 100 queries per minute
budgetTrack and limit API costs$1000 monthly budget
time-accessRestrict to business hoursAllow 9am-5pm only
role-accessRole-based access controlAdmin-only connectors

Graceful Degradation​

When MCP_DYNAMIC_POLICIES_GRACEFUL=true (default), MCP queries continue even if the Orchestrator is unavailable. The response will include orchestrator_reachable: false in the dynamic_policy_info.

Example: Checking Dynamic Policy Results​

resp, err := client.MCPQuery(ctx, req)
if resp.PolicyInfo != nil && resp.PolicyInfo.DynamicPolicyInfo != nil {
    info := resp.PolicyInfo.DynamicPolicyInfo
    fmt.Printf("Evaluated %d dynamic policies\n", info.PoliciesEvaluated)
    for _, match := range info.MatchedPolicies {
        fmt.Printf("  - %s (%s): %s\n",
            match.PolicyName, match.PolicyType, match.Action)
    }
}

Performance​

The policy engine is designed for minimal latency impact:

  • Request Phase: <5ms p99 for policy evaluation
  • Response Phase: <10ms p99 for redaction
  • Caching: Compiled regex patterns cached for reuse
  • Parallel Evaluation: Policies evaluated concurrently

Configuration​

Environment Variables​

VariableDefaultDescription
POLICY_CACHE_TTL5mPolicy cache refresh interval
POLICY_BLOCK_ON_ERRORfalseBlock if policy engine unavailable
PII_ACTIONredactDefault PII action (block/redact/warn/log)

Database Configuration​

Policies are stored in the static_policies table with phase-aware columns:

SELECT id, name, category, phase, action_request, action_response
FROM static_policies
WHERE enabled = true AND tenant_id = 'your-tenant';

Audit Logging​

All MCP connector operations are logged for compliance and security analysis. The mcp_query_audits table captures:

  • Request phase results: Whether the query was blocked, which policies matched
  • Response phase results: Whether PII was redacted, which fields were affected
  • Exfiltration checks: Row counts and volume limit violations
  • Performance metrics: Operation duration, success/failure status

The audit_id field in PolicyInfo correlates with audit log entries for end-to-end traceability:

if resp.PolicyInfo != nil {
    // Use audit_id to look up the full audit record
    fmt.Printf("Audit ID: %s\n", resp.PolicyInfo.AuditID)
}

For detailed information about MCP audit logging, see Audit Logging.

Examples​

See the MCP policy examples in the AxonFlow repository:

All examples include implementations in Go, TypeScript, Python, and Java.

Last updated on