a2a-agent-sdk

A framework-agnostic SDK for building A2A-compliant agent runners that work with kagent's BYO (bring-your-own) agent pattern. Plug in any agent implementation — LangGraph, a plain function, a different framework entirely — and get a fully working A2A service for free: agent-card discovery, message/send, message/stream, health, and API-key auth.

The point of this SDK: your agent implementation can change completely, but the input your callers send and the output they get back never do.

Install

pip install -e /path/to/a2a-agent-sdk

(Not yet published to a package index — install from a local checkout or a git URL.)

Quickstart

Implement one method, invoke, that takes the conversation so far and returns text:

from a2a_agent_sdk import A2ARunner, RunnerConfig, Message

class MyAgent:
    def invoke(self, messages: list[Message]) -> str:
        last_user_text = messages[-1].content
        return f"You said: {last_user_text}"

runner = A2ARunner(
    agent=MyAgent(),
    config=RunnerConfig(name="My Agent", description="Echoes the user."),
)
app = runner.app  # a plain FastAPI instance

Run it:

uvicorn app:app --host 0.0.0.0 --port 8080

Try it:

curl -X POST http://localhost:8080/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":"1","method":"message/send","params":{"message":{"parts":[{"kind":"text","text":"hello"}]}}}'

See examples/function_agent/ for the complete runnable version of this example.

Adding real streaming

If your agent can produce text incrementally, implement astream too and message/stream will emit it token-by-token instead of as one chunk:

from typing import AsyncIterator

class MyStreamingAgent:
    def invoke(self, messages: list[Message]) -> str:
        return "hello there friend "

    async def astream(self, messages: list[Message]) -> AsyncIterator[str]:
        for word in "hello there friend".split():
            yield word + " "

If you don't implement astream, message/stream still works — the SDK calls invoke() once and emits the whole result as a single chunk. Callers never need to know which case they're in.

Any agent that implements astream (in addition to invoke) satisfies the StreamingAgent protocol, importable from a2a_agent_sdk. It's not required anywhere in the SDK's own code, but it's useful for type-annotating a variable that should support real streaming, or for an isinstance(agent, StreamingAgent) check in your own code.

Plugging in LangGraph

Nothing about the SDK is LangGraph-specific — it just needs an object with invoke/astream. See examples/langgraph_agent/ for a complete example that bridges a compiled StateGraph to the Agent protocol:

class LangGraphAgent:
    def __init__(self, graph):
        self._graph = graph

    def invoke(self, messages: list[Message]) -> str:
        result = self._graph.invoke({"messages": [(m.role, m.content) for m in messages]})
        return result["messages"][-1].content

    async def astream(self, messages: list[Message]):
        graph_input = {"messages": [(m.role, m.content) for m in messages]}
        async for chunk, _meta in self._graph.astream(graph_input, stream_mode="messages"):
            if chunk.content:
                yield chunk.content

Swap LangGraphAgent for a class wrapping any other framework and the rest of your deployment — endpoints, request/response shapes, auth, agent card — doesn't change at all.

RunnerConfig reference

Field	Env var (from_env())	Purpose
id	AGENT_ID	Internal identifier; used as an Agent Card name fallback
name	AGENT_NAME	Agent Card name
description	AGENT_DESCRIPTION	Agent Card description
url	AGENT_URL	Agent Card url — the endpoint callers should hit
api_key	RUNNER_API_KEY	If set, invoke requires a matching key (see Auth below)
skills	— (construct SkillDescriptor list directly)	Skills advertised on the Agent Card

RunnerConfig deliberately does not include model/temperature/prompt/skill content — those belong inside your own Agent implementation, since the SDK doesn't assume you're even calling an LLM.

config = RunnerConfig.from_env()          # read AGENT_*/RUNNER_API_KEY from the environment
config = RunnerConfig(name="My Agent")    # or construct directly

Advertising skills

Pass a list of SkillDescriptors to advertise what your agent can do:

from a2a_agent_sdk import RunnerConfig, SkillDescriptor

config = RunnerConfig(
    name="My Agent",
    skills=[SkillDescriptor(name="Refund Lookup", description="Looks up refund status", tags=["billing"])],
)

These show up in the Agent Card's skills array so A2A clients can discover what your agent is capable of.

Endpoints

Route	Behavior
GET /health	{status, ready, agent_id, agent_name}
GET /.well-known/agent-card.json, GET /.well-known/agent.json	A2A Agent Card
POST /, POST /a2a/message	message/stream → SSE; anything else → JSON-RPC task result

Both POST routes accept the same payload shapes: JSON-RPC message/send/ message/stream, a direct {"message": {...}} body, or a bare {"parts": [...]}/{"text": "..."} shortcut for local testing.

Auth

If RunnerConfig.api_key is set, every POST / / POST /a2a/message request must present a matching key, either as:

• the X-API-Key header, or
• params.metadata.runner_api_key in the JSON-RPC payload (for proxies that don't forward custom headers, e.g. kagent's A2A proxy)

If api_key is left unset, the check is skipped entirely — useful for local development. There's no third state: either a key is configured and enforced, or it isn't configured and nothing is enforced.

Errors

Status	When
400	No user text found in the request
401	api_key is configured and the request's key is missing/invalid
502	agent.invoke/agent.astream raised an exception (streaming: emitted as a failed status event on the open connection instead)

Deployment

The SDK produces a plain FastAPI app (runner.app) — deploy it with any ASGI server. For parity with kagent's BYO agent expectation, serve on port 8080:

uvicorn app:app --host 0.0.0.0 --port 8080

Examples

• examples/function_agent/ — the smallest possible agent: one method, no framework.
• examples/langgraph_agent/ — the same SDK, with a LangGraph graph underneath.

Install example dependencies and run their tests with:

pip install -e ".[dev,examples]"
pytest examples/function_agent/test_app.py -v
pytest examples/langgraph_agent/test_app.py -v

Run each example's tests as a separate pytest invocation, not combined — both use bare (non-packaged) app/agent/test_app modules so they can be run exactly like an external consumer would (uvicorn app:app), which means pytest can't collect both test_app.py files in the same session.