> ## Documentation Index > Fetch the complete documentation index at: https://qwed-ai-mintlify-2f129d13.mintlify.site/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. *** ## 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: * 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`. `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`. ```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. **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. ```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`). **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. ```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 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. ```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. 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. | ```python theme={null} 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.