nexus-a2a

Production-grade Agent-to-Agent communication for Python. One decorator. Zero boilerplate. Any AI framework.

❤️ Support

What is nexus-a2a?

nexus-a2a is a Python library that lets AI agents talk to each other over HTTP using the A2A protocol. It handles discovery, routing, authentication, retries, circuit breaking, tracing, and failure recovery — so you focus on what your agent actually does.

Agent A  ──HTTP/JSON-RPC──▶  Agent B  ──▶  Agent C
   │                             │
   └── auto-retry           DLQ on fail
   └── circuit breaker      trace ID
   └── mTLS                 rate limit

Installation
Quickstart
Core Concepts
Building Agents
Serving Agents
Sending Tasks
Agent Registry & Discovery
Orchestration
Security
Storage Backends
Reliability
Streaming & Webhooks
Framework Adapters
Observability & CLI
Configuration (nexus.toml)
Error Handling Reference
Testing

Installation

# Core
pip install nexus-a2a

# With Redis task store
pip install "nexus-a2a[redis]"

# With PostgreSQL task store
pip install "nexus-a2a[postgres]"

# Asymmetric JWT (RS256/ES256) and JWKS
pip install "nexus-a2a[jwks]"

# With the Google ADK adapter (heavy — ~40 transitive packages)
pip install "nexus-a2a[adk]"

# Everything
pip install "nexus-a2a[all]"

# For contributors / testing
pip install "nexus-a2a[dev]"

Requires Python 3.11+.

The core install is deliberately small — 20 packages, no known CVEs. LangGraph, CrewAI, AutoGen and Google ADK are all imported lazily, so install only the adapter you actually use. (google-adk was a required dependency before v1.5.0; it is now the adk extra.)

Quickstart

1. Define your agent

from nexus_a2a import agent
from nexus_a2a import Task

@agent(
    name="SummaryAgent",
    description="Summarises text passed to it.",
    skills=[{
        "id": "summarise",
        "name": "Summarise",
        "description": "Returns a one-paragraph summary.",
        "tags": ["nlp", "text"],
    }],
    url="http://localhost:8001",
)
class SummaryAgent:
    async def run(self, task: Task) -> str:
        text = task.latest_message().text()
        return f"Summary: {text[:100]}..."

2. Start the server

nexus run --module mypackage.agent:SummaryAgent

That serves the agent over the A2A protocol:

Serving agent 'SummaryAgent' on http://localhost:8001
  Agent card: http://localhost:8001/.well-known/agent-card.json
  JSON-RPC:   POST http://localhost:8001/
  Health:     http://localhost:8001/health
  Skills:     summarise

Or start it from Python, which is what nexus run does under the hood:

from nexus_a2a import A2AServer

server = A2AServer(SummaryAgent, port=8001)
await server.start()
# ... or: async with A2AServer(SummaryAgent, port=8001):

3. Send a task from another agent

import asyncio
from nexus_a2a import Message
from nexus_a2a import A2AHttpClient

async def main():
    async with A2AHttpClient("http://localhost:8001") as client:
        task = await client.send_message(Message.user_text("Summarise this long document..."))
        print(task.state)            # TaskState.COMPLETED
        print(task.artifacts[0].parts[0].content)

asyncio.run(main())

Core Concepts

Task state machine

Every task follows a strict state machine. Invalid transitions raise ValueError.

SUBMITTED ──▶ WORKING ──▶ COMPLETED
                │ ──────▶ FAILED        (requires error= message)
                │ ──────▶ INPUT_REQUIRED ──▶ WORKING
                └───────▶ CANCELLED

from nexus_a2a import Task, TaskState, Message

task = Task.create(initial_message=Message.user_text("hello"))
# task.state == TaskState.SUBMITTED

task.transition(TaskState.WORKING)
task.transition(TaskState.COMPLETED)

# ❌ This raises — can't skip WORKING
task2 = Task.create(initial_message=Message.user_text("x"))
task2.transition(TaskState.FAILED)               # ValueError!
task2.transition(TaskState.WORKING)              # ✓
task2.transition(TaskState.FAILED, error="oops") # ✓

Message and Part

from nexus_a2a import Message, Part, PartType, MessageRole

# Shortcut (most common)
msg = Message.user_text("What is the weather today?")

# Full form
msg = Message(
    role=MessageRole.USER,
    parts=[
        Part(type=PartType.TEXT, content="What is the weather today?"),
    ],
)

# Read text back
print(msg.text())   # "What is the weather today?"

Building Agents

`@agent` decorator

The primary way to define an agent. Auto-generates an AgentCard.

from nexus_a2a import agent
from nexus_a2a import Task

@agent(
    name="ResearchAgent",
    description="Searches the web and returns summarised results.",
    version="1.0.0",
    url="http://localhost:8001",
    skills=[
        {
            "id": "web_search",
            "name": "Web Search",
            "description": "Searches the web for a given query.",
            "tags": ["search", "web"],
            "examples": ["Find latest AI papers", "Search Python asyncio docs"],
        }
    ],
    streaming=False,
    push_notifications=False,
)
class ResearchAgent:
    async def run(self, task: Task) -> str:
        query = task.latest_message().text()
        # ... call search API ...
        return f"Results for: {query}"

Use @agent without arguments — class name and docstring become defaults:

@agent
class QuickAgent:
    """A fast agent that does quick tasks."""
    async def run(self, task: Task) -> str:
        return "done"

Access the generated card:

card = ResearchAgent.get_agent_card()
print(card.name)           # "ResearchAgent"
print(card.skills[0].id)   # "web_search"

# Or via helper
from nexus_a2a import get_card
card = get_card(ResearchAgent)

Serving Agents

