Metadata-Version: 2.5
Name: synadia-ai-agents
Version: 0.8.1
Summary: Python client SDK for the Synadia Agent Protocol for NATS — discover and prompt spec-compliant agents over NATS
Project-URL: Homepage, https://github.com/synadia-ai/synadia-agents/tree/main/client-sdk/python
Project-URL: Repository, https://github.com/synadia-ai/synadia-agents
Project-URL: Documentation, https://github.com/synadia-ai/synadia-agents/blob/main/client-sdk/python/docs/protocol-mapping.md
Project-URL: Changelog, https://github.com/synadia-ai/synadia-agents/blob/main/client-sdk/python/CHANGELOG.md
Project-URL: Issues, https://github.com/synadia-ai/synadia-agents/issues
Author: Synadia
License: Apache-2.0
License-File: LICENSE
Keywords: agents,ai,nats,rpc,sdk,streaming
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Networking
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: nats-py>=2.7
Requires-Dist: nkeys>=0.2
Requires-Dist: pydantic>=2.5
Requires-Dist: pynacl>=1.5
Provides-Extra: examples
Requires-Dist: rich>=13; extra == 'examples'
Description-Content-Type: text/markdown

# synadia-ai-agents

Python **client** SDK for the [Synadia Agent Protocol for NATS](https://github.com/synadia-ai/synadia-agent-sdk-docs/blob/main/core-protocol.md).
Discover protocol-compliant agents over NATS and prompt them with
streamed typed responses.

**Hosting an agent (Hermes / claude-code / openclaw / pi)?** That side
of the protocol now ships separately as
[`synadia-ai-agent-service`](../../agent-sdk/python/) (import
`synadia_ai.agent_service`). It depends on this package for the
shared wire primitives — install both when authoring an agent
harness.

**Cross-SDK parity with the [TypeScript SDK](https://github.com/synadia-ai/synadia-agents/tree/main/client-sdk/typescript)**
is tracked in [`tests/test_interop_e2e.py`](tests/test_interop_e2e.py).
Both SDKs declare `protocol_version = "0.3"` in service metadata, so the
test spawns the TS reference agent via `bun` and rounds-trips a prompt
through it. The test `pytest.skip`s cleanly when `bun` or the sibling
`../typescript/` checkout is missing — running the suite without TS
interop is fine for day-to-day work.

**Calling agents?** → [Quickstart - call an agent](#quickstart--call-an-agent).
**Hosting an agent?** → see [`synadia-ai-agent-service`](../../agent-sdk/python/).

## Installation

```bash
pip install synadia-ai-agents
```

Or from this checkout for local development:

```bash
uv pip install -e .
```

You also need a reachable `nats-server`. Pick whichever fits:

```bash
brew install nats-server                          # macOS
# Linux / anywhere with Docker:
docker run --rm -p 4222:4222 nats:2.12-alpine
# Then:
nats-server -a 127.0.0.1 -p 4222
```

See the [nats.io install docs](https://docs.nats.io/running-a-nats-service/introduction/installation)
for more options. [Synadia Cloud](https://www.synadia.com/cloud/) or
any hosted NATS works too - see
[Connecting to NATS in production](#connecting-to-nats-in-production)
below.

## Quickstart - call an agent

The SDK doesn't open NATS connections — you build a
`nats.aio.client.Client` and hand it to `Agents`. That mirrors what
`Svcm(nc)`, `jetstream(nc)`, `Kvm(nc)` do, and lets one connection serve
JetStream, KV, services, and agents at once.

```python
import asyncio
import nats
from synadia_ai.agents import Agents, ResponseChunk, StatusChunk

async def main() -> None:
    nc = await nats.connect(servers="nats://127.0.0.1:4222")
    agents = Agents(nc=nc)
    try:
        found = await agents.discover()           # list[Agent], stall by default
        for a in found:
            print(f"{a.agent}/{a.owner}/{a.name} @ {a.prompt_subject}")

        # Each Agent is directly callable — no bind step.
        async for msg in found[0].prompt("hello"):
            if isinstance(msg, ResponseChunk):
                print(msg.text, end="")
            elif isinstance(msg, StatusChunk) and msg.status == "done":
                print()
    finally:
        await agents.close()                      # SDK state only
        await nc.close()                          # caller owns this

asyncio.run(main())
```

## API matrix

| Symbol | Lives in | Purpose |
| --- | --- | --- |
| `Agents` | [`agents.py`](src/synadia_ai/agents/agents.py) | Caller-side entry point. Construct with `nc=`; owns the heartbeat wildcard sub. |
| `Agent` | [`agent.py`](src/synadia_ai/agents/agent.py) | Live handle from `Agents.discover()`. `.prompt()` is the one method that does I/O. |
| `AgentInfo` | [`discovery.py`](src/synadia_ai/agents/discovery.py) | Pure-data record (parsed `$SRV.INFO` per §4.3). What `build_agent_info()` returns. |
| `Liveness` | [`heartbeat.py`](src/synadia_ai/agents/heartbeat.py) | Frozen snapshot from `Agents.liveness(instance_id)`. |
| `load_context_options` | [`context.py`](src/synadia_ai/agents/context.py) | Resolve a `nats` CLI context into kwargs for `nats.connect(...)`. |
| `resolve_nats_connection_bundle` | [`connection_bundle.py`](src/synadia_ai/agents/connection_bundle.py) | Read connection auth once and optionally derive a signer from that exact snapshot. |
| `Identity`, `signer_from_seed` / `signer_from_creds_file` / `signer_from_context` | [`identity/`](src/synadia_ai/agents/identity/) | Sender identity: sign every `prompt` / `status` with the connection's NKEY — see below. |
| `Agents.self_id()`, `Agents.sign_sender` / `publish_signed` / `request_signed`, `Agents.resolve_sender` | [`agents.py`](src/synadia_ai/agents/agents.py) | The connection's own agent ID; signed publishes for any subject; the reverse lookup. |
| `Agent.status()` | [`agent.py`](src/synadia_ai/agents/agent.py) | The §8.7 status probe (header attached) → `HeartbeatPayload`. |
| `save_attachments` | [`attachments.py`](src/synadia_ai/agents/attachments.py) | Save a reply's attachments to disk under safe, never-overwriting names; returns the absolute paths. |
| `AgentTools` | [`tools/`](src/synadia_ai/agents/tools/) | The agent tools a model calls to discover and prompt other agents — see below. |
| `AgentId`, `verify_sender_header`, `parse_sender_header`, `format_sender` | [`identity/`](src/synadia_ai/agents/identity/) | The shared identity codec (also used by the host package). |
| `AgentService` | [`synadia-ai-agent-service`](../../agent-sdk/python/) | Server-side; ships in a separate distribution. Import from `synadia_ai.agent_service`. |

## Sender identity

The [sender-identity extension](https://github.com/synadia-ai/synadia-agent-sdk-docs/blob/main/core-protocol.md#13-sender-identity-optional-extension)
lets a receiving agent know *who* prompted it, verified per message:
the caller attaches an `Agent-Sender` header that names its agent ID —
the `(account, user)` NKEY pair authenticated on that connection — and, with
a signer, an ed25519 signature bound to the subject, the payload, a
timestamp and a nonce. Nothing in it changes protocol `0.3`: an agent
that implements the extension says so with `min_sender_trust` on its
prompt endpoint (`agent.supports_sender_identity`). Identity is off when
the `identity` argument is omitted.

```python
import nats
from synadia_ai.agents import Agents, Identity, resolve_nats_connection_bundle

bundle = resolve_nats_connection_bundle(
    url="tls://connect.ngs.global",
    creds="~/.config/nats/user.creds",
    identity="signed",
)
nc = None
try:
    nc = await nats.connect(**bundle.connection_options)
    agents = Agents(
        nc=nc,
        identity=Identity(signer=bundle.signer, name="claude-code"),
    )
    try:
        print(await agents.self_id())
    finally:
        await agents.close()
finally:
    if nc is not None:
        await nc.close()
    bundle.wipe()               # after NATS closes; reconnect needs the snapshot
```

What to know:

- **Identity is opt-in.** Omit `identity` for no lookup and no header.
  Pass `Identity()` explicitly for an unsigned claim, or use
  `Identity(send_unsigned_claim=False)` for no automatic identity work.
  An unsigned claim discloses your user NKEY to the receiver.
- **Use the connection's credentials.** Prefer
  `resolve_nats_connection_bundle(context=..., identity="signed")`, or its
  `url=...` plus `creds=...` / `nkey=...` form. It reads the selected context
  and credential file once, builds reconnect-safe NATS options, and derives
  the signer from that exact snapshot. There is no separate identity
  credential path. The older `load_context_options` and `signer_from_*`
  helpers remain available for compatibility and advanced use, but do not
  independently read the same mutable file for a new signed connection.
  Before every signed send, the SDK compares the signer's user and account
  with live `$SYS.REQ.USER.INFO`; a mismatch or unavailable binding fails
  and never downgrades to unsigned or headerless delivery. An HSM / KMS
  signer can implement `SenderSigner` (`sign` may be async).
- **One bundle belongs to one connection.** Multiple `Agents` controllers
  or sessions sharing that connection may reuse `bundle.signer`; it names
  the NATS user, not an individual chat session. Close every controller,
  close NATS, then call the idempotent `bundle.wipe()`. Do not wipe while
  reconnect is possible. `bundle.connection_options` necessarily contains
  authentication configuration: never log or serialize it. The bundle's
  own `repr` / `str` are redacted.
- An endpoint that
  declares `min_sender_trust: signed` fails early with
  `SenderSignatureRequiredError` at call time when no signer is
  configured. The error exposes stable `code` (`401`), `description`
  (`"signature required"`), and `subject` attributes, so callers do not
  need to parse its message. With a signer, an unavailable identity raises
  the `self_id()` error on the first iteration.
- **Cost.** Each identity-bearing request performs a live
  `$SYS.REQ.USER.INFO` lookup (2 s timeout at most; a permission violation
  fails at once), because nats-py exposes no reconnect generation that
  could safely invalidate a cached identity. Explicit diagnostic calls to
  `self_id()` are memoised. A signed header is ~400 bytes and counts against
  `max_payload` (header framing included — `PayloadTooLargeError.header_bytes`).
  nats-py does **not** count headers in its own check, so the SDK's is
  the only guard before the server closes the connection with
  `Maximum Payload Violation`.
- **Behind a service import that remaps the subject** — an export that
  inserts the caller's account token (`account_token_position`), or a
  `to:` / `local_subject` rename by your own account — discovery reports
  the exporter's subject, which you cannot publish to. Pass
  `prompt(text, subject=…)` / `status(subject=…)` with the local name;
  the receiver strips an inserted token by itself. Only for a rename by
  **your own** account also pass `sub=agent.prompt_endpoint.subject`
  (sign the exporter's subject). `sign_sender` / `publish_signed` /
  `request_signed` take the same `sub=` keyword.
- **A trusted server over TLS is a precondition.** The NATS handshake
  signs a server-chosen nonce with the same seed that signs
  `Agent-Sender`; a server you should not have trusted could obtain a
  signature valid for 30 s. Identity is meaningful only over TLS to a
  server whose certificate you verify.
- The verified identity is the **user** key; `account` is the sender's
  signed claim (`format_sender` / `str(sender)` renders
  `… (verified user, claimed account)`). Which verified senders a
  receiver accepts is the receiver's business — see the host package's
  `accept_sender` (`synadia-ai-agent-service` 0.5.0, whose `AgentService`
  classifies every prompt before the ack and hands the handler
  `stream.sender`).
- **Only the request is signed.** Prompt response chunks and mid-stream
  query replies are not independently authenticated; do not infer the
  actor answering a query from the original prompt sender.
- `scripts/whoami.py` prints what `self_id()` resolves for a connection
  (or why it resolves nothing).

## Prompt interceptors

An extension that needs to see or add to every prompt — extra envelope
fields, extra headers, a signed message of its own about the prompt —
plugs in as a `PromptInterceptor`, in two phases:

```python
import json
from synadia_ai.agents import Agents, PromptExtras, PromptInterceptorContext

class Tagging:
    # Phase one: what the prompt carries. No side effects.
    async def before_prompt(self, ctx: PromptInterceptorContext) -> PromptExtras | None:
        # ctx.agent, ctx.prompt, ctx.envelope_extras, ctx.context (prompt(context=...)),
        # ctx.connection
        request_id = str(ctx.context.get("request_id", "none"))
        return PromptExtras(
            fields={"x_request": request_id},
            headers={"X-Request": request_id},
            state=request_id,
        )

    # Phase two (optional): the prompt is signed and checked, and goes out right after.
    async def before_publish(
        self, ctx: PromptInterceptorContext, extras: PromptExtras | None
    ) -> None:
        if ctx.identity.can_sign and extras is not None:
            body = json.dumps({"request": extras.state, "to": ctx.agent.instance_id})
            await ctx.identity.publish_signed("audit.prompts", body)

agents = Agents(nc=nc, identity=identity, interceptors=[Tagging()])
async for msg in agent.prompt("hi", context={"request_id": "r-1"}):
    ...
```

- Both phases run at publish time — on the stream's first `__anext__` —
  in one copy of the `contextvars` context `prompt()` was called in, and
  get the same `ctx`. A prompt that is never iterated, or that `prompt()`
  itself rejects, runs neither.
- `before_prompt` runs after the sender identity is resolved and has no
  side effects: the prompt can still fail after it (the size of the
  envelope its fields make, its identity). An exception fails the prompt
  before anything is sent. Several interceptors are merged in order, the
  later winning a key.
- `before_publish` (optional; `PublishingPromptInterceptor` types it) runs
  after the `Agent-Sender` header is signed and the size checked,
  immediately before the prompt is published, with what the same
  interceptor's `before_prompt` returned (`state` included). Messages
  published there describe a prompt that goes out, barring a transport
  failure. An exception is logged and does not stop the prompt.
- `fields` are written as top-level envelope fields next to the
  protocol's (§5.6 obliges receivers to tolerate them); a host reads them
  back from `Envelope.extras`. A field the envelope defines and the
  `Agent-Sender` header are refused.
- An `Envelope` passed to `prompt()` goes out with its own extra fields,
  so a relay that forwards the envelope it received preserves them (§5.6).
  Interceptors see them as `ctx.envelope_extras`, a read-only copy (empty
  for a text prompt); a field an interceptor adds replaces one of the same
  name.
- `ctx.identity.publish_signed(subject, payload, nonce=...)` signs with the
  prompting client's identity; pass `nonce` when the body carries its own
  id, and it is the header's nonce and the `Nats-Msg-Id` too (also on
  `Agents.publish_signed`).
- Without interceptors nothing changes on the wire.

## Mid-stream queries

Agent handlers can pause their response stream to ask the caller a
question (permission prompt, clarification, menu selection):

```python
async for msg in agent.prompt("do the thing"):
    if isinstance(msg, Query):
        await msg.reply("yes")
    else:
        print(msg)     # ResponseChunk / StatusChunk
```

Server-side, the handler asks via `stream.ask(...)` — see
[`synadia-ai-agent-service`](../../agent-sdk/python/) for the host-side
API.

## Agent tools

`AgentTools` gives a model six tools to discover and prompt other agents:
`discover_agents`, `prompt_agent`, `wait_agent`, `answer_agent`,
`cancel_agent` and `list_agent_calls`. The contract — parameters, results,
states, rules and limits — is [`docs/agent-tools.md`](../../docs/agent-tools.md);
the definitions are in [`test-fixtures/agent-tools/`](../../test-fixtures/agent-tools/).
Nothing changes on the wire.

```python
import json

from synadia_ai.agents import Agents, AgentTools, Identity
from synadia_ai.agent_service import AgentService

tools = AgentTools(Agents(nc=nc, identity=Identity(signer=signer)))

# On a host, a call belongs to the prompt being served: it ends with it.
service = AgentService(
    nc=nc, agent="researcher", owner="acme", session_name="r1",
    interceptors=[tools.request_interceptor],
)

# Show the model the tools, in your model API's format ...
model_tools = [
    {"type": "function", "function": {"name": d["name"], "description": d["description"],
                                      "parameters": d["parameters"]}}
    for d in tools.definitions
]

# ... and run each tool call it makes; the result goes back as JSON text.
result = await tools.execute(call.name, call.arguments, tool_call_id=call.id)
content = json.dumps(result)
```

- `prompt_agent` waits for the reply by default; `wait: false` returns a
  `call_id` at once and the model collects the result with `wait_agent`. A
  question the prompted agent asks goes to the model, which answers it
  with `answer_agent`.
- `tools=` offers fewer than the six. Each definition costs input tokens
  on every model call: as an estimate that varies with the model and its
  tokenizer, roughly 2k for all six and a little over 1k for three. So an
  agent that needs no async calls offers three:
  `AgentTools(agents, tools=BLOCKING_AGENT_TOOLS)`, which are
  `discover_agents`, `prompt_agent` and `answer_agent`. `AGENT_TOOL_NAMES`
  names all six. Without `wait_agent` nothing is detached: `prompt_agent`
  and `answer_agent` lose their `wait` parameter, and their descriptions
  read blocking-only words.
- A call started while a prompt is served belongs to it: still open when
  that prompt ends, it is cancelled and its open question refused. A host
  that serves prompts without `AgentService` wraps each in
  `async with tools.prompt_scope(caller=...)`. Outside a served prompt,
  `on_settled` reports each call that finishes.
- Limits are configuration: `max_wait_s` (10 minutes), `max_wait_agent_s`,
  `max_calls` (256), `attachment_roots`, `staging_dir`,
  `max_saved_bytes_per_call`. `self_address` is left out of discovery and
  refused.
- Files are sent only from the staging directory, where returned files are
  saved, so the model can send one on, and from under `attachment_roots`,
  which add to it. The default names none, so nothing else can be sent: the
  working directory may hold a `.env`. A host that wants the working
  directory, a coding agent's project say, names it. A relative root, or
  `staging_dir`, is taken from the working directory when the helper is
  made.
- The model's tool-call ID reaches every prompt interceptor as
  `ctx.context["tool_call_id"]`. `extensions` (subclasses of
  `AgentToolsExtension`) add discovery fields, rewrite a prompt before it
  is sent, and look at a reply. Cancelling the task that runs a blocking
  call cancels the call.
- `await tools.aclose()` (or `async with AgentTools(...)`) cancels open
  calls and removes the staging directory it created.

## Try the examples

Six runnable client-side demos live under
[`examples/`](examples/README.md). They talk to the reference agent
which now ships with `synadia-ai-agent-service` at
[`agent-sdk/python/examples/_reference_agent.py`](../../agent-sdk/python/examples/_reference_agent.py).
The ritual to see the SDKs work end-to-end:

```shell
# terminal 1 — start the reference agent (from the agent-sdk dist)
uv run --directory ../../agent-sdk/python python examples/_reference_agent.py \
  --url nats://127.0.0.1:4222

# terminal 2 — discover and prompt (from this dist)
uv run python examples/01-discover.py --url nats://127.0.0.1:4222
uv run python examples/02-prompt-text.py --url nats://127.0.0.1:4222 "hello"
```

See [`examples/README.md`](examples/README.md) for the full tour.

## Connecting to NATS in production

For [Synadia Cloud](https://www.synadia.com/cloud/) or any self-hosted
NATS that needs credentials, JWTs, or a non-default URL, resolve a `nats`
CLI context once. Identity is off by default, so ordinary connections do
not expose a signer:

```python
import nats
from synadia_ai.agents import Agents, resolve_nats_connection_bundle

bundle = resolve_nats_connection_bundle(context="prod")
nc = None
try:
    nc = await nats.connect(**bundle.connection_options)
    agents = Agents(nc=nc)
    try:
        ...
    finally:
        await agents.close()
finally:
    if nc is not None:
        await nc.close()
    bundle.wipe()
```

Pass `identity="signed"` and configure `Identity(signer=bundle.signer)`
when outgoing requests should carry signed sender identity. Signed mode
fails clearly for anonymous, token, user/password, or JWT-without-seed
authentication; it never silently sends unsigned requests. URL mode is
also available:

```python
bundle = resolve_nats_connection_bundle(
    url="tls://connect.example.com",
    creds="~/user.creds",  # or: nkey="~/user.nk"
    identity="signed",
)
```

`load_context_options(...)` remains available as the compatibility helper
when a connection-only dict is all you need. It reads
`~/.config/nats/context/<name>.json` — URL, creds file, nkey seed
file, token, user/password, inbox prefix are all honored. See
[`CLAUDE.md`](CLAUDE.md#connecting-to-nats) for the full field-by-field
table (including which NATS-context fields are not yet supported and
fail fast rather than silently).

## Hosting an agent

The agent-host surface (`AgentService`, `PromptStream`,
`PromptHandler`, the heartbeat publisher) ships separately as
[`synadia-ai-agent-service`](../../agent-sdk/python/) — install that
package alongside this one when authoring an agent harness, and
import the host classes from `synadia_ai.agent_service`. The shared
wire types (`Envelope`, `Attachment`, error classes,
`HeartbeatPayload`, `AgentSubject`, the discovery constants) stay in
this package and continue to import from `synadia_ai.agents`.

Probe a running agent with the `nats` CLI (subjects are verb-first
per protocol v0.3):

```bash
nats micro list                                          # see "agents"
nats req  agents.prompt.demo.alice.worker-1 "hello" \
  --replies=0 --reply-timeout=30s --timeout=60s          # prompt it (see docs/using-nats-cli.md)
nats req  agents.status.demo.alice.worker-1 ""           # heartbeat-shaped status reply
nats sub  "agents.hb.demo.alice.worker-1"                # watch heartbeats
```

## Documentation

- [Synadia Agent Protocol for NATS spec](https://github.com/synadia-ai/synadia-agent-sdk-docs/blob/main/core-protocol.md)
  - the wire contract (source of truth, lives in
  [`synadia-ai/synadia-agent-sdk-docs`](https://github.com/synadia-ai/synadia-agent-sdk-docs)).
- [`docs/protocol-mapping.md`](docs/protocol-mapping.md) - every SDK call
  mapped to its spec section; for auditors and other-SDK implementers.
- [`examples/README.md`](examples/README.md) - tour of the runnable
  demos under `examples/`.
- [`CHANGELOG.md`](CHANGELOG.md) - release notes and migration guidance.
- [`CLAUDE.md`](CLAUDE.md) - project context and engineering conventions.

## Development

```bash
uv sync                              # install
uv run ruff check . && uv run ruff format --check . && uv run mypy src tests examples && uv run pytest
```

Integration tests spawn a real `nats-server` per session and record wire
evidence under `tests/_evidence/<test-nodeid>/`. Cross-SDK interop tests
(`tests/test_interop_e2e.py`) additionally spawn the TypeScript
reference agent via `bun`; they skip cleanly if `bun` or the sibling
`../typescript/` checkout isn't present.

## License

Apache-2.0. See [LICENSE](LICENSE).
