Agntcy Reference Application SDK

Build interoperable multi-agent systems for the Internet of Agents

Overview

The reference app-sdk provides a reference integration layer used by CoffeeAGNTCY to demonstrate how multi-agent applications can connect AGNTCY components and selected open source project as needed. It includes factory interfaces, abstractions, and examples that help developers understand interoperability patterns across AGNTCY transports, protocols, and directories.

It is intended as an integration and learning surface primarily used in coffeeAgntcy, not as a canonical or prescriptive SDK for AGNTCY subsystems. Standards and reference guidance remain the responsibility of the working groups. Please see the respective components for their latest SDK or interfaces.

Features

📦 Installation

# Install via pip
pip install agntcy-app-sdk

# Or for uv based projects
uv add agntcy-app-sdk

# Install from source
git clone https://github.com/agntcy/app-sdk.git
pip install -e app-sdk

Quick Start

The Application SDK provides a single factory interface that abstracts over multiple semantic protocols and transports, so you can switch between them without rewriting your agent logic. Full usage guides: A2A · MCP / FastMCP

SlimRPC is the native A2A-over-SLIM RPC transport — simplest setup for 1:1 request/response. SLIM patterns and NATS patterns are experimental transports that unlock pub/sub fan-out and moderated group chat via the same AgntcyFactory interface. The factory negotiates the best transport automatically from the agent card.

Running examples: Save any snippet as a .py file and run with uv run python my_script.py. The SDK is async — wrap top-level await calls in asyncio.run(main()).

1. Serve an A2A Agent

Declare your agent's transports in the card and serve over all of them with a single call:

from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from agntcy_app_sdk.factory import AgntcyFactory
from agntcy_app_sdk.semantic.a2a.server.card_bootstrap import InterfaceTransport

# Declare transports in the card — this is the single source of truth.
# The order of `supported_interfaces` is the server's preference order:
# the first entry is the preferred transport.
agent_card = AgentCard(
    name="My Agent", version="1.0.0",
    default_input_modes=["text"], default_output_modes=["text"],
    capabilities=AgentCapabilities(streaming=True), skills=[...],
    supported_interfaces=[
        AgentInterface(protocol_binding=InterfaceTransport.SLIM_PATTERNS,
                       url="slim://localhost:46357/default/default/My_Agent_slimpatterns"),
        AgentInterface(protocol_binding=InterfaceTransport.SLIM_RPC,
                       url="slim://localhost:46357/default/default/My_Agent_rpc"),
        AgentInterface(protocol_binding=InterfaceTransport.NATS_PATTERNS,
                       url="nats://localhost:4222/default/default/My_Agent_natspatterns"),
    ],
    description="My agent description",
)

# One call does it all — add_a2a_card() reads the card's interfaces.
# a2a-sdk 1.x requires the card at handler construction.
factory = AgntcyFactory()
session = factory.create_app_session(max_sessions=10)
await (
    session.add_a2a_card(agent_card, DefaultRequestHandler(
        agent_executor=MyAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=agent_card))
    .start(keep_alive=True)
)

from agntcy_app_sdk.factory import AgntcyFactory
from agntcy_app_sdk.semantic.a2a.server.srpc import A2ASlimRpcServerConfig, SlimRpcConnectionConfig
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore

config = A2ASlimRpcServerConfig(
    agent_card=agent_card,
    request_handler=DefaultRequestHandler(
        agent_executor=MyAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=agent_card,
    ),
    connection=SlimRpcConnectionConfig(
        identity="default/default/my_agent",
        # Hardcoded on purpose: a demo value, not a real secret.
        shared_secret="my-shared-secret-at-least-32-characters-long",
        endpoint="http://localhost:46357",
    ),
)

factory = AgntcyFactory()
session = factory.create_app_session(max_sessions=1)
session.add(config).with_session_id("default").build()
await session.start_all_sessions(keep_alive=True)

📖 More serving options →

2. Connect to an Agent

Configure all transports once, then let the factory negotiate the best one from the agent card:

from a2a.helpers import get_stream_response_text, new_text_message
from a2a.types import Role, SendMessageRequest
from agntcy_app_sdk.semantic.a2a.client.factory import A2AClientFactory
from agntcy_app_sdk.semantic.a2a.client.config import (
    ClientConfig, SlimRpcConfig, SlimTransportConfig, NatsTransportConfig,
)