A2AServer is the inbound half of the protocol — it turns an @agent class into a reachable agent.

Endpoint	Purpose
`GET /.well-known/agent-card.json`	Discovery — the card `@agent` built
`POST /`	JSON-RPC 2.0: `message/send`, `message/stream`, `tasks/get`, `tasks/cancel`, `tasks/pushNotificationConfig/set` and `/get` (send/stream take an optional `taskId` to continue a paused task)
`GET /stream?taskId=`	Observe a task as Server-Sent Events
`GET /health`	Liveness probe
`GET /ready`	Readiness probe (checks the task store)

from nexus_a2a import A2AServer

server = A2AServer(SummaryAgent, port=8001)
await server.start()
...
await server.stop()

# Or as an async context manager
async with A2AServer(SummaryAgent, port=8001):
    await asyncio.Event().wait()

Host and port default to whatever the agent card's url says, so a single url="http://localhost:8001" on the decorator configures both sides.

What `run()` can return

Return value	Result
`str`	One text `Artifact`
`dict` / `list`	One JSON `Artifact`
`Artifact`	Used as-is
`Message`	Appended to task history as the agent's reply
`AdapterResult`	`.to_artifact()`, or task `FAILED` if `.error` is set
`None`	Completed with no artifact

A run() that raises is not a protocol error: the task is recorded as FAILED with the exception message and returned normally, so the caller can inspect task.error and the DLQ can capture it.

Multi-turn conversations

An agent asks a follow-up by returning NeedsInput instead of a result. The task parks at INPUT_REQUIRED with the question as its last message, and the caller answers against the same task id:

from nexus_a2a import agent, NeedsInput, Task

@agent(name="Planner", description="Plans a trip.", url="http://localhost:8001")
class Planner:
    async def run(self, task: Task):
        turns = sum(1 for m in task.history if m.role == "user")
        if turns == 1:
            return NeedsInput("What is your budget?")
        return f"Plan for {task.history[-1].text()}"

async with A2AHttpClient("http://localhost:8001") as client:
    task = await client.send_message(Message.user_text("plan a trip"))
    print(task.state)               # TaskState.INPUT_REQUIRED
    print(task.history[-1].text())  # "What is your budget?"

    task = await client.send_message(Message.user_text("$500"), task_id=task.id)
    print(task.state)               # TaskState.COMPLETED
    print(task.artifacts[0].parts[0].content)   # "Plan for $500"

run() is re-invoked with the full history each turn, so the agent reads earlier answers straight off task.history. There is no limit on turns.

This is stateless. The conversation lives in the task store, not in a parked coroutine — it survives a restart, and a different client or connection can continue it. Streaming works the same way: pass task_id= to stream_message(), and a pause arrives as a message event followed by task_status: input_required.

Situation	Result
`taskId` omitted	New task
Task is `INPUT_REQUIRED`	Resumes, history preserved
Task already finished	`-32002` naming the actual state
Unknown `taskId`	`-32001`
Agent declares `multi_turn=False`	`-32002`, continuations refused

InputHandler.wait_for_input() remains for in-process suspension, and the same wire call resumes it. Prefer NeedsInput over HTTP — wait_for_input() holds the original request open for its whole timeout.

Error model

Transport-level rejections that happen before dispatch return real HTTP status codes — 401 auth, 403 trust, 413 too large, 429 rate limit (with Retry-After), 400 malformed. Proxies and dashboards can see them, and the client will not retry a rejected credential.

Application-level failures after dispatch return JSON-RPC error objects — -32601 method not found, -32602 invalid params, -32001 task not found — which reach the caller as RemoteAgentError with the code intact.

Enforcing security

The security classes are building blocks; SecurityMiddleware is what makes A2AServer actually enforce them. Stages run cheapest-first: size → rate limit → auth → trust → payload validation. Every stage is optional, so an agent starts open and hardens incrementally.

from nexus_a2a import (
    A2AServer, SecurityMiddleware, AuthManager, AgentCredentialConfig,
    AuthScheme, TrustBoundary, RateLimiter, PayloadValidator,
)

MY_URL = "http://summary-agent:8001"

auth = AuthManager()
auth.register_agent("http://orchestrator:8000", AgentCredentialConfig(
    scheme=AuthScheme.API_KEY, api_key="shared-secret",
))

trust = TrustBoundary()
trust.allow("http://orchestrator:8000", MY_URL, skills=["summarise"])

server = A2AServer(SummaryAgent, security=SecurityMiddleware(
    auth=auth,
    trust=trust,
    rate_limiter=RateLimiter(),
    validator=PayloadValidator(),
    server_url=MY_URL,      # this agent is the trust TARGET
))

Callers identify themselves with caller_url, which auth and trust both need (a trust rule is caller -> target):

async with A2AHttpClient(
    "http://summary-agent:8001",
    caller_url="http://orchestrator:8000",     # sends X-Nexus-Caller
    headers={"X-API-Key": "shared-secret"},
) as client:
    task = await client.send_message(Message.user_text("..."), skill_id="summarise")

Without caller_url the call is anonymous, and a server with auth or trust enabled rejects it with 401.

Ops endpoints live on a separate port. /metrics, /info, /dlq and trace lookup are served by AgentServer — the standard app-port / admin-port split. Run both if you want protocol and ops on one host.

Sending Tasks

A2AHttpClient

from nexus_a2a import (
    A2AHttpClient,
    RetryConfig,
    CircuitBreaker,
)
from nexus_a2a import Message, TaskState

