apcore-a2a (Python)

What is apcore-a2a?

apcore-a2a is the A2A (Agent-to-Agent) protocol adapter for the apcore ecosystem.

It solves a common problem: you've built AI capabilities with apcore modules, but you need them to talk to other AI agents over a standard protocol. apcore-a2a bridges that gap — it reads your existing module metadata (schemas, descriptions, examples) and automatically exposes them as a standards-compliant A2A server. No hand-written Agent Cards, no JSON-RPC boilerplate, no manual task lifecycle management.

In short: apcore modules + apcore-a2a = a fully functional A2A agent, ready to be discovered and invoked by any A2A-compatible client.

Also available in: TypeScript | Rust

Features

• One-call server — launch a compliant A2A server with serve(registry)
• Automatic Agent Card — /.well-known/agent-card.json generated from module metadata (the 0.3 alias /.well-known/agent.json is also served)
• Skill mapping — apcore modules become A2A Skills with names, descriptions, tags, and examples; metadata["display"]["a2a"] overrides surface-facing fields (§5.13)
• Full task lifecycle — submitted, working, completed, failed, canceled, input-required
• SSE streaming — message/stream with real-time status and artifact updates
• Push notifications — optional webhook delivery of task state changes
• JWT authentication — tokens bridged to apcore's Identity context
• A2A Explorer UI — browser UI for discovering and testing skills
• Built-in client — A2AClient for calling remote A2A agents
• Pluggable storage — swap in Redis or PostgreSQL via the TaskStore protocol
• Observability — /health, /metrics endpoints, structured logging
• Dynamic registration — add/remove modules at runtime without restart

Requirements

• Python >= 3.11
• apcore >= 0.30.0
• apcore-toolkit >= 0.11.1 (the openapi extra additionally needs apcore-toolkit[http-proxy])

For Users: Getting Started

Installation

pip install apcore-a2a

Expose your modules as an A2A Agent

If you already have apcore modules, a few lines turn them into a discoverable agent:

from apcore import Executor, Registry
from apcore_a2a import serve

registry = Registry(extensions_dir="./extensions")
registry.discover()

serve(Executor(registry))  # Starts on http://0.0.0.0:8000

Your agent is now live at http://localhost:8000/.well-known/agent-card.json (the 0.3 alias /.well-known/agent.json is also served).

Serve an OpenAPI document instead

No apcore project required — point it at an OpenAPI 3.0/3.1 document and every operation becomes an A2A Skill, proxied over HTTP to the API that published it:

pip install "apcore-a2a[openapi]"

apcore-a2a serve --from-openapi https://petstore3.swagger.io/api/v3/openapi.json \
                 --openapi-prefix petstore

Or programmatically:

from apcore_a2a import openapi_backend, serve

registry = openapi_backend(
    "https://petstore3.swagger.io/api/v3/openapi.json",
    prefix="petstore",
)
serve(registry)

[!WARNING] An OpenAPI document describes an API's shape, not the consequences of calling it, so requires_approval is never inferred — a POST /charges that moves money is annotated exactly like a POST /echo. Such an operation is therefore advertised on the public Agent Card, which is served without authentication. apcore-a2a warns about this at startup, and the warning is not silenced by merely attaching an ACL.

The recommended shape is --openapi-prefix plus an ACL with default_effect: deny and a prefixed catch-all deny rule; see the feature spec for why an allow-list of operation names is the fail-safe direction when an upstream API renames something.

Try the Examples

Run all 5 example modules (3 class-based + 2 binding YAML) with the Explorer UI:

PYTHONPATH=./examples/binding_demo python examples/run.py

Open http://127.0.0.1:8000/explorer/ to browse skills, send messages, and stream responses.

See examples/README.md for more options (CLI, binding-only, JWT auth).

Call a remote A2A Agent

Use the built-in client to discover and invoke any A2A-compliant agent:

import asyncio
from apcore_a2a import A2AClient

async def main():
    async with A2AClient("http://remote-agent:8000") as client:
        # Discover what the agent can do
        card = await client.discover()
        print(f"Agent: {card['name']}, Skills: {len(card['skills'])}")

        # Send a message (route to a skill via metadata.skillId)
        message = {"role": "user", "parts": [{"kind": "text", "text": "Hello!"}]}
        task = await client.send_message(
            message,
            metadata={"skillId": "my.skill"},
        )
        print(f"Result: {task['status']['state']}")

        # Or stream the response
        async for event in client.stream_message(message, metadata={"skillId": "my.skill"}):
            print(event)

asyncio.run(main())

Add authentication

from apcore_a2a import serve
from apcore_a2a.auth.jwt import JWTAuthenticator, ClaimMapping

auth = JWTAuthenticator(
    key="your-secret-key",
    algorithms=["HS256"],
    issuer="https://auth.example.com",
    audience="my-agent",
    claim_mapping=ClaimMapping(
        id_claim="sub",
        type_claim="type",
        roles_claim="roles",
        attrs_claims=["org", "dept"],
    ),
    require_claims=["sub"],
)

serve(registry, auth=auth)

For Developers: API Reference

serve()

Blocking call — starts uvicorn and serves until SIGINT/SIGTERM.

from apcore_a2a import serve

serve(
    registry_or_executor,   # apcore Registry or Executor
    *,
    host="0.0.0.0",
    port=8000,
    name=None,              # Agent name (fallback: registry config)
    description=None,       # Agent description
    version=None,           # Agent version
    url=None,               # Public URL (default: f"http://{host}:{port}")
    auth=None,              # Authenticator instance
    task_store=None,        # TaskStore instance (default: InMemoryTaskStore)
    cors_origins=None,      # List of allowed CORS origins
    push_notifications=False,
    explorer=False,         # Enable A2A Explorer UI
    explorer_prefix="/explorer",
    cancel_on_disconnect=True,
    shutdown_timeout=30,
    execution_timeout=300,
    log_level=None,
    metrics=False,          # Enable /metrics endpoint
    sys_modules=False,      # Register apcore sys.* modules (requires executor.use())
)

async_serve()

Returns the ASGI app without starting a server — use for embedding in larger applications.

from apcore_a2a import async_serve

app = await async_serve(registry_or_executor, **kwargs)
# app is a Starlette ASGI application

TaskStore

Default in-memory task store. Implement the TaskStore protocol for persistent backends (Redis, PostgreSQL, etc.).

from apcore_a2a.storage import InMemoryTaskStore

store = InMemoryTaskStore()
serve(registry, task_store=store)

Architecture

apcore-a2a acts as a thin protocol layer on top of apcore. The mapping is straightforward:

Contributing

git clone https://github.com/aiperceivable/apcore-a2a-python.git
cd apcore-a2a-python
pip install -e ".[dev]"
pytest

Documentation

• Product Requirements (PRD)
• Technical Design
• Software Requirements (SRS)

License

Apache 2.0 — see LICENSE.