> ## Documentation Index > Fetch the complete documentation index at: https://docs.qwedai.tech/llms.txt > Use this file to discover all available pages before exploring further. # QWED A2A verification interceptor pipeline and verdicts > How the QWED A2A verification interceptor pipeline works: schema validation, engine routing, verdict generation, and payload_type enforcement. ## Overview The `A2AVerificationInterceptor` is the central component of QWED A2A. Every inter-agent message passes through it before reaching the recipient. ```python theme={null} 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](/a2a/deployment#gateway-authentication) 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 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 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. 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. 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`. ```python theme={null} 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](#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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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-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](#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: ```python theme={null} 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)`](/a2a/crypto-attestations#check-admissibility). ### 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: ```python theme={null} if verdict.status.value == "forwarded": process(message) ``` After: ```python theme={null} 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](#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](#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. | ```python theme={null} 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](#code-guard). ```python theme={null} 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`. ```python theme={null} @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 - 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 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. ## Related topics - [QWED A2A architecture for agent verification](/a2a/architecture.md) - [QWED A2A quick start for agent-to-agent verification](/a2a/quickstart.md) - [ES256 JWT crypto attestations for QWED A2A verdicts](/a2a/crypto-attestations.md) - [QWED A2A zero-trust protocol for agent-to-agent verification](/a2a/overview.md) - [Changelog](/changelog.md) This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.