Agent-to-Agent
Agent-to-agent delegation using the A2A protocol.
The Agent-to-Agent package adds Keycard authentication to the A2A protocol, so one AI agent can delegate tasks to another while preserving the user’s identity and authorization context.
When to Use
Section titled “When to Use”- Building agents that delegate work to other specialized agents
- Exposing an agent as a service that other agents can call
- Wiring Keycard auth into an existing a2a-sdk server
- Implementing the A2A protocol with Keycard authentication
Installation
Section titled “Installation”pip install keycardai-a2aPulls in keycardai-oauth, keycardai-starlette, and a2a-sdk[http-server] >= 1.0.
npm install @keycardai/a2aPulls in @a2a-js/sdk and Express handlers.
Key Exports
Section titled “Key Exports”Both packages follow a wrap-don’t-reinvent pattern: you implement the executor against the native A2A SDK, and the Keycard package contributes auth wiring, agent card construction, and outbound delegation.
keycardai.a2a
| Export | Description |
|---|---|
AgentServiceConfig | Service identity, Keycard credentials, and agent card metadata |
build_agent_card_from_config | Construct an a2a.types.AgentCard from an AgentServiceConfig |
KeycardServerCallContextBuilder | Propagate the verified bearer token onto ServerCallContext.state so executors can use it for downstream token exchange |
DelegationClient | Async client for remote agent invocation with Keycard token exchange |
DelegationClientSync | Synchronous variant of DelegationClient |
ServiceDiscovery | Resolve a remote service’s .well-known/agent-card.json with caching |
Server-side auth uses KeycardAuthBackend from keycardai-starlette. Executors implement a2a.server.agent_execution.AgentExecutor directly from a2a-sdk.
@keycardai/a2a
| Export | Description |
|---|---|
AgentServiceConfig | Service identity and agent card metadata (type) |
buildAgentCard | Construct an A2A AgentCard from AgentServiceConfig |
createKeycardRequestHandler | Wire an executor and agent card into the A2A request handler |
keycardUserBuilder | Validate Keycard JWTs on incoming A2A requests |
getKeycardAuth | Extract the verified AccessToken from a RequestContext |
KeycardUser | User type the auth builder injects into the A2A request context |
DelegationClient | Call remote agents with Keycard token exchange |
ServiceDiscovery | Discover available agent services |
The package also re-exports Express handlers (agentCardHandler, jsonRpcHandler) and types (AgentCard, Message, Task, AgentExecutor) from @a2a-js/sdk.
Quickstart
Section titled “Quickstart”Serve an Agent
Section titled “Serve an Agent”Compose the Keycard helpers with a2a-sdk’s route factories and Starlette’s AuthenticationMiddleware:
from a2a.server.agent_execution import AgentExecutorfrom a2a.server.request_handlers import DefaultRequestHandlerfrom a2a.server.routes import create_agent_card_routes, create_jsonrpc_routesfrom a2a.server.tasks import InMemoryTaskStorefrom starlette.applications import Starlettefrom starlette.middleware import Middlewarefrom starlette.middleware.authentication import AuthenticationMiddlewarefrom starlette.routing import Mount
from keycardai.a2a import ( AgentServiceConfig, KeycardServerCallContextBuilder, build_agent_card_from_config,)from keycardai.oauth.server.credentials import ClientSecretfrom keycardai.starlette import AuthProvider, KeycardAuthBackend, keycard_on_error
config = AgentServiceConfig( service_name="Research Agent", description="Searches the web and summarizes findings", client_id="<client-id>", client_secret="<client-secret>", identity_url="http://localhost:9000", zone_id="<zone-id>",)
auth_provider = AuthProvider( zone_url=config.auth_server_url, server_name=config.service_name, server_url=config.identity_url, application_credential=ClientSecret((config.client_id, config.client_secret)),)
class ResearchExecutor(AgentExecutor): async def execute(self, context, event_queue): # The verified bearer token is available for downstream delegation token = context.call_context.state["access_token"] # Process the delegated task...
async def cancel(self, context, event_queue): pass
agent_card = build_agent_card_from_config(config)request_handler = DefaultRequestHandler( agent_executor=ResearchExecutor(), task_store=InMemoryTaskStore(), agent_card=agent_card,)
app = Starlette(routes=[ *create_agent_card_routes(agent_card=agent_card), Mount( "/a2a", routes=create_jsonrpc_routes( request_handler=request_handler, rpc_url="/jsonrpc", context_builder=KeycardServerCallContextBuilder(), ), middleware=[Middleware( AuthenticationMiddleware, backend=KeycardAuthBackend( auth_provider.get_token_verifier(), require_authentication=True, ), on_error=keycard_on_error, )], ),])For a runnable end-to-end version, see examples/keycard_protected_server in the SDK repo.
The TypeScript package builds on top of @a2a-js/sdk and Express. Compose the Keycard helpers with the SDK’s Express handlers:
import express from "express";import { buildAgentCard, createKeycardRequestHandler, keycardUserBuilder, getKeycardAuth, agentCardHandler, jsonRpcHandler, type AgentExecutor,} from "@keycardai/a2a";
const config = { serviceName: "Research Agent", description: "Searches the web and summarizes findings", clientId: process.env.KEYCARD_CLIENT_ID!, clientSecret: process.env.KEYCARD_CLIENT_SECRET!, identityUrl: "http://localhost:9000", zoneId: process.env.KEYCARD_ZONE_ID!,};
const executor: AgentExecutor = { async execute(requestContext, eventBus) { const auth = getKeycardAuth(requestContext); if (!auth) throw new Error("unauthenticated");
// Process the delegated task eventBus.publish({ messageId: crypto.randomUUID(), role: "agent", parts: [{ kind: "text", text: "Research complete" }], }); eventBus.finished(); }, async cancelTask() {},};
const agentCard = buildAgentCard(config);const requestHandler = createKeycardRequestHandler(executor, agentCard);const userBuilder = keycardUserBuilder({ issuer: `https://${config.zoneId}.keycard.cloud`,});
const app = express();app.use(express.json());app.use("/.well-known/agent-card.json", agentCardHandler({ agentCardProvider: requestHandler }));app.use("/a2a/jsonrpc", jsonRpcHandler({ requestHandler, userBuilder }));app.listen(9000);Call a Remote Agent
Section titled “Call a Remote Agent”Use DelegationClient from inside your AgentExecutor to invoke a downstream agent on behalf of the calling user:
from keycardai.a2a import DelegationClient
client = DelegationClient(config) # same AgentServiceConfig from the server setup
class ResearchExecutor(AgentExecutor): async def execute(self, context, event_queue): user_token = context.call_context.state["access_token"]
# Exchange the user's token for a delegation token bound to the remote service delegation_token = await client.get_delegation_token( "<remote-agent-url>", subject_token=user_token, )
result = await client.invoke_service( "<remote-agent-url>", {"task": "Summarize the latest news on AI agents"}, delegation_token, ) # Forward result.message back to the caller via event_queue...Call DelegationClient from inside an agent executor. The user’s bearer token comes from the current request context, so token exchange happens on the caller’s behalf:
import { DelegationClient, getKeycardAuth, type AgentExecutor } from "@keycardai/a2a";
// config is the same AgentServiceConfig from the server setupconst client = new DelegationClient(config);
const executor: AgentExecutor = { async execute(requestContext, eventBus) { const auth = getKeycardAuth(requestContext); if (!auth) throw new Error("unauthenticated");
const result = await client.invokeService( "<remote-agent-url>", "Summarize the latest news on AI agents", { subjectToken: auth.token }, );
eventBus.publish(result.message); eventBus.finished(); }, async cancelTask() {},};