SLIMA2A is a SLIM transport for A2A using slim's RPC protocol (srpc). It allows agents to communicate over the SLIM network using the A2A protocol, using SLIM identities for authentication and addressing.
slim-bindings2.xa2a-sdk1.1.x
Both A2A protocol versions are supported: v1.0 (slima2a.types.v1, the default)
and v0.3 (slima2a.types.v0, via the compatibility layer in slima2a.compat.v3_0).
- Refer to this documentation to download the correct slirpc compiler for your system (2.x, matching
slim-bindings2.x) and make sure to have in in $PATH. - Install bufbuild
Run the following from the repo root:
task generateThis regenerates both the A2A v0.3 (slima2a/types/v0) and v1.0 (slima2a/types/v1)
SlimRPC bindings. Use task generate-v0 / task generate-v1 to regenerate one of them.
import os
import slim_bindings
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from slima2a import setup_slim_client
from slima2a.handler import SRPCHandler
from slima2a.types.v1.a2a_pb2_slimrpc import add_A2AServiceServicer_to_server
# Initialize and connect to SLIM (simplified helper)
service, local_app, local_name, conn_id = await setup_slim_client(
namespace="agntcy",
group="demo",
name="echo_agent",
secret=os.environ["SLIM_SHARED_SECRET"], # or identity=..., see below
)
# Create request handler
agent_executor = MyAgentExecutor()
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
agent_executor=agent_executor,
task_store=task_store,
agent_card=agent_card,
)
# Create servicer
servicer = SRPCHandler(agent_card, request_handler)
# Create server
server = slim_bindings.Server.new_with_connection(local_app, local_name, conn_id)
add_A2AServiceServicer_to_server(servicer, server)
# Run server
await server.serve_async()agent_card is an a2a.types.a2a_pb2.AgentCard (the v1.0 protobuf message), and
the agent is reachable on the SLIM network under agntcy/demo/echo_agent — the
namespace/group/name triple passed to setup_slim_client.
If you need more control over the SLIM configuration:
import asyncio
import os
import slim_bindings
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from slima2a.handler import SRPCHandler
from slima2a.types.v1.a2a_pb2_slimrpc import add_A2AServiceServicer_to_server
# Set the event loop for slim_bindings. Call the top-level wrapper: as of
# slim-bindings 2.0 slimrpc lives in its own UniFFI namespace and the loop has
# to be registered on both, which the wrapper takes care of.
slim_bindings.uniffi_set_event_loop(asyncio.get_running_loop())
# Initialize slim_bindings service
tracing_config = slim_bindings.new_tracing_config()
runtime_config = slim_bindings.new_runtime_config()
service_config = slim_bindings.new_service_config()
tracing_config.log_level = "info"
slim_bindings.initialize_with_configs(
tracing_config=tracing_config,
runtime_config=runtime_config,
service_config=[service_config],
)
service = slim_bindings.get_global_service()
# Create local name
local_name = slim_bindings.Name("agntcy", "demo", "echo_agent")
# Connect to SLIM
client_config = slim_bindings.new_insecure_client_config("http://localhost:46357")
conn_id = await service.connect_async(client_config)
# Create app with a shared secret (>= 32 chars), or use
# service.create_app(local_name, provider, verifier) for a real identity
local_app = service.create_app_with_secret(
local_name, os.environ["SLIM_SHARED_SECRET"]
)
# Subscribe to local name
await local_app.subscribe_async(local_name, conn_id)
# Create request handler
agent_executor = MyAgentExecutor()
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
agent_executor=agent_executor,
task_store=task_store,
agent_card=agent_card,
)
# Create servicer
servicer = SRPCHandler(agent_card, request_handler)
# Create server
server = slim_bindings.Server.new_with_connection(local_app, local_name, conn_id)
add_A2AServiceServicer_to_server(servicer, server)
# Run server
await server.serve_async()SRPCCompatHandler exposes the same DefaultRequestHandler over the A2A v0.3
service definition. Register both servicers on one server to accept v0.3 and v1.0
clients simultaneously:
from slima2a.compat.v3_0.handler import SRPCCompatHandler
from slima2a.handler import SRPCHandler
from slima2a.types.v0.a2a_pb2_slimrpc import (
add_A2AServiceServicer_to_server as add_v0,
)
from slima2a.types.v1.a2a_pb2_slimrpc import (
add_A2AServiceServicer_to_server as add_v1,
)
server = slim_bindings.Server.new_with_connection(local_app, local_name, conn_id)
add_v0(SRPCCompatHandler(agent_card, request_handler), server)
add_v1(SRPCHandler(agent_card, request_handler), server)
await server.serve_async()Note that SRPCCompatHandler takes the v1.0 agent card and request handler —
it converts to and from the v0.3 wire types internally.
import os
import httpx
from a2a.client import minimal_agent_card
from a2a.helpers import new_text_message
from a2a.types.a2a_pb2 import Role, SendMessageRequest
from slima2a import setup_slim_client
from slima2a.client_transport import (
ClientConfig,
MultiAgentClientFactory,
slimrpc_channel_factory,
)
# Initialize and connect to SLIM (simplified helper)
service, slim_local_app, local_name, conn_id = await setup_slim_client(
namespace="agntcy",
group="demo",
name="client",
secret=os.environ["SLIM_SHARED_SECRET"], # or identity=..., see below
)
# Create client config
httpx_client = httpx.AsyncClient()
client_config = ClientConfig(
supported_protocol_bindings=["slimrpc"],
streaming=True,
httpx_client=httpx_client,
slimrpc_channel_factory=slimrpc_channel_factory(slim_local_app, conn_id),
)
# Create client factory (the "slimrpc" transport is registered automatically)
client_factory = MultiAgentClientFactory(client_config)
# Create client with minimal agent card
agent_card = minimal_agent_card("agntcy/demo/echo_agent", ["slimrpc"])
client = client_factory.create(card=agent_card)
# Send message
request = SendMessageRequest(message=new_text_message("Hello, world!", role=Role.ROLE_USER))
async for stream_response in client.send_message(request=request):
match stream_response.WhichOneof("payload"):
case "message":
parts = stream_response.message.parts
case "task":
parts = [p for a in stream_response.task.artifacts for p in a.parts]
case "artifact_update":
parts = stream_response.artifact_update.artifact.parts
case _:
parts = []
for part in parts:
if part.WhichOneof("content") == "text":
print(part.text)client.send_message() yields StreamResponse messages whose payload oneof is
one of message, task or artifact_update — which one you get depends on the
agent, so handle all three. See examples/echo_agent/client.py
for a fuller version.
If you need more control over the SLIM configuration:
import asyncio
import os
import httpx
import slim_bindings
from a2a.client import minimal_agent_card
from a2a.helpers import new_text_message
from a2a.types.a2a_pb2 import Role, SendMessageRequest
from slima2a.client_transport import (
ClientConfig,
MultiAgentClientFactory,
slimrpc_channel_factory,
)
# Set the event loop for slim_bindings. Call the top-level wrapper: as of
# slim-bindings 2.0 slimrpc lives in its own UniFFI namespace and the loop has
# to be registered on both, which the wrapper takes care of.
slim_bindings.uniffi_set_event_loop(asyncio.get_running_loop())
# Initialize slim_bindings service
tracing_config = slim_bindings.new_tracing_config()
runtime_config = slim_bindings.new_runtime_config()
service_config = slim_bindings.new_service_config()
tracing_config.log_level = "info"
slim_bindings.initialize_with_configs(
tracing_config=tracing_config,
runtime_config=runtime_config,
service_config=[service_config],
)
service = slim_bindings.get_global_service()
# Create local name
local_name = slim_bindings.Name("agntcy", "demo", "client")
# Connect to SLIM
client_config_slim = slim_bindings.new_insecure_client_config("http://localhost:46357")
conn_id = await service.connect_async(client_config_slim)
# Create app with a shared secret (>= 32 chars), or use
# service.create_app(local_name, provider, verifier) for a real identity
slim_local_app = service.create_app_with_secret(
local_name, os.environ["SLIM_SHARED_SECRET"]
)
# Subscribe to local name
await slim_local_app.subscribe_async(local_name, conn_id)
# Create client config
httpx_client = httpx.AsyncClient()
client_config = ClientConfig(
supported_protocol_bindings=["slimrpc"],
streaming=True,
httpx_client=httpx_client,
slimrpc_channel_factory=slimrpc_channel_factory(slim_local_app, conn_id),
)
# Create client factory (the "slimrpc" transport is registered automatically)
client_factory = MultiAgentClientFactory(client_config)
# Create client with minimal agent card
agent_card = minimal_agent_card("agntcy/demo/echo_agent", ["slimrpc"])
client = client_factory.create(card=agent_card)
# Send message
request = SendMessageRequest(message=new_text_message("Hello, world!", role=Role.ROLE_USER))
async for stream_response in client.send_message(request=request):
match stream_response.WhichOneof("payload"):
case "message":
parts = stream_response.message.parts
case "task":
parts = [p for a in stream_response.task.artifacts for p in a.parts]
case "artifact_update":
parts = stream_response.artifact_update.artifact.parts
case _:
parts = []
for part in parts:
if part.WhichOneof("content") == "text":
print(part.text)Override the registered slimrpc transport with the v0.3 compatibility transport:
from slima2a.compat.v3_0.client_transport import SRPCCompatTransport
client_factory = MultiAgentClientFactory(client_config)
client_factory.register("slimrpc", SRPCCompatTransport.create, multiagent=True)Passing a list of agent cards to MultiAgentClientFactory.create() returns a
MulticastClient that sends a single request to every agent in the group and
yields (source, response) tuples as the replies arrive. This requires
slimrpc_group_channel_factory on the config:
from slima2a.client_transport import (
ClientConfig,
MultiAgentClientFactory,
slimrpc_channel_factory,
slimrpc_group_channel_factory,
)
client_config = ClientConfig(
supported_protocol_bindings=["slimrpc"],
streaming=True,
httpx_client=httpx_client,
slimrpc_channel_factory=slimrpc_channel_factory(slim_local_app, conn_id),
slimrpc_group_channel_factory=slimrpc_group_channel_factory(
slim_local_app, conn_id
),
)
client_factory = MultiAgentClientFactory(client_config)
client = client_factory.create(
card=[
minimal_agent_card("agntcy/demo/echo_agent_1", ["slimrpc"]),
minimal_agent_card("agntcy/demo/echo_agent_2", ["slimrpc"]),
]
)
async for source, stream_response in client.send_message(request):
print(source, stream_response.WhichOneof("payload"))source is a RpcMessageContext identifying the agent that produced the response,
so replies from different agents can be told apart as they interleave.
setup_slim_client and connect_and_subscribe configure two independent things:
- App identity (required, exactly one):
secret, a shared secret (>= 32 chars), oridentity=SlimIdentity(provider, verifier), anyIdentityProviderConfig/IdentityVerifierConfigthe bindings support (shared secret, static or signed JWT including EdDSA, JWT verification with PEM/JWK/JWKS keys, SPIRE). It is passed toService.create_app. There is no default secret: omitting both raisesValueError. - Node connection (optional, at most one):
slim_urlfor a plaintext connection (defaulthttp://localhost:46357), orclient_config, a fullslim_bindings.ClientConfigwith TLS, auth, headers and timeouts.
Conflicting pairs (secret + identity, slim_url + client_config) raise
ValueError before any connection is opened.
With a shared secret, anyone who holds the secret can claim any SLIM name. A per-agent JWT identity binds each app to a key you control. Every member of a session needs a compatible identity type. Here an agent signs its own Ed25519 JWT, trusts tokens verified against a JWKS file, and connects to a node over TLS with a bearer token:
from datetime import timedelta
import slim_bindings
from slima2a import SlimIdentity, setup_slim_client
name = "agntcy/demo/my_agent"
hour = timedelta(hours=1)
identity = SlimIdentity(
provider=slim_bindings.IdentityProviderConfig.JWT(
config=slim_bindings.ClientJwtAuth(
key=slim_bindings.JwtKeyType.ENCODING(
key=slim_bindings.JwtKeyConfig(
algorithm=slim_bindings.JwtAlgorithm.ED_DSA,
format=slim_bindings.JwtKeyFormat.PEM,
key=slim_bindings.JwtKeyData.FILE(path="/etc/slim/agent-key.pem"),
)
),
audience=["agntcy/demo"],
issuer="agntcy",
subject=name,
duration=hour,
)
),
verifier=slim_bindings.IdentityVerifierConfig.JWT(
config=slim_bindings.JwtAuth(
# A JWKS file of trusted public keys acts as the allowlist.
key=slim_bindings.JwtKeyType.DECODING(
key=slim_bindings.JwtKeyConfig(
algorithm=slim_bindings.JwtAlgorithm.ED_DSA,
format=slim_bindings.JwtKeyFormat.JWKS,
key=slim_bindings.JwtKeyData.FILE(path="/etc/slim/trusted.jwks.json"),
)
),
audience=["agntcy/demo"],
issuer="agntcy",
subject=None,
duration=hour,
)
),
)
# TLS (system CA pool) plus a bearer token read from a file.
client_config = slim_bindings.new_secure_client_config("https://slim.example.com:443")
client_config.auth = slim_bindings.ClientAuthenticationConfig.STATIC_JWT(
config=slim_bindings.StaticJwtAuth(
token_file="/var/run/secrets/slim/token", duration=hour
)
)
service, local_app, local_name, conn_id = await setup_slim_client(
"agntcy", "demo", "my_agent", identity=identity, client_config=client_config
)tests/test_identity_e2e.py runs the same
identity setup end to end with an in-memory key.
SRPCHandler's DefaultCallContextBuilder sets the caller to
UnauthenticatedUser. The slimrpc server Context in slim-bindings 2.x
exposes no source name or verified identity for the caller, only the session
id, deadline and caller-supplied metadata. SLIM still verifies the peer's
identity before a request reaches the handler, but the handler cannot see who
the peer is. Metadata is set by the caller and is not used as identity.
Exposing the verified subject/issuer depends on a bindings upgrade.
The slima2a package provides convenient helper functions to simplify SLIM setup:
setup_slim_client(namespace, group, name, slim_url=None, secret=None, log_level="info", *, identity=None, client_config=None)- Complete SLIM client setup in one callinitialize_slim_service(log_level="info")- Initialize SLIM service with default configurationconnect_and_subscribe(service, local_name, slim_url=None, secret=None, *, identity=None, client_config=None)- Connect to SLIM server and subscribe to a local name
Both require exactly one of secret or identity; see
Identity and node connection.
Runnable end-to-end examples live under examples/:
examples/echo_agent- minimal echo agent, covering unary, streaming, A2A v0.3 compatibility and multicastexamples/travel_planner_agent- a LangChain-based agent
They expect a SLIM node reachable at http://localhost:46357, and use the
shared secret from $SLIM_SHARED_SECRET (or --secret for the echo agent),
falling back to a publicly known demo-only placeholder that the sibling
slim-a2a SDK examples share.