QWED A2A verification interceptor pipeline and verdicts

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

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

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

Any agent IDs in config.trusted_agents are pre-added to the trust boundary allowlist at interceptor construction time. There is no separate bypass step — trusted agents still flow through engine routing and receive a normal verdict.

Engine Routing

Based on payload_type, the message is routed to the appropriate verification engine (finance_guard, logic_guard, code_guard). Payload types with no engine (GENERAL, DATA_QUERY) return an unverifiable engine result instead of being silently forwarded.

Verdict & Attestation

The engine result is wrapped in a VerificationVerdict. FORWARDED, BLOCKED, and HEURISTIC_PASS verdicts are signed with an ES256 JWT attestation. UNVERIFIABLE verdicts carry no JWT — issuing a signed token for content that was never verified would be a false cryptographic claim.

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.

Empty or malformed payloads return UNVERIFIABLE, not FORWARDED. If a FINANCIAL_TRANSACTION payload is missing data, line_items, or claimed_total, or contains non-numeric amounts, the finance guard returns status=unverifiable with no JWT attestation. Earlier releases collapsed these cases to a signed FORWARDED verdict, which falsely attested that the math had been checked.

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).

Empty or malformed logic payloads return UNVERIFIABLE. A LOGIC_ASSERTION payload with a missing, non-list, or empty assertions array — or with malformed assertion entries — returns status=unverifiable and no JWT attestation. There is nothing to check, so nothing is attested.

Code guard

Scans code payloads using a two-layer approach. AST structural analysis runs first; regex heuristics only run when the AST layer finds nothing. Layer 1 — AST structural analysis (primary). The code is parsed into an abstract syntax tree and inspected for direct dangerous constructs:

Threat	Example
Dangerous calls	`eval(`, `exec(`, `compile(`, `__import__(`
Dangerous receiver methods	`subprocess.run(`, `os.system(`, `os.popen(`
Dangerous imports	`import subprocess`, `import importlib`, `import ctypes`

Layer 2 — Regex heuristic scan (secondary). Catches obfuscation patterns that survive AST parsing:

Pattern	Catches
`getattr_builtin`	`getattr(__builtins__, ...)`
`builtins_dict_access`	`__builtins__.__dict__[`
`base64_exec`	`b64decode(` encoded payloads
`dynamic_import`	`__import__(` dynamic imports
`os_system`	`os.system(` shell execution
`os_popen`	`os.popen(` process spawning

If either layer finds a threat, the verdict is BLOCKED (the reason notes which layer triggered). If both layers are clean, the verdict is HEURISTIC_PASS — a signed attestation that no known dangerous constructs were found, not a deterministic guarantee that the code is safe to execute.

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 (AST): subprocess import"

Passthrough (unverifiable)

Messages with payload_type of GENERAL or DATA_QUERY have no verification engine. The interceptor returns an UNVERIFIABLE verdict with engine="passthrough", no JWT attestation, and reason "No verification engine available for this payload type". The message is not silently forwarded — callers must decide whether to route unverified content downstream based on their own policy.

verdict = await interceptor.intercept(chat_message, trace_id="chat_001")
# verdict.status = "unverifiable"
# verdict.engine_used = "passthrough"
# verdict.attestation_jwt is None

Verdict statuses

Every call to intercept() returns a VerificationVerdict with one of five statuses. Only statuses backed by a real verification decision carry a signed JWT attestation.

Status	Meaning	JWT attestation
`forwarded`	Engine verified the payload. Forward the message.	Signed
`blocked`	Engine detected a violation (bad math, contradiction, dangerous code). Do not forward.	Signed
`heuristic_pass`	Code guard ran, no known dangerous constructs found. Not a proof of safety.	Signed
`unverifiable`	No engine could evaluate the payload (`GENERAL`/`DATA_QUERY`, empty/malformed finance or logic payload). No attestation is issued.	None
`error`	Internal exception during verification. Behavior depends on `block_on_error`.	Signed

UNVERIFIABLE verdicts have attestation_jwt=None. If your downstream code assumes every verdict has a JWT, guard for this — issuing an attestation for content that was never verified would be a false cryptographic claim.

Configuration reference

The InterceptorConfig controls which engines are active:

Field	Type	Default	Description
`enable_financial_verification`	`bool`	`True`	Route financial payloads to math verification
`enable_logic_verification`	`bool`	`True`	Route logic assertions to contradiction checks
`enable_code_verification`	`bool`	`True`	Route code payloads to the AST + regex heuristic security scanner
`block_on_error`	`bool`	`True`	Block forwarding if verification encounters an internal error
`max_payload_size_bytes`	`int`	`1,048,576`	Maximum payload size (1 KB – 10 MB)
`trusted_agents`	`List[str]?`	`None`	Agent IDs pre-added to the trust boundary allowlist at startup. These agents still run through the appropriate verification engine — they are not bypassed.

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.

block_on_error=True (default)

Engine exception → BLOCKED verdict with error reason. Safe default for production.

block_on_error=False (observability mode)

Engine exception → FORWARDED verdict. The error is logged but doesn’t block communication. Use for shadow deployments.

Overview

Core Components

Guides

QWED A2A verification interceptor pipeline and verdicts

Overview

Verification pipeline

Verification engines

Finance guard

Logic guard

Code guard

Passthrough (unverifiable)

Verdict statuses

Configuration reference

Error handling

​Overview

​Verification pipeline

​Verification engines

​Finance guard

​Logic guard

​Code guard

​Passthrough (unverifiable)

​Verdict statuses

​Configuration reference

​Error handling

Overview

Verification pipeline

Verification engines

Finance guard

Logic guard

Code guard

Passthrough (unverifiable)

Verdict statuses

Configuration reference

Error handling