QWED A2A quick start for agent-to-agent verification

Installation

pip install qwed-a2a

git clone https://github.com/QWED-AI/qwed-a2a.git
cd qwed-a2a
pip install -e ".[dev]"

QWED A2A requires Python 3.10+. The cryptography and PyJWT packages are installed automatically for JWT attestation support.

Configure a signing key

QWED A2A signs every verdict with a persistent ECDSA P-256 key so attestations remain verifiable across restarts. Set QWED_A2A_SIGNING_KEY_PEM before importing qwed_a2a — without it the interceptor fails closed on the first call.

openssl ecparam -name prime256v1 -genkey -noout \
  | openssl pkcs8 -topk8 -nocrypt \
  > qwed_a2a_signing_key.pem

export QWED_A2A_SIGNING_KEY_PEM="$(cat qwed_a2a_signing_key.pem)"
export QWED_A2A_DEPLOYMENT_ID="local-dev"

See Crypto attestations for key rotation and multi-replica guidance.

Your first verification

Import the interceptor

import asyncio
from qwed_a2a.interceptor import A2AVerificationInterceptor
from qwed_a2a.protocol.schema import AgentMessage, PayloadType

Create an interceptor

interceptor = A2AVerificationInterceptor()

This creates an interceptor with:

All verification engines enabled
Crypto attestation enabled (if packages available)
default_allow=True trust boundary

Build a test message

message = AgentMessage(
    sender_agent_id="demo-sender",
    receiver_agent_id="demo-receiver",
    payload_type=PayloadType.FINANCIAL_TRANSACTION,
    payload={
        "data": {
            "claimed_total": 150.00,
            "line_items": [
                {"description": "Widget A", "amount": 50.00, "quantity": 2},
                {"description": "Widget B", "amount": 25.00, "quantity": 2},
            ]
        }
    }
)

Run verification

async def main():
    verdict = await interceptor.intercept(
        message,
        trace_id="quickstart_001"
    )

    print(f"Status:      {verdict.status.value}")
    print(f"Engine:      {verdict.engine_used}")
    print(f"Trace ID:    {verdict.audit_trace_id}")
    print(f"Attestation: {verdict.attestation_jwt[:50]}...")

asyncio.run(main())

Check the output

Status:      forwarded ✅
Engine:      finance_guard
Trace ID:    quickstart_001
Attestation: eyJhbGciOiJFUzI1NiIsInR5cCI6InF3ZWQtYTJhLWF0...

The financial totals match (50×2 + 25×2 = 150), so the message is forwarded with a signed attestation.

Try different scenarios

Block a hallucinated total

bad_message = AgentMessage(
    sender_agent_id="demo-sender",
    receiver_agent_id="demo-receiver",
    payload_type=PayloadType.FINANCIAL_TRANSACTION,
    payload={
        "data": {
            "claimed_total": 999.99,  # Wrong!
            "line_items": [
                {"amount": 100.00, "quantity": 1},
                {"amount": 50.00, "quantity": 1},
            ]
        }
    }
)

verdict = await interceptor.intercept(bad_message, trace_id="bad_001")
print(verdict.status)  # blocked
print(verdict.reason)  # "Mathematical hallucination detected..."

Block dangerous code

code_message = AgentMessage(
    sender_agent_id="code-agent",
    receiver_agent_id="executor",
    payload_type=PayloadType.CODE_EXECUTION,
    payload={"code": "import os; os.system('rm -rf /')"}
)

verdict = await interceptor.intercept(code_message, trace_id="code_001")
print(verdict.status)  # blocked
print(verdict.reason)  # "Dangerous code patterns detected: os.system"

Catch logical contradictions

logic_message = AgentMessage(
    sender_agent_id="reasoning-agent",
    receiver_agent_id="planner",
    payload_type=PayloadType.LOGIC_ASSERTION,
    payload={
        "assertions": [
            {"claim": "budget_approved", "negated": False},
            {"claim": "budget_approved", "negated": True},
        ]
    }
)

verdict = await interceptor.intercept(logic_message, trace_id="logic_001")
print(verdict.status)  # blocked
print(verdict.reason)  # "Logical contradiction detected..."

Run the FastAPI gateway

QWED A2A includes a ready-to-use HTTP gateway:

from fastapi import FastAPI
from qwed_a2a.protocol.endpoints import router

app = FastAPI(title="QWED A2A Gateway")
app.include_router(router)

# Run with: uvicorn main:app --host 0.0.0.0 --port 8000

Test it:

curl -X POST http://localhost:8000/a2a/intercept \
  -H "Content-Type: application/json" \
  -d '{
    "sender_agent_id": "agent-A",
    "receiver_agent_id": "agent-B",
    "payload_type": "general",
    "payload": {"message": "Hello!"}
  }'

Response:

{
  "status": "unverifiable",
  "audit_trace_id": "a2a_7f3c2a1b9e4d",
  "engine_used": "passthrough",
  "attestation_jwt": null,
  "reason": "No verification engine available for this payload type"
}

GENERAL and DATA_QUERY payloads have no verification engine, so the interceptor returns an unverifiable verdict with no JWT attestation — signing a token for content that was never verified would be a false cryptographic claim. Callers must decide how to handle unverified traffic. To see a signed forwarded verdict, send a financial_transaction, logic_assertion, or code_execution payload instead. See the verdict status table for the full contract.

Next steps

Architecture deep-dive — understand the full pipeline
Trust boundary — configure zero-trust policies
Crypto attestations — understand JWT signing
Production deployment — Docker, monitoring, CI/CD

Overview

Core Components

Guides

QWED A2A quick start for agent-to-agent verification

Installation

Configure a signing key

Your first verification

Try different scenarios

Run the FastAPI gateway

Next steps

​Installation

​Configure a signing key

​Your first verification

​Try different scenarios

​Run the FastAPI gateway

​Next steps

Installation

Configure a signing key

Your first verification

Try different scenarios

Run the FastAPI gateway

Next steps