Metadata-Version: 2.5
Name: a2a-agent-sdk
Version: 0.1.0
Summary: Framework-agnostic SDK for building A2A/kagent-compliant agent runners.
Requires-Python: >=3.10
Requires-Dist: fastapi>=0.115.0
Requires-Dist: starlette>=0.38.0
Provides-Extra: dev
Requires-Dist: anyio>=4.0.0; extra == 'dev'
Requires-Dist: httpx>=0.27.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Provides-Extra: examples
Requires-Dist: langchain-core>=0.3.0; extra == 'examples'
Requires-Dist: langgraph>=0.2.0; extra == 'examples'
Description-Content-Type: text/markdown

# a2a-agent-sdk

A framework-agnostic SDK for building [A2A](https://github.com/a2aproject/A2A)-compliant
agent runners that work with [kagent](https://kagent.dev)'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

```bash
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:

```python
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:

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

Try it:

```bash
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:

```python
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:

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

```python
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 `SkillDescriptor`s to advertise what your agent can do:

```python
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:

```bash
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:

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