async with A2AHttpClient(
    "http://localhost:8001",
    timeout=30.0,
    retry=RetryConfig(
        max_retries=3,
        base_delay=0.5,
        max_delay=30.0,
        jitter=True,
        retry_on={500, 502, 503, 504},
    ),
) as client:

    # Send a task
    task = await client.send_message(Message.user_text("hello"))
    assert task.state == TaskState.COMPLETED

    # Fetch an existing task
    task = await client.get_task(task.id)

    # Cancel a task
    cancelled = await client.cancel_task(task.id)

    # Fetch the agent's card
    card = await client.fetch_agent_card()
    print(card.name, card.skills)

Circuit Breaker

Prevents hammering a failing agent. Automatically opens after N failures and lets a test request through after the recovery window.

from nexus_a2a import CircuitBreaker, CircuitOpenError

cb = CircuitBreaker(
    failure_threshold=5,    # open after 5 consecutive failures
    recovery_timeout=30.0,  # try again after 30 s
    success_threshold=2,    # close after 2 successes
)

async with A2AHttpClient("http://localhost:8001", circuit_breaker=cb) as client:
    try:
        task = await client.send_message(Message.user_text("ping"))
    except CircuitOpenError as e:
        print(f"Circuit is OPEN. Retry after {e.retry_after:.0f}s")

Agent Registry & Discovery

from nexus_a2a import AgentRegistry

registry = AgentRegistry(
    card_ttl_seconds=300.0,       # re-fetch card after 5 min
    health_check_timeout=5.0,
)

# Register by URL (fetches AgentCard automatically)
card = await registry.register_url("http://localhost:8001")

# Register multiple
for url in ["http://agent-a:8001", "http://agent-b:8002"]:
    await registry.register_url(url)

# List all registered agents
cards = registry.list_all()          # list[AgentCard]
healthy = registry.list_healthy()    # only healthy ones

# Find agents that have a specific skill
matches = registry.find_by_skill("web_search")   # list[AgentCard]

# Lookup by name or URL
card = registry.get_by_name("ResearchAgent")
card = registry.get_by_url("http://localhost:8001")

# Health checks
health_map = await registry.check_all_health()  # {"http://...": True/False}
is_up = await registry.check_health("http://localhost:8001")

# Summary
print(registry.summary())
# {"total": 3, "healthy": 2, "unhealthy": 1, "skills": [...]}

Orchestration

Sequential pipeline

Each agent's output becomes the next agent's input.

from nexus_a2a import Orchestrator
from nexus_a2a import A2AHttpClient
from nexus_a2a import Message, TaskState

async def runner(url: str, message: Message) -> Task:
    async with A2AHttpClient(url) as client:
        return await client.send_message(message)

orchestrator = Orchestrator(runner=runner, stop_on_error=True)

result = await orchestrator.sequential(
    agent_urls=["http://agent-a:8001", "http://agent-b:8002", "http://agent-c:8003"],
    initial_message=Message.user_text("Research quantum computing trends"),
)

print(result.succeeded)        # True / False
print(result.total_sec)        # wall-clock seconds
print(result.final_output)     # last Task produced

for step in result.steps:
    print(step.agent_url, step.duration_sec)
    if step.succeeded:
        print("  ✓", step.task.state)
    else:
        print("  ✗", step.error)

Parallel

All agents receive the same input simultaneously.

result = await orchestrator.parallel(
    agent_urls=["http://agent-a:8001", "http://agent-b:8002"],
    message=Message.user_text("Summarise this document"),
)

for step in result.steps:
    print(step.agent_url, "→", step.task.artifacts[0].parts[0].content)

DAG (Directed Acyclic Graph)

Dependency-aware execution. Agents with no pending dependencies run concurrently.

from nexus_a2a import DAGNode

nodes = [
    DAGNode(agent_url="http://fetcher:8001",    depends_on=[]),
    DAGNode(agent_url="http://parser:8002",     depends_on=["http://fetcher:8001"]),
    DAGNode(agent_url="http://summariser:8003", depends_on=["http://fetcher:8001"]),
    DAGNode(agent_url="http://reporter:8004",   depends_on=["http://parser:8002",
                                                             "http://summariser:8003"]),
]

result = await orchestrator.dag(
    nodes=nodes,
    initial_message=Message.user_text("Process this dataset"),
)

Security

Authentication

Three schemes supported. Each registered agent can use a different scheme.

from nexus_a2a import AuthManager, AgentCredentialConfig
from nexus_a2a import AuthScheme

auth = AuthManager()

# API Key
auth.register_agent("http://agent-a:8001", AgentCredentialConfig(
    scheme=AuthScheme.API_KEY,
    api_key="my-secret-key",
    header_name="X-API-Key",    # default
))

# JWT
auth.register_agent("http://agent-b:8002", AgentCredentialConfig(
    scheme=AuthScheme.JWT,
    jwt_secret="super-secret",
))
token = auth.issue_jwt("http://agent-b:8002", expires_in=3600)

# No auth (dev/testing)
auth.register_agent("http://agent-c:8003", AgentCredentialConfig(
    scheme=AuthScheme.NONE,
))

# Verify an incoming request
from nexus_a2a import AuthError
try:
    await auth.verify("http://agent-a:8001", headers={"X-API-Key": "my-secret-key"})
except AuthError as e:
    print("Auth failed:", e)

Auth fails closed (since v1.5.0). Verifying an agent that was never registered raises UnknownAgentError rather than silently passing as "no auth required". Before v1.5.0 an unregistered URL — a typo, a trailing-slash variant, or an attacker-supplied string — bypassed authentication entirely.
auth = AuthManager()                              # unregistered -> raises
auth = AuthManager(allow_unregistered=True)       # old behaviour, dev only
This applies to inbound verification only. build_auth_headers() still returns {} for unknown agents, so outbound calls to agents you hold no credentials for simply carry no auth headers.