# Multi-transport config — factory picks the right one from the agent card
config = ClientConfig(
    slimrpc_config=SlimRpcConfig(
        namespace="default",
        group="default",
        name="my_client",
        slim_url="http://localhost:46357",
        # Hardcoded on purpose: a demo value, not a real secret.
        secret="my-shared-secret-at-least-32-characters-long",
    ),
    slim_config=SlimTransportConfig(
        endpoint="http://localhost:46357",
        name="default/default/my_client",
        # Hardcoded on purpose: a demo value, not a real secret.
        shared_secret_identity="my-shared-secret-at-least-32-characters-long",
    ),
    nats_config=NatsTransportConfig(
        endpoint="nats://localhost:4222",
    ),
)

# Create the factory — supports slimrpc, slimpatterns, natspatterns, and JSONRPC
a2a = A2AClientFactory(config)

# Create a client — factory negotiates over the card's supported_interfaces
# (server order wins unless ClientConfig(use_client_preference=True))
client = await a2a.create(agent_card)

# Send a message and stream the response. Events are `StreamResponse`
# objects carrying a message, task, or status/artifact update.
request = SendMessageRequest(
    message=new_text_message("Hello, agent!", role=Role.ROLE_USER)
)
async for event in client.send_message(request):
    text = get_stream_response_text(event)
    if text:
        print(text)

from a2a.client import minimal_agent_card
from a2a.helpers import get_stream_response_text, new_text_message
from a2a.types import Role, SendMessageRequest
from slima2a import setup_slim_client
from slima2a.client_transport import slimrpc_channel_factory

from agntcy_app_sdk.factory import AgntcyFactory
from agntcy_app_sdk.semantic.a2a.client.config import ClientConfig

# Set up the SLIM connection
service, slim_app, local_name, conn_id = await setup_slim_client(
    namespace="default", group="default", name="my_client",
    slim_url="http://localhost:46357",
    # Hardcoded on purpose: a demo value, not a real secret. Must match the
    # server's shared_secret.
    secret="my-shared-secret-at-least-32-characters-long",
)

# Enable slimrpc on the SDK's ClientConfig and let the factory negotiate
config = ClientConfig(
    slimrpc_channel_factory=slimrpc_channel_factory(slim_app, conn_id),
)
# With IDENTITY_AUTH_ENABLED, every slimrpc call carries this agent's
# identity token, whatever the card declares: only call peers you trust.
card = minimal_agent_card("default/default/my_agent", ["slimrpc"])
client = await AgntcyFactory().a2a(config).create(card)

request = SendMessageRequest(
    message=new_text_message("Hello, agent!", role=Role.ROLE_USER)
)
async for event in client.send_message(request):
    print(get_stream_response_text(event))

📖 Complete A2A guide →

3. Broadcast to Many Agents

Fan out a single request to N agents simultaneously. Broadcast uses the SLIM patterns or NATS patterns transport (not SlimRPC) — the factory returns an A2AExperimentalClient with broadcast_message() when transport negotiation selects slimpatterns or natspatterns.

from a2a.helpers import new_text_message
from a2a.types import (
    AgentCapabilities, AgentCard, AgentInterface, Role, SendMessageRequest,
)

from agntcy_app_sdk.factory import AgntcyFactory
from agntcy_app_sdk.semantic.a2a import A2AExperimentalClient
from agntcy_app_sdk.semantic.a2a.client.config import ClientConfig, SlimTransportConfig
from agntcy_app_sdk.semantic.a2a.server.card_bootstrap import InterfaceTransport

factory = AgntcyFactory()

# SLIM patterns transport — creates an A2AExperimentalClient
# (swap SlimTransportConfig for NatsTransportConfig + nats:// scheme for NATS)
config = ClientConfig(
    slim_config=SlimTransportConfig(
        endpoint="http://localhost:46357",
        name="default/default/broadcaster",
    ),
)
card = AgentCard(
    name="default/default/agent1",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(),
    skills=[],
    supported_interfaces=[
        AgentInterface(
            protocol_binding=InterfaceTransport.SLIM_PATTERNS,
            url="slim://default/default/agent1",
        ),
    ],
    description="Agent 1",
)

client = await factory.a2a(config).create(card)
# create() is typed as Client; patterns negotiation yields the experimental client
assert isinstance(client, A2AExperimentalClient)

request = SendMessageRequest(
    message=new_text_message("Status check", role=Role.ROLE_USER)
)

# Fan-out to 3 agents at once
responses = await client.broadcast_message(
    request,
    broadcast_topic="status_channel",
    recipients=[
        "default/default/agent1",
        "default/default/agent2",
        "default/default/agent3",
    ],
)

# `responses` is a list of SendMessageResponse (message or task per agent)
for resp in responses:
    print(f"Response: {resp}")

