synadia-ai-agents

Python client SDK for the Synadia Agent Protocol for NATS. 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 (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 is tracked in 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.skips 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. Hosting an agent? → see synadia-ai-agent-service.

Installation

pip install synadia-ai-agents

Or from this checkout for local development:

uv pip install -e .

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

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 for more options. Synadia Cloud or any hosted NATS works too - see 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.

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

Sender identity

The sender-identity 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.

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:

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):

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 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; the definitions are in test-fixtures/agent-tools/. Nothing changes on the wire.

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/. They talk to the reference agent which now ships with synadia-ai-agent-service at agent-sdk/python/examples/_reference_agent.py. The ritual to see the SDKs work end-to-end:

# 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 for the full tour.

Connecting to NATS in production

For Synadia 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:

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:

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 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 — 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):

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
  • the wire contract (source of truth, lives in synadia-ai/synadia-agent-sdk-docs).
• docs/protocol-mapping.md - every SDK call mapped to its spec section; for auditors and other-SDK implementers.
• examples/README.md - tour of the runnable demos under examples/.
• CHANGELOG.md - release notes and migration guidance.
• CLAUDE.md - project context and engineering conventions.

Development

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.