Asymmetric JWT and JWKS

With HS256 the verifier holds the same secret as the signer — so any agent that can check a peer's token can also mint one. In a network of more than two parties, every verifier is a forger. A key pair splits those roles.

pip install "nexus-a2a[jwks]"

The signer keeps the private key:

auth.register_agent("http://summary-agent:8002", AgentCredentialConfig(
    scheme=AuthScheme.JWT,
    jwt_private_key=PRIVATE_PEM,      # signs
    jwt_public_key=PUBLIC_PEM,        # verifies peers
    jwt_algorithm="RS256",            # or ES256, PS256, ...
    jwt_issuer="summary-agent",
))

Verifiers hold only the public half — or fetch it from the issuer's key set, so rotation is a publish rather than a coordinated secret swap:

auth.register_agent("http://summary-agent:8002", AgentCredentialConfig(
    scheme=AuthScheme.JWT,
    jwks_url="https://summary-agent.example.com/.well-known/jwks.json",
    jwt_algorithm="RS256",
    jwt_issuer="summary-agent",
))

Algorithm confusion is structurally prevented. An RSA public key is public, so a verifier that accepted HS256 alongside RS256 would accept a token signed with that public key as an HMAC secret — the classic JWT forgery. Acceptable algorithms come from the configured key, never from the token's alg header, and the symmetric and asymmetric families never overlap. Setting more than one of jwt_secret / jwt_public_key / jwks_url is rejected at registration.

Configured	Accepts
`jwt_secret`	HS256 only
`jwt_public_key` / `jwks_url`	the one configured asymmetric algorithm

HS256 keeps working unchanged, and needs no crypto library — cryptography lives in the jwks extra so the core install stays at 20 packages.

Rate Limiting

Token-bucket algorithm. In-process, zero dependencies.

from nexus_a2a import RateLimiter, RateLimitConfig, RateLimitError

limiter = RateLimiter()
limiter.set_limit("http://agent-a:8001", RateLimitConfig(
    rate=10.0,    # 10 requests per second (sustained)
    burst=20,     # allow bursts up to 20
))

try:
    await limiter.check("http://agent-a:8001")
except RateLimitError as e:
    print(f"Slow down! Retry in {e.retry_after:.2f}s")

RateLimiter is per-process. Three replicas configured for 10 req/s allow 30 — the limit scales with your deployment, silently. Use RedisRateLimiter when the limit has to actually hold:

from nexus_a2a import RedisRateLimiter, SecurityMiddleware

limiter = RedisRateLimiter(url="redis://localhost:6379")
await limiter.connect()

security = SecurityMiddleware(rate_limiter=limiter)   # drop-in
assert limiter.is_distributed          # RateLimiter().is_distributed is False

One bucket per agent lives in Redis, and refill-and-consume runs as a Lua script so it is atomic — a read-modify-write from Python would let two replicas both spend the same last token. The script reads the clock from Redis, so a skewed replica cannot rewind a bucket. Requires Redis 7+.

Trust Boundaries

Default-deny permission matrix between agents.

from nexus_a2a import TrustBoundary

trust = TrustBoundary()

# Allow specific pairs
trust.allow("http://orchestrator:8000", "http://worker-a:8001")
trust.allow("http://orchestrator:8000", "http://worker-b:8002")

# Wildcard: orchestrator can talk to anything
trust.allow("http://orchestrator:8000", "*")

# Block a specific agent
trust.block("http://untrusted:9999", "*")

# Check before sending
if trust.is_allowed("http://orchestrator:8000", "http://worker-a:8001"):
    await client.send_message(msg)

Admin Endpoints

AgentServer serves Kubernetes probes and Prometheus metrics publicly, but the endpoints that expose task data or re-execute work require a token.

Endpoint	Auth	Purpose
`GET /health`	public	Liveness probe
`GET /ready`	public	Readiness probe
`GET /metrics`	public	Prometheus scrape
`GET /info`	admin	Network topology summary
`GET /traces/{id}`	admin	Distributed trace lookup
`GET /dlq`	admin	Dead Letter Queue listing
`POST /dlq/replay`	admin	Re-execute failed tasks

server = AgentServer(network=network, port=8080, admin_token="...")
# or: export NEXUS_ADMIN_TOKEN=...

curl -H "X-Admin-Token: $NEXUS_ADMIN_TOKEN" http://agent:8080/dlq

Admin endpoints are disabled by default (since v1.5.0). With no admin_token and no NEXUS_ADMIN_TOKEN they return 403. Before v1.5.0 they were served unauthenticated on a 0.0.0.0 bind, so anyone who could reach the port could replay every failed task in the queue and read task payloads, error strings, and every registered agent URL.

Payload Validation

from nexus_a2a import PayloadValidator

validator = PayloadValidator(
    max_size_bytes=1_000_000,   # 1 MB
    max_parts=50,
)

from nexus_a2a import PayloadTooLargeError, TooManyPartsError, InvalidPartError, BlankTextPartError
try:
    validator.validate(message)
except (PayloadTooLargeError, TooManyPartsError, InvalidPartError, BlankTextPartError) as e:
    print("Invalid payload:", e)

Mutual TLS (mTLS)

Both agents verify each other's certificates.

from nexus_a2a import MutualTLSConfig, build_client_ssl_context

config = MutualTLSConfig(
    cert_file="/certs/agent.crt",
    key_file="/certs/agent.key",
    ca_file="/certs/ca.crt",
)

# Or from environment variables (NEXUS_MTLS_CERT, NEXUS_MTLS_KEY, NEXUS_MTLS_CA)
config = MutualTLSConfig.from_env()

ssl_ctx = build_client_ssl_context(config)

