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
- One-call server — launch a compliant A2A server with
serve(registry) - Automatic Agent Card —
/.well-known/agent-card.jsongenerated from module metadata (the 0.3 alias/.well-known/agent.jsonis 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/streamwith 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 —
A2AClientfor calling remote A2A agents - Pluggable storage — swap in Redis or PostgreSQL via the
TaskStoreprotocol - Observability —
/health,/metricsendpoints, structured logging - Dynamic registration — add/remove modules at runtime without restart
- Python >= 3.11
apcore>= 0.22.0apcore-toolkit>= 0.8.0
pip install apcore-a2aIf you already have apcore modules, a few lines turn them into a discoverable agent:
fromapcoreimportExecutor, Registryfromapcore_a2aimportserveregistry=Registry(extensions_dir="./extensions")
registry.discover()
serve(Executor(registry)) # Starts on http://0.0.0.0:8000Your agent is now live at http://localhost:8000/.well-known/agent-card.json (the 0.3 alias /.well-known/agent.json is also served).
Run all 5 example modules (3 class-based + 2 binding YAML) with the Explorer UI:
PYTHONPATH=./examples/binding_demo python examples/run.pyOpen 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).
Use the built-in client to discover and invoke any A2A-compliant agent:
importasynciofromapcore_a2aimportA2AClientasyncdefmain():
asyncwithA2AClient("http://remote-agent:8000") asclient:
# Discover what the agent can docard=awaitclient.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=awaitclient.send_message(
message,
metadata={"skillId": "my.skill"},
)
print(f"Result: {task['status']['state']}")
# Or stream the responseasyncforeventinclient.stream_message(message, metadata={"skillId": "my.skill"}):
print(event)
asyncio.run(main())fromapcore_a2aimportservefromapcore_a2a.auth.jwtimportJWTAuthenticator, ClaimMappingauth=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)Blocking call — starts uvicorn and serves until SIGINT/SIGTERM.
fromapcore_a2aimportserveserve(
registry_or_executor, # apcore Registry or Executor*,
host="0.0.0.0",
port=8000,
name=None, # Agent name (fallback: registry config)description=None, # Agent descriptionversion=None, # Agent versionurl=None, # Public URL (default: f"http://{host}:{port}")auth=None, # Authenticator instancetask_store=None, # TaskStore instance (default: InMemoryTaskStore)cors_origins=None, # List of allowed CORS originspush_notifications=False,
explorer=False, # Enable A2A Explorer UIexplorer_prefix="/explorer",
cancel_on_disconnect=True,
shutdown_timeout=30,
execution_timeout=300,
log_level=None,
metrics=False, # Enable /metrics endpointsys_modules=False, # Register apcore sys.* modules (requires executor.use())
)Returns the ASGI app without starting a server — use for embedding in larger applications.
fromapcore_a2aimportasync_serveapp=awaitasync_serve(registry_or_executor, **kwargs)
# app is a Starlette ASGI applicationDefault in-memory task store. Implement the TaskStore protocol for persistent backends (Redis, PostgreSQL, etc.).
fromapcore_a2a.storageimportInMemoryTaskStorestore=InMemoryTaskStore()
serve(registry, task_store=store)apcore-a2a acts as a thin protocol layer on top of apcore. The mapping is straightforward:
| A2A Concept | apcore Mapping |
|---|---|
| Agent Card | Derived from Registry configuration |
| Skill id | module_id |
| Skill name | metadata["display"]["a2a"]["alias"] or humanized module_id |
| Skill desc | metadata["display"]["a2a"]["description"] or module.description |
| Skill tags | metadata["display"]["tags"] or module.tags |
| Task | Managed execution of Executor.call_async() |
| Streaming | Wrapped Executor.stream() via SSE |
| Security | Bridged to apcore's Identity context |
git clone https://github.com/aiperceivable/apcore-a2a-python.git
cd apcore-a2a-python
pip install -e ".[dev]"
pytestApache 2.0 — see LICENSE.