# A broadcast only returns normally when *every* recipient replied
# successfully. Otherwise it raises A2ABroadcastError, which keeps the
# partial result:
from agntcy_app_sdk.semantic.a2a import A2ABroadcastError

try:
    responses = await client.broadcast_message(request, recipients=[...])
except A2ABroadcastError as err:
    responses = err.responses             # the successful replies
    for failure in err.failures:          # error replies, with their sender
        print(f"{failure.sender} failed: {failure.error}")
    print(f"missing: {err.missing_recipients or err.missing}")

# Streaming variant — yields a StreamResponse for each event as it arrives
async for resp in client.broadcast_message_streaming(
    request,
    broadcast_topic="status_channel",
    recipients=[
        "default/default/agent1",
        "default/default/agent2",
        "default/default/agent3",
    ],
    message_limit=3,
):
    print(f"Streaming response: {resp}")
# The streaming variant yields every successful event first, then raises
# A2ABroadcastError at the end if a final reply failed or never arrived
# (err.responses is empty there, err.succeeded names who answered).

4. Run a Group Chat

Start a moderated multi-party conversation between agents. Each participant processes the message and forwards it to the next. Group chat uses the SLIM patterns transport — the same A2AExperimentalClient from above.

Note: Group chat currently requires SLIM transport.

from a2a.helpers import new_text_message
from a2a.types import (
    AgentCapabilities, AgentCard, AgentInterface, Role, SendMessageRequest,
)

from agntcy_app_sdk.factory import AgntcyFactory
from agntcy_app_sdk.semantic.a2a import A2AExperimentalClient
from agntcy_app_sdk.semantic.a2a.client.config import ClientConfig, SlimTransportConfig
from agntcy_app_sdk.semantic.a2a.server.card_bootstrap import InterfaceTransport

factory = AgntcyFactory()

# SLIM patterns transport — creates an A2AExperimentalClient
config = ClientConfig(
    slim_config=SlimTransportConfig(
        endpoint="http://localhost:46357",
        name="default/default/moderator",
    ),
)
card = AgentCard(
    name="default/default/agent_a",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(),
    skills=[],
    supported_interfaces=[
        AgentInterface(
            protocol_binding=InterfaceTransport.SLIM_PATTERNS,
            url="slim://default/default/agent_a",
        ),
    ],
    description="Agent A",
)

client = await factory.a2a(config).create(card)
assert isinstance(client, A2AExperimentalClient)

request = SendMessageRequest(
    message=new_text_message("Plan a team lunch", role=Role.ROLE_USER)
)

# Non-streaming group chat — collects all messages then returns
responses = await client.start_groupchat(
    init_message=request,
    group_channel="lunch_planning",
    participants=["default/default/agent_a", "default/default/agent_b"],
    end_message="DELIVERED",
    timeout=60,
)

for resp in responses:
    print(f"Group message: {resp}")

# Streaming group chat — yields each message as it arrives
async for message in client.start_streaming_groupchat(
    init_message=request,
    group_channel="lunch_planning",
    participants=["default/default/agent_a", "default/default/agent_b"],
    end_message="DELIVERED",
    timeout=60,
):
    print(f"Live: {message}")

5. Use MCP Tools over a Transport

Connect to a remote MCP server and call its tools — same transport layer as A2A, different protocol:

import asyncio
from agntcy_app_sdk.factory import AgntcyFactory

factory = AgntcyFactory()

# Create a transport (swap "SLIM" for "NATS" + endpoint to switch)
transport = factory.create_transport(
    "SLIM", endpoint="http://localhost:46357", name="default/default/mcp_client"
)

async def main():
    mcp_client = await factory.mcp().create_client(
        topic="my_weather_agent.mcp",
        transport=transport,
    )

    async with mcp_client as client:
        tools = await client.list_tools()
        print("Available tools:", tools)

        result = await client.call_tool(
            name="get_forecast",
            arguments={"location": "Colombia"},
        )
        print(f"Forecast: {result}")

    await transport.close()

asyncio.run(main())

📖 Complete MCP & FastMCP guide →

Transport & Protocol Support

The SDK negotiates the best transport automatically by intersecting the server's AgentCard transports with your ClientConfig capabilities. When the winning transport is slimpatterns or natspatterns, the factory returns an A2AExperimentalClient that extends the standard Client with broadcast and group-chat methods.

Architecture

Layered Architecture