import httpx
async with httpx.AsyncClient(verify=ssl_ctx) as http:
    # all requests use mTLS
    pass

Storage Backends

In-Memory (default)

from nexus_a2a import InMemoryTaskStore

store = InMemoryTaskStore()

Redis

from nexus_a2a import RedisTaskStore

store = RedisTaskStore(
    url="redis://localhost:6379",
    ttl_seconds=86400,    # tasks expire after 24h
)
await store.connect()

PostgreSQL

from nexus_a2a import PostgresTaskStore

store = PostgresTaskStore(dsn="postgresql://user:pass@localhost/nexus")
await store.connect()   # creates tables if not present

Durable Dead Letter Queue

The DLQ holds failed tasks so they can be replayed. By default it lives in memory, which means a crash or redeploy loses them — and a DLQ entry exists precisely because that work did not complete.

from nexus_a2a import AgentNetwork, RedisDLQStore

store = RedisDLQStore(url="redis://localhost:6379")
await store.connect()

network = AgentNetwork(dlq_store=store)
await network.dead_letter_queue.load()      # rehydrate after a restart

dlq = network.dead_letter_queue
dlq.is_durable                   # True
await dlq.load()                 # entries captured before the restart
await dlq.replay_all()           # and they are replayable again
await dlq.purge_replayed()       # removes from the store too

Durable push targets

A webhook is registered by a caller who will not wait around — so losing it on restart defeats the purpose.

from nexus_a2a import A2AServer, RedisPushStore, WebhookConfig

store = RedisPushStore(url="redis://localhost:6379")
await store.connect()

server = A2AServer(
    MyAgent,
    push_store=store,
    push_config=WebhookConfig(signing_secret="shared-secret"),
)

Targets are released once a task reaches a terminal state, and carry a TTL for tasks that never finish.

A push config holds the caller's token — the secret their receiver checks callbacks against. Persisting it puts that secret at rest in the store, so point RedisPushStore at a Redis you would be willing to keep credentials in: authenticated, and TLS if it is not on loopback.

clear_replayed() is synchronous so it cannot reach an async store — it only clears the local view, and entries return on the next load(). Use purge_replayed() when a store is configured.

Task Manager

Wraps any store with lifecycle operations and a watchdog that auto-fails stuck tasks.

from nexus_a2a import TaskManager

manager = TaskManager(
    store=store,
    timeout_sec=120.0,    # auto-fail tasks stuck in WORKING after 2 min
)

task = await manager.create(message=Message.user_text("hello"))
await manager.start(task.id)
await manager.complete(task.id, artifact_text="done")

# Or fail with reason
await manager.fail(task.id, error="API call timed out")

Reliability

Dead Letter Queue (DLQ)

Failed tasks land in the DLQ for inspection and replay.

from nexus_a2a import DeadLetterQueue

dlq = DeadLetterQueue(
    max_retries=3,
    retry_delay=2.0,       # seconds between retries (exponential backoff)
    max_queue_size=500,
)

# Capture a failed task
entry = await dlq.capture(
    task,
    agent_url="http://worker:8001",
    skill_id="web_search",
)

# Inspect
print(dlq.count())           # total entries
print(dlq.pending_count())   # not yet replayed
entries = dlq.all_entries()  # list[DLQEntry]
pending = dlq.pending_entries()

# Filter by skill
web_failures = [e for e in dlq.all_entries() if e.skill_id == "web_search"]

# Replay a single task
result = await dlq.replay(task_id="abc-123")
print(result.succeeded, result.error)

# Replay all pending
results = await dlq.replay_all()

# Replay filtered by skill or agent
results = await dlq.replay_where(skill_id="web_search")
results = await dlq.replay_where(agent_url="http://worker:8001")

# Failure hook
@dlq.on_failure
async def on_fail(entry):
    print(f"Task {entry.task_id} failed: {entry.error}")

# Clean up replayed entries
removed = dlq.clear_replayed()
print(f"Cleaned {removed} entries")

Input Required (Human-in-the-Loop)

Pause a task, wait for human input, then resume.

from nexus_a2a import InputHandler

handler = InputHandler()

# Pause and wait for input (async — does not block other tasks)
await handler.request_input(task.id, prompt="Please provide your API key:")
response = await handler.wait_for_input(task.id, timeout=300.0)

# From another process / API endpoint — resume the task
await handler.provide_input(task.id, "user-provided-api-key")

Graceful Shutdown

from nexus_a2a import GracefulShutdown

shutdown = GracefulShutdown(
    manager=task_manager,
    drain_timeout=30.0,    # wait up to 30s for WORKING tasks to finish
)

# Registers SIGTERM and SIGINT handlers automatically
shutdown.register()

# Manual shutdown (e.g. from a test or lifecycle hook)
await shutdown.shutdown()

Streaming & Webhooks

SSE Streaming

An agent streams by writing run() as an async generator — every yield becomes one chunk on the wire:

from nexus_a2a import agent, A2AServer, Task

@agent(
    name="Writer",
    description="Writes text a word at a time.",
    streaming=True,
    skills=[{"id": "write", "name": "Write", "description": "Write text."}],
    url="http://localhost:8001",
)
class Writer:
    async def run(self, task: Task):
        for word in ["Hello", ", ", "world", "!"]:
            yield word

await A2AServer(Writer, port=8001).start()

Consume it with stream_message():

from nexus_a2a import A2AHttpClient, Message
from nexus_a2a import StreamEventType

async with A2AHttpClient("http://localhost:8001") as client:
    async for event in client.stream_message(Message.user_text("go")):
        if event.type == StreamEventType.ARTIFACT_CHUNK:
            print(event.data["content"], end="", flush=True)

