Skip to content
API Reference

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.

  • 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
Terminal window
pip install keycardai-a2a

Pulls in keycardai-oauth, keycardai-starlette, and a2a-sdk[http-server] >= 1.0.

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

ExportDescription
AgentServiceConfigService identity, Keycard credentials, and agent card metadata
build_agent_card_from_configConstruct an a2a.types.AgentCard from an AgentServiceConfig
KeycardServerCallContextBuilderPropagate the verified bearer token onto ServerCallContext.state so executors can use it for downstream token exchange
DelegationClientAsync client for remote agent invocation with Keycard token exchange
DelegationClientSyncSynchronous variant of DelegationClient
ServiceDiscoveryResolve 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.

Compose the Keycard helpers with a2a-sdk’s route factories and Starlette’s AuthenticationMiddleware:

from a2a.server.agent_execution import AgentExecutor
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.routing import Mount
from keycardai.a2a import (
AgentServiceConfig,
KeycardServerCallContextBuilder,
build_agent_card_from_config,
)
from keycardai.oauth.server.credentials import ClientSecret
from 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.

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...