┌─────────────────────────────────────────────────────────────┐
│                        AgntcyFactory                        │
│                 factory = AgntcyFactory()                   │
│                                                             │
│  factory.a2a(config)        → A2AClientFactory              │
│  factory.mcp()              → MCPClientFactory              │
│  factory.fast_mcp()         → FastMCPClientFactory          │
│  factory.create_transport("SLIM"|"NATS", ...)               │
│  factory.create_app_session(max_sessions=10)                │
└──────┬──────────────────┬──────────────────┬────────────────┘
       │                  │                  │
       ▼                  ▼                  ▼
┌──────────────┐  ┌───────────────────┐  ┌─────────────────────┐
│  Transport   │  │  Protocol Client  │  │     AppSession      │
│    Layer     │  │      Layer        │  │  (server-side)      │
│              │  │                   │  │                     │
│ BaseTransport│  │ a2a.client.Client │  │ .add(target)        │
│  ├ SLIM      │  │  (SlimRPC p2p)    │  │ .with_transport()   │
│  ├ NATS      │  │                   │  │ .with_topic()       │
│  └ HTTP      │  │ A2AExperimental-  │  │ .with_session_id()  │
│              │  │   Client          │  │ .build()            │
│ SlimRPC      │  │  (SLIM/NATS       │  │                     │
│  (native A2A │  │   patterns)       │  │ Targets:            │
│   transport) │  │                   │  │ ├ A2AServerConfig   │
│              │  │ MCPClientSession  │  │ ├ A2ASlimRpcServer │
│ 3 Mixins:    │  │ FastMCPClient     │  │ ├ MCP Server        │
│  ├ P2P       │  └───────────────────┘  │ └ FastMCP           │
│  ├ FanOut    │                         └─────────────────────┘
│  └ GroupChat │
└──────────────┘

Card-Driven Bootstrap (Recommended)

add_a2a_card() reads the agent card's supported_interfaces and creates all transports automatically — no manual builder chains required:

from agntcy_app_sdk.semantic.a2a.server.card_bootstrap import InterfaceTransport

agent_card = AgentCard(
    ...,
    supported_interfaces=[   # order = server preference (first is preferred)
        AgentInterface(protocol_binding=InterfaceTransport.SLIM_PATTERNS, url="slim://host:46357/topic"),
        AgentInterface(protocol_binding=InterfaceTransport.NATS_PATTERNS, url="nats://host:4222/topic"),
        AgentInterface(protocol_binding=InterfaceTransport.JSONRPC, url="http://0.0.0.0:9000"),
    ],
)

session = factory.create_app_session(max_sessions=10)
await (
    session.add_a2a_card(agent_card, request_handler)
    .with_factory(factory)
    .skip("jsonrpc")              # optionally skip a transport
    .start(keep_alive=True)
)