Event sequence: task_created → artifact_chunk (one per yield) → task_status → done. Iteration stops at the first terminal event. task_created carries the task id, so you can poll or cancel while it runs.

If you prefer a separate method, add stream() instead — it wins over run() when both exist. A class attribute STREAMING = True declares that a framework adapter handles streaming internally.

The two calling styles are interchangeable. Neither side needs to know how the other is written:

	`message/send`	`message/stream`
Streaming agent	chunks folded into one artifact	one chunk per yield
Non-streaming agent	normal result	whole output as one chunk

Observing a task: GET /stream?taskId=... reports a task's state and artifacts, then closes — this is what SSEStreamer targets. It does not follow a run in progress, because message/send executes the agent inline and the task is already terminal by the time it can be looked up. Use message/stream to watch work as it happens.

Security refusals happen before the stream opens — once SSE starts the status line is already sent and an HTTP status can no longer be signalled. A failure after that point arrives as a terminal error event and the task is marked FAILED.

Push notifications (webhooks)

For callers that will not sit and wait — a long task, a mobile client, a serverless function, or a task that pauses at INPUT_REQUIRED and needs to tell somebody. The agent POSTs task_completed, task_failed, task_cancelled and task_input_required to a URL you register.

from nexus_a2a import A2AServer, WebhookConfig

server = A2AServer(
    MyAgent,
    push_config=WebhookConfig(signing_secret="shared-secret"),
)

async with A2AHttpClient("http://localhost:8001") as client:
    task = await client.send_message(
        Message.user_text("go"),
        push_notification={"url": "https://me.example.com/hook", "token": "abc"},
    )

    # ...or against an existing task
    await client.set_push_config(task.id, {"url": "https://me.example.com/hook"})

Verify the signature on your side:

from nexus_a2a import WebhookDispatcher

async def hook(request):
    body = await request.body()
    ok = WebhookDispatcher.verify_signature(
        body, request.headers["X-Nexus-Signature-256"], "shared-secret"
    )

Webhook URLs are an SSRF vector. The URL comes from whoever called the agent, and the server then makes a request to it. URLs resolving to private, loopback, link-local or reserved addresses are refused, as is any scheme other than http/https. Pass WebhookConfig(allow_private_urls=True) only for local development.

Situation	Result
Agent declares `push_notifications=False`	`-32003`
URL is private / bad scheme	`-32004`
Config missing `url`	`-32602`

Delivery is detached, so a slow or dead receiver never stalls the response or fails the task. The registered target is dropped once the task is terminal. Targets live in memory and do not survive a restart.

Dispatching manually

from nexus_a2a import WebhookDispatcher

dispatcher = WebhookDispatcher(
    secret="your-webhook-secret",
    max_retries=3,
)

# Dispatch — auto-retries on 5xx, skips 4xx (client error = no retry)
await dispatcher.dispatch(
    url="https://your-app.com/webhooks/nexus",
    event="task.completed",
    payload={"task_id": "abc-123", "result": "..."},
)

# Non-raising version (logs failures silently)
await dispatcher.dispatch_silent(url=..., event=..., payload=...)

# Verify incoming webhook on your server
is_valid = dispatcher.verify_signature(
    payload=request.body,
    signature=request.headers["X-Nexus-Signature-256"],
)

Framework Adapters

Wrap existing agents from popular frameworks with zero changes to your existing code.

LangGraph

from nexus_a2a import LangGraphAdapter

adapter = LangGraphAdapter(graph=compiled_graph)
result = await adapter.run(task)
print(result.text)

CrewAI

from nexus_a2a import CrewAIAdapter

adapter = CrewAIAdapter(crew=my_crew)
result = await adapter.run(task)

AutoGen

from nexus_a2a import AutoGenAdapter

adapter = AutoGenAdapter(agent=my_autogen_agent)
result = await adapter.run(task)

Google ADK

from nexus_a2a import GoogleADKAdapter

adapter = GoogleADKAdapter(agent=my_adk_agent)
result = await adapter.run(task)

Observability & CLI

`nexus` CLI

Install with the package, then:

# Ping an agent — name, version, skills, latency, health
nexus ping http://localhost:8001

# Inspect full AgentCard
nexus inspect http://localhost:8001

# Network status table (all agents, queue depth, DLQ count)
nexus status --network

# Trace a task — call tree with per-hop latency
nexus trace abc-123-task-id
nexus trace abc-123-task-id --agent http://localhost:8001

# Replay DLQ entries
nexus replay --failed
nexus replay --failed --skill web_search
nexus replay --failed --last 1h
nexus replay --failed --dry-run        # preview without replaying

# Serve an agent over the A2A protocol
nexus run --module mypackage.agent:MyAgent
nexus run --module mypackage.agent:MyAgent --host 0.0.0.0 --port 8080

# Without --module: ops server only (health, metrics, admin)
nexus run

# JSON output for all commands
nexus --format json status --network

Runnable as a module too, when the console script is not on PATH (a container, a CI step, an uninstalled virtualenv):
python -m nexus_a2a.cli ping http://localhost:8001

Audit Logger

from nexus_a2a import AuditLogger
import sys

logger = AuditLogger(
    stream=sys.stdout,         # or open("audit.ndjson", "a")
    buffer_size=100,           # flush every 100 events
)

# 8 event types: task_created, task_started, task_completed,
# task_failed, task_cancelled, message_sent, auth_failed, rate_limited
await logger.log("task_completed", {
    "task_id": task.id,
    "agent_url": "http://worker:8001",
    "duration_sec": 1.23,
})

Metrics

from nexus_a2a import MetricsCollector

metrics = MetricsCollector()

