Agent Development with S2SP Servers

This guide shows how to build an agent that uses two S2SP servers: a resource server (@s2sp_resource_tool()) that provides weather alerts, and a consumer server (@s2sp_consumer_tool()) that generates charts from the data.

The example uses real NWS (National Weather Service) data with ~30 columns per alert. It demonstrates both async and sync modes, and works with Claude Desktop, OpenAI, and MCP Inspector.

Architecture

  Weather Server             Agent (LLM)                Stats Server
  @s2sp_resource_tool()                                           @s2sp_consumer_tool()
  get_alerts()  ──abstract──▶  filters by             draw_chart()
  get_forecast()                event, severity   ◀── receives abstract
                                selects _row_ids       + resource_url
                                      │
                                      │ async mode:
                                      ▼
  Weather Server  ◀──POST /s2sp/data/{id}──  Stats Server
                     body domains fetched directly (data plane)
                     agent never sees this data

Weather Server (Resource)

A standard MCP server with @server.s2sp_resource_tool(). Returns all ~30 NWS alert fields. The decorator adds abstract_domains and mode parameters 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 — ~30 columns per alert."""
    data = await make_nws_request(
        f"https://api.weather.gov/alerts/active?area={area}"
    )
    return [f["properties"] for f in data["features"]]

server.run()  # mcp dev demos/weather_agent/weather_server.py

Stats Server (Consumer)

Uses @server.s2sp_consumer_tool(). The decorator handles all S2SP plumbing — parsing, fetching body, remapping columns, merging by _row_id. 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:
    """Receives merged abstract + body rows."""
    return generate_chart(rows)

server.run()

# The decorator adds: abstract_data, resource_url, body_data, column_mapping
# Agent calls: draw_chart(abstract_data='[...]', resource_url='http://...')

Async Mode Demo

Body data stays cached on the Weather Server. The Stats Server fetches it directly via the data plane. The agent never sees the body.

# Step 1: Agent calls Weather Server with abstract_domains
result = get_alerts(area="CA",
    abstract_domains="event,severity,urgency,status",
    mode="async")

# Agent sees only:
# { "abstract": [{"_row_id": 0, "event": "Wind Advisory", ...}, ...],
#   "resource_url": "http://..." }
# No body data. No resource_url means body is inline.

# Step 2: Agent filters on abstract (control plane)
wind_rows = [r for r in result["abstract"]
             if "Wind" in r["event"]]

# Step 3: Agent passes abstract rows + resource ref to Stats Server
draw_chart(
    abstract_data=json.dumps(wind_rows),
    
    resource_url=result["resource_url"],
)
# Stats Server fetches body via POST /s2sp/data/dK7x_...
# Merges abstract + body by _row_id, generates chart

$ python demos/weather_agent/run_async.py --area CA --event Wind

Sync Mode Demo

Body data is returned inline alongside the abstract. No resource_url — no server-to-server fetch needed. The agent's SDK layer has the body but should not send it to the LLM.

# Step 1: Agent calls Weather Server in sync mode
result = get_alerts(area="CA",
    abstract_domains="event,severity,urgency,status",
    mode="sync")

# Agent sees abstract AND body inline:
# { "abstract": [{"_row_id": 0, "event": "Wind Advisory", ...}, ...],
#   "body": [{"_row_id": 0, "description": "...", ...}, ...] }
# No resource_url — body is right here.

# Step 2: Agent filters on abstract (control plane)
wind_rows = [r for r in result["abstract"]
             if "Wind" in r["event"]]

# Step 3: Filter body to match, pass both to Stats Server
wind_ids = {r["_row_id"] for r in wind_rows}
wind_body = [b for b in result["body"]
             if b["_row_id"] in wind_ids]

draw_chart(
    abstract_data=json.dumps(wind_rows),
    body_data=json.dumps(wind_body),
)
# No data-plane fetch. Stats Server merges and generates chart.

$ python demos/weather_agent/run_sync.py --area CA --event Wind

Running the Servers

To use these servers with Claude Desktop or an interactive agent, see Use S2SP Servers in Claude.

MCP Inspector

Debug each server independently with the MCP Inspector:

# Weather Server (source) — tools: get_alerts, get_forecast
$ mcp dev demos/weather_agent/weather_server.py

# Stats Server (consumer) — tool: draw_chart
$ CLIENT_PORT=6280 SERVER_PORT=6281 mcp dev demos/weather_agent/stats_server.py

Measured Results

Metric	Traditional MCP	S2SP Async	S2SP Sync
Tokens in agent context	~10,000 (all 30 columns)	~600 (abstract only)	~600 (abstract only*)
Body data through agent	Yes (all in LLM context)	No (server-to-server)	Through SDK, not LLM
Server-to-server fetch	N/A	Yes (POST /s2sp/data/)	No (inline)
Token savings	—	85-96%	85-96%

* In sync mode, body is in the tool response but a well-designed agent SDK feeds only the abstract to the LLM. The token count reflects what the LLM processes.

File Structure

demos/weather_agent/
├── weather_server.py    # Resource: @s2sp_resource_tool() — get_alerts, get_forecast
├── stats_server.py      # Consumer: @s2sp_consumer_tool() — draw_chart
├── run_async.py         # Scripted async demo
├── run_sync.py          # Scripted sync demo
├── agent_async.py       # Interactive agent (async mode)
└── agent_sync.py        # Interactive agent (sync mode)