Python SDK

The mcp-s2sp Python package provides everything you need to build MCP servers with server-to-server data transfer capabilities. The S2SPServer class embeds a FastMCP instance automatically and provides the @s2sp_resource_tool() decorator, which wraps normal MCP tools with a control-plane/data-plane split: when abstract_domains is set, fields are filtered (control plane), full data is cached (data plane), and a resource_url (presigned URL) is returned for later retrieval.

Installation

Install from PyPI:

Or install with optional dependencies for async HTTP:

Requirements: Python 3.10+, mcp SDK, httpx, pydantic.

SDK Architecture

The SDK is organized around the S2SPServer class and supporting modules:

S2SPServer API

The S2SPServer class is the primary interface for building MCP servers with S2SP capabilities. It embeds a FastMCP instance and provides the @s2sp_resource_tool() decorator as the primary way to expose data tools.

Constructor

from mcp_s2sp import S2SPServer

server = S2SPServer("my-server")  # Name for the server; embeds FastMCP automatically

server.mcp

Access the embedded FastMCP instance directly. Use this to register custom tools, resources, or prompts that don't need the S2SP control-plane/data-plane split.

# Register a plain MCP tool via the embedded FastMCP instance
@server.mcp.tool()
async def ping() -> str:
    return "pong"

Resource Tools vs Consumer Tools

S2SP distinguishes two types of MCP tools:

@server.s2sp_resource_tool() — Resource Tool

Use this for tools that return tabular data. The decorator wraps a function returning list[dict] and automatically adds abstract_domains and mode parameters. When abstract_domains is set, the tool filters to those columns (+ _row_id), caches the full data, and returns a resource_url for data-plane access.

@server.s2sp_resource_tool()
async def get_alerts(area: str) -> list[dict]:
    """Get weather alerts for a US state."""
    data = await fetch_from_nws(area)
    return [f["properties"] for f in data["features"]]

Input (added by decorator)

Output (async mode)

{
  "total_rows": 8,
  "columns": ["id", "event", "severity", ... all 30],
  "abstract_domains": ["event", "severity"],
  "body_domains": ["description", "instruction", ... remaining 28],
  "resource_url": "http://host:port/s2sp/data/TOKEN...",
  "abstract": [
    {"_row_id": 0, "event": "Wind Advisory", "severity": "Moderate"},
    {"_row_id": 1, "event": "Flood Watch", "severity": "Minor"},
    ...
  ]
}

Output (sync mode)

{
  "total_rows": 8,
  "abstract_domains": ["event", "severity"],
  "body_domains": ["description", ...],
  "abstract": [
    {"_row_id": 0, "event": "Wind Advisory", "severity": "Moderate"},
    ...
  ],
  "body": [
    {"_row_id": 0, "description": "North winds 25-35 mph...", ...},
    ...
  ]
  // No resource_url — body is inline
}

Output (no abstract_domains)

Returns the raw list[dict] as-is — standard MCP behavior.

server.run()

Start the server as an MCP server over stdio. Compatible with mcp dev for local development.

server.run()

DirectChannel API (used by Consumer Tools)

The DirectChannel class provides a static method for consumer servers to fetch body data directly from a resource server's data plane.

DirectChannel.fetch_data()

Static method that consumer tools call to fetch rows from the resource server. Takes the presigned resource_url, optional row_ids, optional columns, and optional column_mapping to rename columns.

from mcp_s2sp.direct_channel import DirectChannel

items = await DirectChannel.fetch_data(
    resource_url,        # Presigned URL from s2sp_tool response
    row_ids=[0, 2, 5],   # Optional: select rows by _row_id
    columns=None,        # Optional: select columns (before mapping)
    column_mapping={     # Optional: rename columns after fetch
        "event": "alert_type",
        "areaDesc": "location",
    },
)

Note: The column_mapping parameter in @s2sp_consumer_tool() remaps body columns fetched from the resource server's data plane. Abstract columns are already in the agent's context, so the agent can rename them directly. DirectChannel.remap_columns(rows, mapping) is available as a standalone utility for advanced use cases.

Complete Example: Resource Server

A resource server exposes data via @server.s2sp_resource_tool(). The tool returns list[dict] and the S2SP decorator handles caching and field filtering automatically.

from mcp_s2sp import S2SPServer

server = S2SPServer("weather-server")

@server.s2sp_resource_tool()
async def get_alerts(area: str) -> list[dict]:
    """Get weather alerts for a US state."""
    data = await fetch_from_nws(area)
    return [f["properties"] for f in data["features"]]

server.run()

Complete Example: Consumer Server

A consumer server uses @server.s2sp_consumer_tool(). The decorator handles fetching, remapping, and merging — your function just receives merged rows.

from mcp_s2sp import S2SPServer

server = S2SPServer("stats-server")

@server.s2sp_consumer_tool()
async def draw_chart(rows: list[dict]) -> str:
    # rows = merged abstract + body, columns remapped
    return generate_chart(rows)

server.run()

Input (added by decorator)

What your function receives

A single rows: list[dict] argument — the merged result of abstract + body, joined by _row_id, with columns renamed if column_mapping was provided. Each row has all columns (both abstract and body). Example:

[
  {"_row_id": 0, "event": "Wind Advisory", "severity": "Moderate",
   "description": "North winds 25-35 mph...", "instruction": "Use caution...", ...},
  {"_row_id": 2, "event": "High Wind Warning", "severity": "Severe",
   "description": "Gusts up to 55 mph...", ...},
]