# Record events
await metrics.record_task_completed(agent_url="http://worker:8001", duration_sec=1.2)
await metrics.record_task_failed(agent_url="http://worker:8001")
await metrics.record_auth_failure(agent_url="http://worker:8001")

# Query
print(metrics.task_count())               # total tasks
print(metrics.error_rate())               # 0.0 – 1.0
print(metrics.avg_latency_sec())          # float
print(metrics.p99_latency_sec())          # float

# Prometheus text format (expose via /metrics endpoint)
text = metrics.to_prometheus()

Distributed Tracing

from nexus_a2a import Tracer, TraceStore

tracer = Tracer()
trace_store = TraceStore(max_traces=1000)

# Start a trace
trace_id = tracer.new_trace_id()

# Use in client — automatically injected into X-Nexus-Trace-ID header
async with A2AHttpClient(
    "http://localhost:8001",
    trace_id=trace_id,
    trace_store=trace_store,
) as client:
    task = await client.send_message(Message.user_text("hello"))

# Retrieve trace
trace = trace_store.get(trace_id)
print(tracer.format_tree(trace))

Configuration (nexus.toml)

Zero-config wiring from a single TOML file.

[agent]
name        = "ResearchAgent"
description = "Searches the web and summarises results."
version     = "1.0.0"
url         = "http://localhost:8001"

[[agent.skills]]
id          = "web_search"
name        = "Web Search"
description = "Searches the web for a given query."
tags        = ["search", "web"]

[storage]
backend = "redis"               # memory | redis | postgres
url     = "redis://localhost:6379"

[security]
auth_scheme     = "api_key"     # none | api_key | jwt
auth_secret     = "my-secret"
mtls_cert_file  = "/certs/agent.crt"
mtls_key_file   = "/certs/agent.key"
mtls_ca_file    = "/certs/ca.crt"

[reliability]
task_timeout_sec = 120.0
dlq_max_size     = 500
dlq_max_retries  = 3

[observability]
log_level  = "INFO"
tracing    = true

[network]
agents = [
    "http://agent-a:8001",
    "http://agent-b:8002",
]

Load programmatically:

from nexus_a2a import AgentNetwork

network = AgentNetwork.from_config("nexus.toml")

Environment variable overrides (container-friendly):

Variable	Overrides
`NEXUS_AGENT_NAME`	`[agent].name`
`NEXUS_AGENT_URL`	`[agent].url`
`NEXUS_AUTH_SCHEME`	`[security].auth_scheme`
`NEXUS_AUTH_SECRET`	`[security].auth_secret`
`NEXUS_STORAGE_BACKEND`	`[storage].backend`
`NEXUS_STORAGE_URL`	`[storage].url`
`NEXUS_TASK_TIMEOUT`	`[reliability].task_timeout_sec`
`NEXUS_LOG_LEVEL`	`[observability].log_level`

Error Handling Reference

Every error in nexus-a2a is typed. Catch specific exceptions rather than bare Exception.

Transport errors

from nexus_a2a import (
    AgentUnreachableError,   # agent is down / DNS failure
    AgentCardFetchError,     # /.well-known/agent-card.json failed
    RemoteAgentError,        # agent returned JSON-RPC error
    CircuitOpenError,        # circuit breaker is OPEN
    TransportError,          # base class for all transport errors
)

try:
    task = await client.send_message(msg)
except CircuitOpenError as e:
    print(f"Circuit open. Retry in {e.retry_after}s")
except AgentUnreachableError as e:
    print(f"Cannot reach {e.url}: {e.reason}")
except RemoteAgentError as e:
    print(f"Agent error {e.code}: {e.message}")
except TransportError as e:
    print(f"Transport error: {e}")

Auth errors

from nexus_a2a import (
    AuthError,                   # base
    MissingCredentialsError,     # no credentials in request
    InvalidCredentialsError,     # wrong key / bad token
    ExpiredCredentialsError,     # JWT expired
)

try:
    await auth.verify(url, headers)
except ExpiredCredentialsError:
    new_token = auth.issue_jwt(url, expires_in=3600)
except InvalidCredentialsError as e:
    print(f"Auth failed: {e.reason}")

Rate limit errors

from nexus_a2a import RateLimitError

try:
    await limiter.check(agent_url)
except RateLimitError as e:
    await asyncio.sleep(e.retry_after)
    await limiter.check(agent_url)   # retry

Task state errors

from nexus_a2a import TaskNotFoundError, TaskAlreadyDoneError

try:
    await manager.complete(task_id)
except TaskNotFoundError:
    print("Task does not exist")
except TaskAlreadyDoneError:
    print("Task already in terminal state")

Orchestration errors

from nexus_a2a import (
    OrchestratorError,      # base
    WorkflowCycleError,     # DAG has a cycle
    WorkflowStepError,      # individual step failed
)

try:
    result = await orchestrator.dag(nodes, initial_message)
except WorkflowCycleError as e:
    print(f"Cycle detected: {e.cycle}")

Config errors

from nexus_a2a import ConfigError

try:
    network = AgentNetwork.from_config("nexus.toml")
except ConfigError as e:
    print(f"Bad config at key '{e.key}': {e}")

Testing

Integration tests

Real in-process agents on random ports — no HTTP mocking.

import pytest
from starlette.applications import Starlette
import uvicorn

# See tests/integration/conftest.py for full fixture helpers
# Run with:
pytest tests/integration/

Run all tests

# Unit + integration
pytest

# Skip integration (faster CI)
pytest -m "not integration"

# With coverage
pytest --cov=nexus_a2a --cov-report=term-missing

# Specific test file
pytest tests/integration/test_sequential_pipeline.py -v

Type checking and linting

mypy nexus_a2a --strict
ruff check nexus_a2a
ruff format nexus_a2a

