Skip to main content

​
Overview

The A2AVerificationInterceptor is the central component of QWED A2A. Every inter-agent message passes through it before reaching the recipient. 
from qwed_a2a.interceptor import A2AVerificationInterceptor

interceptor = A2AVerificationInterceptor(
    config=config,              # InterceptorConfig
    crypto_service=crypto,      # A2ACryptoService (optional)
    trust_boundary=boundary,    # TrustBoundary (optional)
)

verdict = await interceptor.intercept(message, trace_id="a2a_trace_001")
The trace_id parameter is required and must be provided by the caller. This ensures all verdicts and JWT attestations are deterministic and auditable. The HTTP gateway (POST /a2a/intercept) generates this automatically.

​
Verification pipeline

1

Schema Validation

The incoming AgentMessage is validated by Pydantic:
  • sender_agent_id and receiver_agent_id must be 1–256 chars, no control characters
  • payload is required (dict)
  • payload_type defaults to GENERAL if not specified
  • Timestamps are timezone-aware UTC
2

Trust Boundary

The trust boundary evaluates the sender→receiver pair:
  • Global blocklist check
  • Pair-level block check
  • Allowlist check (in strict mode)
  • Token-bucket rate limiting
3

Trusted Agent Bypass

If the sender is in config.trusted_agents, the message is forwarded immediately with a bypass engine verdict.
4

Engine Routing

Based on payload_type, the message is routed to the appropriate verification engine.
5

Verdict & Attestation

The engine result is wrapped in a VerificationVerdict with an ES256 JWT attestation proving the verification took place.

​
Verification engines

​
Finance Guard

Verifies financial claims using deterministic Decimal arithmetic. Recomputes totals from line items and compares against the claimed_total. 
message = AgentMessage(
    sender_agent_id="sales-agent",
    receiver_agent_id="treasury-agent",
    payload_type=PayloadType.FINANCIAL_TRANSACTION,
    payload={
        "data": {
            "claimed_total": 999.99,  # Wrong!
            "line_items": [
                {"description": "Product X", "amount": 100.00, "quantity": 1},
                {"description": "Product Y", "amount": 50.00, "quantity": 1},
            ]
        }
    }
)

verdict = await interceptor.intercept(message, trace_id="fin_001")
# verdict.status = "blocked"
# verdict.reason = "Mathematical hallucination detected:
#     claimed_total=999.99, computed_total=150.00"
All financial comparisons use decimal.Decimal with ROUND_HALF_UP quantization to 0.01. Floating-point arithmetic is never used in the verification path.

​
Logic Guard

Detects logical contradictions — claims where the same proposition is both asserted and negated. 
message = AgentMessage(
    sender_agent_id="reasoning-agent",
    receiver_agent_id="planner-agent",
    payload_type=PayloadType.LOGIC_ASSERTION,
    payload={
        "assertions": [
            {"claim": "sky_is_blue", "negated": False},
            {"claim": "sky_is_blue", "negated": True},  # Contradiction!
        ]
    }
)

verdict = await interceptor.intercept(message, trace_id="logic_001")
# verdict.status = "blocked"
# verdict.reason = "Logical contradiction detected:
#     claims both asserted and negated: ['sky_is_blue']"
Contradictions are sorted before output, ensuring deterministic results regardless of Python’s hash randomization (PYTHONHASHSEED).

​
Code Guard

Scans code payloads for dangerous patterns using case-insensitive compiled regex:
PatternRegexCatches
eval\beval\s*\(eval(, EVAL (, eval\n(
exec\bexec\s*\(exec(, Exec (
subprocesssubprocess\.|import subprocess|from subprocess importsubprocess.run, import subprocess as sp, from subprocess import run
os.system\bos\.system\s*\(os.system(, OS.SYSTEM (
os.popen\bos\.popen\s*\(All casing/spacing variants
__import____import__\s*\(Dynamic imports
compile\bcompile\s*\(Code compilation
importlib\bimportlib\s*\.Runtime module loading
message = AgentMessage(
    sender_agent_id="code-agent",
    receiver_agent_id="executor-agent",
    payload_type=PayloadType.CODE_EXECUTION,
    payload={"code": "import subprocess as sp\nsp.run(['ls'])"}
)

verdict = await interceptor.intercept(message, trace_id="code_001")
# verdict.status = "blocked"
# verdict.reason = "Dangerous code patterns detected: subprocess"

​
Passthrough

Messages with payload_type of GENERAL or DATA_QUERY are forwarded without verification.

​
Configuration reference

The InterceptorConfig controls which engines are active:
FieldTypeDefaultDescription
enable_financial_verificationboolTrueRoute financial payloads to math verification
enable_logic_verificationboolTrueRoute logic assertions to contradiction checks
enable_code_verificationboolTrueRoute code payloads to regex security scanning
block_on_errorboolTrueBlock forwarding if verification encounters an internal error
max_payload_size_bytesint1,048,576Maximum payload size (1 KB – 10 MB)
trusted_agentsList[str]?NoneAllowlist of agent IDs that bypass verification
from qwed_a2a.protocol.schema import InterceptorConfig

config = InterceptorConfig(
    enable_financial_verification=True,
    enable_code_verification=True,
    block_on_error=True,
    trusted_agents=["internal-orchestrator-001"],
)

​
Error handling

When block_on_error=True (default), any exception in a verification engine results in a BLOCKED verdict. When False, the message is FORWARDED despite the error — useful for observability-only deployments.
Engine exception → BLOCKED verdict with error reason. Safe default for production.
Engine exception → FORWARDED verdict. The error is logged but doesn’t block communication. Use for shadow deployments.