Documentation Index Fetch the complete documentation index at: https://afk.arpan.sh/llms.txt
Use this file to discover all available pages before exploring further.
The A2A protocol enables agent communication across system boundaries — between services, organizations, or deployment environments. It builds on internal messaging by adding authentication, authorization, and external transport.
Architecture Three integration layers Layer What it does When you need it Internal Protocol Typed envelopes with idempotency Always (agents in the same system) Auth Provider Token validation, caller identity When agents are in different services External Adapter HTTP/gRPC transport, discovery When agents are in different systems
Request flow
Client builds invocation request
from afk.messaging import AgentInvocationRequest request = AgentInvocationRequest( run_id = "run-42" , thread_id = "thread-42" , conversation_id = "conversation-42" , correlation_id = "analysis-42" , source_agent = "system-a" , target_agent = "analyzer" , payload = { "user_message" : "Analyze this dataset" }, metadata = { "source" : "system-a" }, idempotency_key = "analysis-42" , )
Auth provider validates
The server validates the auth token, extracts the caller identity, and checks authorization rules.
Server runs the agent
The target agent executes locally with a Runner, using the invocation request as input.
Response returned
response = await client.invoke(request) print (response.output) # Agent output payload print (response.success) # True when handled successfully print (response.run_id) # For tracing
Invocation contracts class AgentInvocationRequest ( BaseModel ): run_id: str thread_id: str conversation_id: str correlation_id: str idempotency_key: str source_agent: str target_agent: str payload: dict = {} metadata: dict = {} timeout_s: float | None = None
class AgentInvocationResponse ( BaseModel ): run_id: str thread_id: str conversation_id: str correlation_id: str idempotency_key: str source_agent: str target_agent: str success: bool output: dict | str | None = None error: str | None = None
Hosting an A2A service Expose your agents as an A2A-accessible service:
from afk.agents import Agent, A2AServiceHost, APIKeyA2AAuthProvider from afk.core import Runner # Define the agent analyzer = Agent( name = "analyzer" , model = "gpt-4.1-mini" , instructions = "Analyze data." ) # Create auth provider auth = APIKeyA2AAuthProvider( keys = { "system-a" : "token-abc" , "system-c" : "token-xyz" }, server_secret = "hmac-secret-for-key-hashing" , ) # Start the server server = A2AServiceHost( agents = { "analyzer" : analyzer}, runner_factory = lambda : Runner(), auth_provider = auth, host = "0.0.0.0" , port = 8080 , ) await server.start()
Authentication providers AFK ships with three auth providers:
AllowAll (dev only)
API Key
JWT
Permits all requests without authentication. Never use in production. from afk.agents import AllowAllA2AAuthProvider auth = AllowAllA2AAuthProvider()
Validates requests against pre-shared API keys with HMAC-SHA256 hashing. from afk.agents import APIKeyA2AAuthProvider auth = APIKeyA2AAuthProvider( keys = { "caller-id" : "secret-key" }, server_secret = "hmac-server-secret" , )
Validates JWT tokens with configurable issuer and audience claims. from afk.agents import JWTA2AAuthProvider auth = JWTA2AAuthProvider( secret = "jwt-signing-secret" , algorithm = "HS256" , issuer = "https://auth.example.com" , audience = "agent-service" , )
Google A2A adapter For interoperability with Google’s A2A protocol, use the Google adapter:
from afk.agents import GoogleA2AProtocolAdapter adapter = GoogleA2AProtocolAdapter( client = google_a2a_client)
The adapter wraps a configured Google A2A SDK client behind AFK’s AgentCommunicationProtocol.
Security considerations Concern Mechanism Authentication Token-based (API keys, JWT, OAuth) Authorization Per-agent access control (which callers can invoke which agents) Idempotency idempotency_key prevents duplicate processing on retriesRate limiting Configure per-caller request limits Input validation All requests validated against AgentInvocationRequest schema Cost isolation Each invocation has its own FailSafeConfig budget
Always authenticate A2A endpoints. An unauthenticated A2A server allows
anyone to invoke your agents, consuming your LLM API credits.
Next steps
Task Queues Async job processing for long-running work.
MCP Server Expose tools via the Model Context Protocol.