The JSONRPC URL in the card (http://0.0.0.0:9000 above) is the address the server binds to, and it is also what the served card advertises. Clients usually cannot reach 0.0.0.0 or an in-container address, so call .with_public_url("https://agent.example.com") to bind to the card's URL while advertising the public one. Your own card object is not modified.

SLIM shared secret

Every SLIM peer authenticates with a shared secret of at least 32 characters, and server and client must use the same value. The SDK picks it in this order:

1. The value you pass: shared_secret_identity on SLIMTransport / SlimTransportConfig, secret on SlimRpcConfig, shared_secret on SlimRpcConnectionConfig, or CardBuilder.with_shared_secret().
2. The SLIM_SHARED_SECRET environment variable.
3. A public placeholder, with a warning in the log. It only suits demos and local tests: any peer with the default can join your sessions.

CardBuilder stops before step 3: if neither a secret nor SLIM_SHARED_SECRET is set, or the value is the placeholder, start() and dry_run() raise ValueError. Call .allow_placeholder_secret() to run a local demo with the placeholder.

All SLIMTransports in one process share one SLIM connection, so they must use the same credentials; a transport asking for a different secret or JWT than the open connection raises ValueError. In production, set SLIM_SHARED_SECRET from a secret manager to a random value, for example python -c "import secrets; print(secrets.token_urlsafe(48))", and never commit it.

Manual Session Builder

The AppSession builder chains configuration into a single readable expression for fine-grained single-transport control. The SDK auto-detects the handler type from the target you pass to add():

factory = AgntcyFactory()
session = factory.create_app_session(max_sessions=3)

# A2A over SLIM patterns
session.add(a2a_server) \
    .with_transport(slim_transport) \
    .with_topic("my_agent") \
    .with_session_id("a2a") \
    .build()

# A2A over SlimRPC (transport managed internally)
session.add(a2a_srpc_config) \
    .with_session_id("slimrpc") \
    .build()

# A2A over HTTP (no transport needed)
session.add(a2a_server) \
    .with_host("0.0.0.0") \
    .with_port(9000) \
    .with_public_url("https://agent.example.com") \
    .with_session_id("http") \
    .build()

# Start everything
await session.start_all_sessions(keep_alive=True)

.with_public_url() is the address clients should use (it is added to the card as a JSONRPC interface if the card does not already declare one). The bind address from .with_host() / .with_port() is never written to the card, since it is often not reachable by clients.

Observability

Enable distributed tracing across all A2A and SLIM operations with a single flag:

factory = AgntcyFactory(enable_tracing=True)

This initialises the Agntcy Observe SDK (OpenTelemetry-based) and auto-instruments SLIM transports and A2A client calls. Traces are exported to the configured OTLP_HTTP_ENDPOINT (default: http://localhost:4318).

enable_tracing=True records the request in this process only. It does not set TRACING_ENABLED, so child processes do not inherit it. Create the factory before transports and A2A servers when you can. Components created earlier are remembered and installed when Observe.init() returns, including when you call Observe.init() yourself.

📁 Project Structure

📁 src/
└── 📦 agntcy_app_sdk/
    ├── 🏭 factory.py            # Main factory interface
    ├── 🔄 app_sessions.py       # Session management & fluent builder
    ├── 📂 directory/            # Agent directory services
    ├── 🔐 identity/             # Authentication & identity
    ├── 🧠 semantic/             # Protocol layer
    │   ├── a2a/                 #   A2A protocol + experimental patterns
    │   ├── mcp/                 #   MCP protocol bridge
    │   └── fast_mcp/            #   FastMCP protocol bridge
    ├── 🌐 transport/            # Transport implementations
    │   ├── slim/                #   SLIM (Secure Low-Latency Interactive Messaging)
    │   ├── nats/                #   NATS messaging
    │   └── streamable_http/     #   HTTP-based transport
    └── 🛠️  common/              # Shared utilities & logging

Reference Application

For a fully functional distributed multi-agent sample app, check out our coffeeAgntcy!

Agntcy Component Usage

Development

Prerequisites

• Python 3.12+
• uv — Package manager
• Task — Build orchestration (used locally and in CI)
• Docker — Required for E2E tests

Setup

# Install dependencies (including dev)
uv sync

# Run linting and type checking
task lint

# Auto-fix lint issues
task fix-lint

# Run unit tests
task test

# Run E2E tests (requires Docker services)
task test-e2e

Code Quality

The project uses Ruff for linting/formatting, mypy for type checking, and Codespell for spell checking. Pre-commit hooks enforce all checks automatically:

pre-commit run --all-files

Testing

The /tests directory contains both unit and end-to-end (E2E) tests for Agntcy components and workflows.

Prerequisites

Before running E2E tests, start the required message bus services:

docker-compose -f services/docker/docker-compose.yaml up

Running Tests

Unit tests (no Docker required):

task test

All E2E tests (requires Docker services):

task test-e2e

You can also pass additional arguments to pytest via the CLI_ARGS variable or run specific tests directly:

A2A Client Tests

Run all transports

uv run pytest tests/e2e/test_a2a_starlette.py::test_client -s

Run a single transport (e.g. SLIM):

uv run pytest tests/e2e/test_a2a_starlette.py::test_client -s -k "SLIM"

SlimRPC A2A

uv run pytest tests/e2e/test_a2a_slimrpc.py::test_client -s

Broadcast messaging

uv run pytest tests/e2e/test_a2a_starlette.py::test_broadcast -s

Group chat (SLIM only):

uv run pytest tests/e2e/test_a2a_starlette.py::test_groupchat -s -k "SLIM"

FastMCP Client Tests

Single transport (e.g. SLIM):

uv run pytest tests/e2e/test_fast_mcp.py::test_client -s -k "SLIM"

CI/CD

Continuous Integration

Every push to main and every pull request triggers the build-and-test workflow, which runs the following jobs in parallel:

Release Flow

Releases are automated via release-please:

1. Merge pull requests with Conventional Commits (e.g., feat: add X, fix: resolve Y).
2. release-please automatically creates a release PR that bumps the version in pyproject.toml and updates the changelog.
3. Merge the release PR to create a GitHub release and push a v* tag.
4. The release workflow builds multi-platform wheels (Linux x86/ARM, macOS ARM, Windows) and publishes to PyPI.

PR merged → release-please creates release PR → merge → tag v* → build wheels → publish to PyPI

Contributing

Contributions are welcome! Please see the contribution guide for details on how to contribute to the Agntcy Application SDK.