Getting Started¶
Installation¶
Your First Agent¶
Create a file called my_agent.py:
from a2akit import A2AServer, AgentCardConfig, TaskContext, Worker
class EchoWorker(Worker): # (1)!
async def handle(self, ctx: TaskContext) -> None: # (2)!
await ctx.complete(f"Echo: {ctx.user_text}") # (3)!
server = A2AServer(
worker=EchoWorker(),
agent_card=AgentCardConfig( # (4)!
name="Echo Agent",
description="Echoes your input back.",
version="0.1.0",
),
)
app = server.as_fastapi_app() # (5)!
- Subclass
Workerto define your agent's behavior. - Implement the single required method:
handle(ctx). - Call a lifecycle method to complete the task with a text artifact.
- Configure the agent discovery card (served at
/.well-known/agent-card.json). - Get a fully configured FastAPI app — storage, broker, event bus, and all endpoints wired automatically.
Start the Server¶
Your agent is now running at http://localhost:8000 with all A2A endpoints active.
Test It¶
curl -X POST http://localhost:8000/v1/message:send \
-H "Content-Type: application/json" \
-d '{"message":{"role":"user","parts":[{"text":"hello"}],"messageId":"1"}}'
Expected response:
{
"id": "...",
"contextId": "...",
"status": {
"state": "completed",
"timestamp": "..."
},
"artifacts": [
{
"artifactId": "final-answer",
"parts": [{"text": "Echo: hello"}]
}
],
"history": [...]
}
Discover the Agent¶
This returns the agent's discovery card with capabilities, skills, and supported transport protocols.
What's Next?¶
- Streaming — emit artifacts word by word via SSE
- Middleware — extract secrets, inject headers, sanitize payloads
- Dependency Injection — share DB pools, HTTP clients, and config
- Architecture — understand how the components fit together