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.

Sender identity is bound by the caller, not the message. Over HTTP, the gateway authenticates each request with a per-agent API key and overrides message.sender_agent_id with the authenticated agent before calling intercept(). Direct library callers get no such override: you own identity binding and must never pass an untrusted sender_agent_id to intercept().

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. Over HTTP the sender is the authenticated API-key identity, never the body-declared sender_agent_id:

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. VERIFIED, BLOCKED, HEURISTIC_PASS and ERROR verdicts are signed with an ES256 JWT attestation whose verdict claim names the status. 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.

When the recomputed total matches claimed_total, the finance guard returns verified — the only admissible verdict in QWED A2A. See Admissibility.

Negative values return UNVERIFIABLE. A negative claimed_total, line-item amount, or line-item quantity returns status=unverifiable with no JWT attestation. The payload has no credit/debit convention, so the finance guard cannot verify a sign-inverted transaction even when its arithmetic holds.

Empty or malformed payloads return UNVERIFIABLE, not VERIFIED. 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 pass 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).

The logic guard normalizes each claim before comparing:

Unicode NFKC normalization
Removal of invisible format characters, such as zero-width spaces
casefold()
Collapsing of repeated whitespace

As a result, "UserIsAdmin" and "userisadmin" are treated as the same claim, and so are "payment sent" and "payment sent". The negated field must be a boolean. Any other value, such as the string "true", returns unverifiable.

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__(` (also via `builtins.` and `__builtins__.`)
Dangerous receiver methods	`subprocess.run(`; `os.system(`, `os.popen(`, `os.exec(`, `os.spawn(`, `os.posix_spawn(`, `os.startfile(`. The same methods on `posix` and `nt` are blocked too.
Dangerous imports	`import subprocess`, `import importlib`, `import ctypes`; `from os import *`, `from os import system`
`getattr` on a dangerous module	`getattr(os, "system")`, or any runtime-chosen name such as `getattr(os, name)`

The AST layer resolves receivers through every binding a name may have. Each of these is blocked:

Import aliases: import os as o; o.system(...)
Alias chains of any length and order: x = os; y = x; y.system(...). A later re-assignment does not hide an earlier binding.
Re-bound functions: s = os.system; s(...), e = eval; e(...), g = getattr; g(os, "system")
Aliased from-imports: from posix import system as s; s(...)
Annotated (x: T = os), walrus ((x := os)), and tuple-unpacking (a, b = os, 1) bindings

A local re-binding of eval cannot hide a builtin eval(...) call. Before parsing, the code guard checks the input:

Code larger than 256 KiB is BLOCKED before ast.parse runs, because parsing amplifies memory roughly 140-250x.
A code value that is not a string returns UNVERIFIABLE with no attestation.

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

Pattern	Catches
`getattr_builtin`	`getattr(__builtins__, ...)`
`builtins_dict_access`	`__builtins__.__dict__[`, `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. A HEURISTIC_PASS is not admissible as proof of safety.

Known gaps. A blocklist over arbitrary code cannot be made sound. The code guard does not detect these today:

Subscript callees such as [eval][0](...) or d["e"](...)
Modules reached through objects, such as sys.modules["os"]
Bindings made by for, with, function arguments, or attributes of other objects
Unlisted modules with side effects, such as shutil, pickle, runpy, code, and socket

Never execute a heuristic_pass payload without your own sandbox or allowlist.

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 constructs detected via AST analysis: import:subprocess"

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. Every status except unverifiable carries a signed JWT attestation whose verdict claim names that status, so a signature on its own does not mean the payload passed. Only verified is admissible: gate on verdict.status.admissible (or ADMISSIBLE_VERDICTS), never on status != blocked, which would admit heuristic_pass, unverifiable and error. error is returned when verification cannot complete — for example when an engine run exceeds its execution budget (engine engine_timeout) or the engine pool is saturated (engine engine_busy).

Status	Meaning	JWT attestation
`verified`	A deterministic engine (finance guard) recomputed the claims and they checked out. The only admissible status.	Signed
`blocked`	Engine detected a violation (bad math, contradiction, dangerous code), or the trust boundary refused the message. Do not forward.	Signed
`heuristic_pass`	Code guard’s AST and regex scan, or logic guard’s contradiction match, found no problem. Not a proof, and not admissible.	Signed
`unverifiable`	No engine could evaluate the payload (`GENERAL`/`DATA_QUERY`, empty/malformed finance, logic or code payload). No attestation is issued. Not admissible.	None
`error`	The engine raised, timed out or was saturated before verification completed. Fail closed; not admissible. See Error handling.	Signed

VerdictStatus.FORWARDED is a deprecated alias of VerdictStatus.VERIFIED (value "verified"), and a legacy "forwarded" value still parses to VERIFIED on input. Neither is ever emitted; both will be removed in a future minor release. If your code compares against "forwarded", switch it to "verified" or to verdict.status.admissible.

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.

Admissibility

Only verified is admissible — a verdict a consumer may act on without running its own checks. QWED A2A defines admissibility in one place:

ADMISSIBLE_VERDICTS: a frozenset of admissible VerdictStatus members. It currently contains only VerdictStatus.VERIFIED.
VerdictStatus.admissible: True when the status is in ADMISSIBLE_VERDICTS.
VerificationVerdict.admissible: a computed field on every verdict, derived from status. It is serialized with the verdict, so the POST /a2a/intercept JSON response carries it too.

Gate on the verdict’s admissible flag instead of comparing status strings:

from qwed_a2a import ADMISSIBLE_VERDICTS, VerdictStatus

verdict = await interceptor.intercept(message, trace_id="a2a_trace_001")
if verdict.admissible:
    process(message)
else:
    reject_or_escalate(verdict.status.value, verdict.reason)

assert VerdictStatus.VERIFIED in ADMISSIBLE_VERDICTS
assert not VerdictStatus.HEURISTIC_PASS.admissible

To check admissibility from a signed attestation instead of a verdict object, use A2ACryptoService.is_admissible(claims).

Migrating from `forwarded` to `verified`

Breaking change. The forwarded verdict is renamed to verified, and only finance passes are verified. A logic guard pass now returns heuristic_pass, which is not admissible.

For compatibility:

VerdictStatus.FORWARDED remains a deprecated alias of VerdictStatus.VERIFIED.
The legacy wire value "forwarded" still parses to verified in HTTP verdicts and attestation claims, so tokens from 0.3.x issuers keep verifying on upgraded readers.
/a2a/metrics reports total_verified and keeps total_forwarded as a deprecated alias with the same count.

QWED A2A will remove the alias and the legacy value in a future minor release. Before:

if verdict.status.value == "forwarded":
    process(message)

After:

if verdict.admissible:
    process(message)

If you must compare strings, replace "forwarded" with "verified". If you relied on logic assertions being admitted, they now return heuristic_pass. Add your own policy for them.

Upgrade consumers before, or together with, issuers. A 0.3.x consumer that compares the attestation verdict claim to "forwarded" denies tokens carrying "verified" until you update it. This fails closed, but blocks legitimate traffic.

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`	Deprecated, no effect. Engine errors always fail closed with an `error` verdict; setting it to `False` logs a deprecation warning. Kept for config compatibility.
`max_payload_size_bytes`	`int`	`1,048,576`	Maximum payload size (1 MiB; allowed range 1 KiB to 10 MiB). Enforced on the HTTP body (`413`) and in `intercept()` (`blocked`). See Request size limits.
`engine_timeout_seconds`	`float`	`5.0`	Wall-clock budget per engine run. Must be greater than 0 and at most 60. Exceeding it returns `error` (engine `engine_timeout`). See Engine execution budget.
`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,
    max_payload_size_bytes=512 * 1024,  # 512 KiB
    engine_timeout_seconds=2.0,
    trusted_agents=["internal-orchestrator-001"],
)

Request size limits

The interceptor enforces max_payload_size_bytes at two layers:

HTTP body. POST /a2a/intercept rejects bodies larger than max_payload_size_bytes plus a 64 KiB envelope allowance with 413, before the body is buffered or parsed. The router checks a declared Content-Length up front and caps chunked bodies as they stream in. An invalid Content-Length returns 400. The check is built into the router, so it applies wherever you mount router with no extra middleware.
Payload. intercept() returns blocked with engine size_limit when the payload’s compact UTF-8 JSON exceeds max_payload_size_bytes. This check runs before trust evaluation or any engine, so it also covers direct library callers.

The code guard applies a separate 256 KiB cap on code. See Code guard.

verdict = await interceptor.intercept(oversized_message, trace_id="big_001")
# verdict.status = "blocked"
# verdict.engine_used = "size_limit"
# verdict.reason = "Payload is 2097152 bytes; max_payload_size_bytes=1048576"

Keep a reverse-proxy body limit as well, such as nginx client_max_body_size, as an outer layer of defense.

Engine execution budget

Verification engines run on a dedicated thread pool, off the event loop. /a2a/health, /a2a/metrics, JWKS, and other intercepts keep responding while an engine works.

Each engine run has a wall-clock budget set by engine_timeout_seconds. A run that exceeds it returns error with engine engine_timeout.
Each interceptor allows at most 36 runs in flight: 4 running and 32 waiting for a thread. Beyond that, a run is refused immediately with error and engine engine_busy instead of queueing without bound.
A waiting run whose caller already timed out is skipped.

Call close() from your application’s shutdown hook to release the pool. Call it only after in-flight intercepts have drained. An intercept that starts after close() returns error.

@app.on_event("shutdown")
async def shutdown():
    interceptor.close()

Limits. Python cannot kill a running thread. A timed-out run that already started finishes in the background and holds its thread until it completes. Under sustained overload, runs time out or are refused, both fail-closed. Long C-level calls that hold the GIL, such as ast.parse or regex, can still delay the event loop. The size limits above bound those calls. Each worker process has its own pool and state, so multiple workers add capacity but not isolation.

Error handling

Verification that does not complete always fails closed with an ERROR verdict. An engine error is neither a pass (VERIFIED) nor a content rejection (BLOCKED), and no configuration turns it into a pass. The verdict is signed and its verdict claim is "error".

Cause	`engine_used`	Reason
Engine raised an exception	`error_handler`	`Verification engine raised <ExceptionType> - fail closed` (only the exception type is surfaced, since its message may echo payload data)
Engine exceeded `engine_timeout_seconds` (default 5 s)	`engine_timeout`	`Verification engine timed out after <n>s - fail closed`
All engine slots busy	`engine_busy`	`Verification engines are saturated - fail closed`

block_on_error is deprecated and ignored: earlier releases let block_on_error=False pass a message through despite an engine error. Setting it to False now only logs a deprecation warning.

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

Admissibility

Migrating from `forwarded` to `verified`

Configuration reference

Request size limits

Engine execution budget

Error handling

​Overview

​Verification pipeline

​Verification engines

​Finance guard

​Logic guard

​Code guard

​Passthrough (unverifiable)

​Verdict statuses

​Admissibility

​Migrating from forwarded to verified

​Configuration reference

​Request size limits

​Engine execution budget

​Error handling

Overview

Verification pipeline

Verification engines

Finance guard

Logic guard

Code guard

Passthrough (unverifiable)

Verdict statuses

Admissibility

Migrating from `forwarded` to `verified`

Configuration reference

Request size limits

Engine execution budget

Error handling