SDKs

Python SDK

The planned Python SDK (svc-sdk): what it will look like, and what to use until it ships.

This page previews the planned API so you can see how a Python agent will feel. Nothing below can be installed yet.

What it will be

Item	Planned value
Package	`svc-sdk` on PyPI (import name `svc_sdk`)
Python	3.10 and newer
Runtime model	`asyncio`; synchronous handlers run in worker threads
Core dependency	`httpx`
Optional extras	`svc-sdk[pydantic]`, `svc-sdk[otel]`, `svc-sdk[telemetry]`, `svc-sdk[all]`

Like every SDK, it connects out to the gateway (no inbound port), registers your tools, resources and prompts, runs your handlers when calls arrive, reconnects with backoff, and can report telemetry. See the SDK overview.

Install

bash

pip install svc-sdk                 # core
pip install "svc-sdk[pydantic]"     # richer schemas from Pydantic v2 models
pip install "svc-sdk[otel]"         # OpenTelemetry bridge (opentelemetry-api only)
pip install "svc-sdk[telemetry]"    # process metrics via psutil

A minimal agent

agent.py

import asyncio
import os

from svc_sdk import Agent

agent = Agent(api_key=os.environ["SPACEFIX_API_KEY"])  # URL comes from SPACEFIX_SERVICE_URL


@agent.tool(description="Look up a single order by its id. Returns null if it does not exist.")
async def get_order(id: str) -> dict | None:
    return await orders.find_by_id(id)


@agent.resource("orders://all", description="Every open order", mime_type="application/json")
async def all_orders() -> list[dict]:
    return await orders.find_open()


@agent.prompt(description="Summarise an order for a support reply.")
async def summarise_order(id: str) -> str:
    order = await orders.find_by_id(id)
    return f"Summarise this order politely:\n{order}"


if __name__ == "__main__":
    asyncio.run(agent.run())  # connects, reconnects, and shuts down cleanly on SIGINT/SIGTERM

bash

export SPACEFIX_API_KEY=agent_key_from_the_dashboard
export SPACEFIX_SERVICE_URL=https://api.spacefix.io   # optional; defaults to the production URL
python agent.py
# [svc-sdk] connected with 1 tool(s), 1 resource(s), 1 prompt(s)

agent.run() is the long-running entry point: it installs signal handlers, connects, and disconnects on SIGINT or SIGTERM. If you own the event loop, use await agent.connect() and agent.disconnect() instead.

Defining a tool

The SDK builds the tool's JSON Schema from the function's signature, and its description from the decorator argument or the docstring.

python

from typing import Annotated, Literal
from pydantic import Field  # optional; plain annotations work without it

@agent.tool
async def search_orders(
    customer_id: Annotated[str, Field(description="Customer id, for example cus_123")],
    status: Literal["open", "shipped", "cancelled"] = "open",
    limit: Annotated[int, Field(ge=1, le=100)] = 20,
) -> list[dict]:
    """Search a customer's orders by status. Returns at most `limit` orders, newest first."""
    return await orders.search(customer_id, status, limit)

A parameter without a default is required. Tool names default to the function name and must match ^[a-zA-Z0-9_-]{1,64}$. For richer arguments, pass a single Pydantic model, or give raw JSON Schema:

python

from pydantic import BaseModel, Field

class RefundArgs(BaseModel):
    order_id: str = Field(description="Order to refund")
    amount_cents: int = Field(ge=1, description="Amount in minor units")
    reason: str | None = None

@agent.tool(name="refund_order", description="Refund part or all of an order.")
async def refund(args: RefundArgs) -> dict:
    return await billing.refund(args.order_id, args.amount_cents, args.reason)

A handler may return any JSON-serialisable value. Raise ToolError("message") to fail a call on purpose; the message of any exception reaches the caller.

Configuration

python

Agent(
    api_key="...",                    # required: the project's agent key from the dashboard
    url="https://api.spacefix.io",    # optional
    otel={"arg_keys": True},          # optional: OpenTelemetry bridge; False turns it off
)

Option	Default	Meaning
`api_key`	none, required	The agent key. A missing or empty value raises `ValueError("api_key is required.")`
`url`	`SPACEFIX_SERVICE_URL`, then `https://api.spacefix.io`	The gateway base URL. Set it per environment
`otel`	on	`False` turns the bridge off. `{"arg_keys": True}` also records argument names, never values

The design also proposes max_concurrent_calls, sync_executor, logger, http_client, max_result_bytes and on_state_change.

Running in production

Run the agent as an always-on process (a container, a VM, a worker), next to your web service. Request-scoped serverless runtimes such as AWS Lambda freeze the process between requests and break the connection; they are not supported.

Handlers. async def handlers run concurrently on the event loop. Do not block the loop; a blocking handler delays every other call. Plain def handlers run in worker threads.
Scaling. Run more processes. Each opens its own connection, and the gateway spreads calls across them, so keep state in a database.
Shutdown. run() handles SIGINT and SIGTERM and waits for running handlers before it returns.
Health. agent.state is one of "connecting", "connected", "retrying", "stopped", with agent.connected as a boolean:

python

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/readyz")
async def readyz(response: Response):
    if not agent.connected:
        response.status_code = 503
    return {"state": agent.state}

Inside an existing FastAPI app, start the agent from the lifespan hook:

python

import asyncio, contextlib
from fastapi import FastAPI
from svc_sdk import Agent

agent = Agent(api_key=settings.mcp_api_key)

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(agent.connect())     # returns when disconnect() is called
    try:
        yield
    finally:
        agent.disconnect()
        await task

app = FastAPI(lifespan=lifespan)

With several Uvicorn or Gunicorn workers, each worker opens its own connection and each counts toward your plan's connection limit. Prefer a dedicated agent process unless you want that.

Telemetry

An optional module reports process health to the Agents page, the counterpart of @spacefix/telemetry:

python

from svc_sdk.telemetry import telemetry

agent.use(telemetry())                       # every 10s, over the agent's own connection
agent.use(telemetry(interval_s=30, report=lambda m: statsd.gauge("heap", m["heapUsed"])))

Option	Default	Meaning
`interval_s`	`10`	How often to sample and report
`report`	built in	Send samples elsewhere. Supplying it replaces the gateway transport
`otel`	`True`	Also publish the measurements through OpenTelemetry

OpenTelemetry

Installed with svc-sdk[otel], the bridge uses only the OpenTelemetry API. It is a no-op until your application registers an OpenTelemetry SDK, and it switches itself off when the API package is not installed. Each call that reaches your agent produces a server span (tools/call <name>, resources/read <uri> or prompts/get <name>) that continues the caller's trace, plus the duration histogram mcp.server.operation.duration, the counter spacefix.agent.reconnects, the gauge spacefix.agent.connected and connection logs. Argument values and results are never recorded. The attributes and metric names are the same in every SDK; see OpenTelemetry.

Next steps

Node.js SDK and Go SDK: available now
The agent protocol: connect from Python today without an SDK
SDK overview: every language and its status

Python SDK

#What it will be

#Install

#A minimal agent

#Defining a tool

#Configuration

#Running in production

#Telemetry

#OpenTelemetry

#Next steps

What it will be

Install

A minimal agent

Defining a tool

Configuration

Running in production

Telemetry

OpenTelemetry

Next steps