What's in each version

Version	What it added
v0.1.0	`@agent` decorator, `Task` state machine, `Message`/`Part`/`Artifact` models
v0.2.0	`InMemoryTaskStore`, `TaskManager`, `A2AHttpClient`, `AgentRegistry`
v0.3.0	`AuthManager`, `TrustBoundary`, `RateLimiter`, `PayloadValidator`
v0.4.0	`Orchestrator` (sequential/parallel/dag), SSE streaming, `WebhookDispatcher`, `AgentNetwork`
v1.0.0	LangGraph/CrewAI/AutoGen/GoogleADK adapters, `RedisTaskStore`, `AuditLogger`, `MetricsCollector`
v1.1.0	Task timeout watchdog, `InputHandler`, `DeadLetterQueue`, `CircuitBreaker`, `Tracer`, `CapabilityGuard`
v1.8.0	Production scale: durable DLQ and push targets, distributed rate limiting (atomic via Lua), asymmetric JWT + JWKS with algorithm-confusion prevention
v1.7.0	Streaming, multi-turn (`NeedsInput`) and push notifications. Every capability a card advertises is now served; HMAC webhook signing fixed (it had never validated) and webhook URLs are SSRF-checked
v1.6.0	`A2AServer` (inbound protocol: agent card + JSON-RPC), `SecurityMiddleware`, `caller_url` on the client, `nexus run` actually serves the agent
v1.5.0	Security hardening: admin endpoints gated, auth fails closed, JWT audience validated, PyJWT replaces python-jose, `google-adk` moved to an extra
v1.2.0	`GracefulShutdown`, `AgentServer` (K8s probes), mTLS, `PostgresTaskStore`, `nexus.toml`, CI/CD workflows
v1.3.0	`nexus` CLI (ping/inspect/status/trace/replay/run), integration test suite, `CHANGELOG.md`

License

MIT — see LICENSE.

Built by HongZiro.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

nexus_a2a-1.8.0.tar.gz (369.2 kB view details)

Uploaded Sep 14, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

nexus_a2a-1.8.0-py3-none-any.whl (187.8 kB view details)

Uploaded Sep 14, 2026 Python 3

File details

Details for the file nexus_a2a-1.8.0.tar.gz.

File metadata

Download URL: nexus_a2a-1.8.0.tar.gz
Upload date: Sep 14, 2026
Size: 369.2 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.13.0

File hashes

Hashes for nexus_a2a-1.8.0.tar.gz
Algorithm	Hash digest
SHA256	`6c282e73f6741a893f827f4e46c59b6f9c661886306636b5a4e20e05ddf38298`
MD5	`40014ca895ccbf6a6e4a00d580f473f2`
BLAKE2b-256	`26624aa6790139151a943e900e26620dceeb0e874e8b067eabf069db74146473`

See more details on using hashes here.

File details

Details for the file nexus_a2a-1.8.0-py3-none-any.whl.

File metadata

Download URL: nexus_a2a-1.8.0-py3-none-any.whl
Upload date: Sep 14, 2026
Size: 187.8 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.13.0

File hashes

Hashes for nexus_a2a-1.8.0-py3-none-any.whl
Algorithm	Hash digest
SHA256	`065d491eed00413e0ccfbb175d284ca7363004b948e2602b062b851147f7b75a`
MD5	`84fccbeab63203e79bc8699f5eb01c58`
BLAKE2b-256	`8f6e1b5da8b6c24844926e7ff0972e88d89a40c98b116bfe9549008375a1c12d`

See more details on using hashes here.

Release history Release notifications | RSS feed

This release

1.8.0 This release

Sep 14, 2026 2 files

Sep 10, 2026 2 files

Sep 9, 2026 2 files

Aug 5, 2026 2 files

Jul 25, 2026 2 files

Jun 1, 2026 2 files

Jun 1, 2026 2 files

Jun 1, 2026 2 files

Apr 10, 2026 2 files

Apr 10, 2026 2 files

Apr 9, 2026 2 files

Apr 8, 2026 2 files

Apr 3, 2026 2 files

nexus-a2a 1.8.0

nexus-a2a

❤️ Support

What is nexus-a2a?

Table of Contents

Installation

Quickstart

1. Define your agent

2. Start the server

3. Send a task from another agent

Core Concepts

Task state machine

Message and Part

Building Agents

@agent decorator

Serving Agents

What run() can return

Multi-turn conversations

Error model

Enforcing security

Sending Tasks

A2AHttpClient

Circuit Breaker

Agent Registry & Discovery

Orchestration

Sequential pipeline

Parallel

DAG (Directed Acyclic Graph)

Security

Authentication

Asymmetric JWT and JWKS

Rate Limiting

Trust Boundaries

Admin Endpoints

Payload Validation

Mutual TLS (mTLS)

Storage Backends

In-Memory (default)

Redis

PostgreSQL

Durable Dead Letter Queue

Durable push targets

Task Manager

Reliability

Dead Letter Queue (DLQ)

Input Required (Human-in-the-Loop)

Graceful Shutdown

Streaming & Webhooks

SSE Streaming

Push notifications (webhooks)

Dispatching manually

Framework Adapters

LangGraph

CrewAI

AutoGen

Google ADK

Observability & CLI

nexus CLI

Audit Logger

Metrics

Distributed Tracing

Configuration (nexus.toml)

Error Handling Reference

Transport errors

Auth errors

Rate limit errors

Task state errors

Orchestration errors

Config errors

Testing

Integration tests

Run all tests

Type checking and linting

What's in each version

License

Project links

Key dates

1 maintainer

Credits

License

`@agent` decorator

What `run()` can return

`nexus` CLI