Nate Swenson

Reuse your Claude Agent SDK tool server as a standalone MCP endpoint

Local Fitness · No. 011 · 8 min read ·v0.3.0

Shipped

Until v0.3.0, local-fitness’s Claude agent tools only ran inside my own process. The same @tool-decorated functions backed the daily brief and the chat REPL, but nothing outside that process could reach them. This release exposed that exact tool server to real MCP clients, Claude Code and Claude Desktop among them, over both stdio and streamable-HTTP. It added a coach prompt that loads the day’s snapshot in one round trip, two read-only resources for the DB schema and the latest brief, a one-call daily_snapshot tool, and a write surface so an interactive session can log observations and manual workouts that feed the training-load model. A design and quality-gate pass on it earned its keep: three fresh-eyes rounds caught SDK-level bugs that only showed up when the code actually ran, not when I read it.

The reusable part isn’t the fitness app. It’s the pattern. If you already built a Claude Agent SDK tool server for your own agent loop, you can expose that same server to any other MCP client without writing a second implementation, restrict an unattended caller to a safe subset of tools with an allow-list instead of a runtime check, and close the handful of gaps a network-reachable localhost server has that a subprocess-launched one doesn’t. Here’s the build.

Setup: one tool server, in-process

create_sdk_mcp_server wraps @tool-decorated functions into an in-process MCP server that your own query() calls can use directly, no subprocess, no separate schema to maintain (Claude docs: give Claude custom tools). A small version of the pattern local-fitness runs:

# tools.py
from typing import Any
from claude_agent_sdk import tool, create_sdk_mcp_server

NOTES: list[str] = []

@tool("add_note", "Save a short note", {"text": str})
async def add_note(args: dict[str, Any]) -> dict[str, Any]:
    NOTES.append(args["text"])
    return {"content": [{"type": "text", "text": f"saved ({len(NOTES)} total)"}]}

@tool("list_notes", "List saved notes", {})
async def list_notes(_args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": "\n".join(NOTES) or "no notes yet"}]}

ALL_TOOLS = [add_note, list_notes]

def make_server():
    return create_sdk_mcp_server(name="notes", version="1.0.0", tools=ALL_TOOLS)

What create_sdk_mcp_server hands back is easy to skim past. Run it and look:

$ python3 -c "from tools import make_server; c = make_server(); print(c['type'], type(c['instance']))"
sdk <class 'mcp.server.lowlevel.server.Server'>

instance is a real mcp.server.lowlevel.Server, the same low-level object the standalone MCP SDK builds by hand. That’s the whole trick: the SDK already built you a fully-wired server. You don’t reimplement your tool schemas for a second transport, you reuse the object.

Reuse it as a standalone endpoint

The MCP Python SDK ships a StreamableHTTPSessionManager you can mount into any ASGI app, with the manager’s run() started from the app’s lifespan (MCP Python SDK: mounting into Starlette). Point it at the same server instance from above, add a token gate, and mount it before any catch-all route your app already serves:

# transport.py
import contextlib, secrets
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.server.transport_security import TransportSecuritySettings
from starlette.applications import Starlette
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Mount, Route
from tools import make_server

TOKEN = "test-token"

def build_app(allowed_hosts: list[str]) -> Starlette:
    server = make_server()["instance"]
    manager = StreamableHTTPSessionManager(
        app=server,
        stateless=True,
        json_response=True,   # single JSON reply per call, no SSE stream to manage
        security_settings=TransportSecuritySettings(allowed_hosts=allowed_hosts),
    )

    @contextlib.asynccontextmanager
    async def lifespan(app: Starlette):
        # Starts the session manager's task group. Skip this and every
        # /mcp/ request raises "Task group is not initialized".
        async with manager.run():
            yield

    class TokenGate(BaseHTTPMiddleware):
        async def dispatch(self, request, call_next):
            path = request.url.path
            if path == "/mcp" or path.startswith("/mcp/"):
                auth = request.headers.get("authorization", "")
                if not secrets.compare_digest(auth, f"Bearer {TOKEN}"):
                    return JSONResponse({"error": "unauthorized"}, status_code=401)
            return await call_next(request)

    async def spa_catch_all(request):
        return PlainTextResponse("SPA-SHELL")

    app = Starlette(
        routes=[
            Mount("/mcp", app=manager.handle_request),   # registered first
            Route("/{full_path:path}", spa_catch_all),   # catch-all, registered second
        ],
        lifespan=lifespan,
    )
    app.add_middleware(TokenGate)
    return app

One definition of add_note and list_notes, reachable from your own agent loop and from any external MCP client that can hit this endpoint. Nothing about the tools themselves changed.

Fence an unattended caller with an allow-list

Not every caller of this server should get the same tools. local-fitness’s brief runs on a schedule with nobody watching, so it should never be able to write to the database, while an interactive chat session, where a human reads every reply, is fine with the full set. The fence is an explicit allow-list, not a denylist, so a new tool defaults to excluded until someone deliberately adds it:

# The unattended caller's set. A future tool is left out unless
# someone adds it here on purpose.
_READ_ONLY_TOOL_NAMES = ("list_notes",)

def allowed_tool_names() -> list[str]:
    return [f"mcp__notes__{t.name}" for t in ALL_TOOLS]

def read_only_tool_names() -> list[str]:
    return [f"mcp__notes__{n}" for n in _READ_ONLY_TOOL_NAMES]

The scheduled job builds its ClaudeAgentOptions with allowed_tools=read_only_tool_names(). The interactive one uses allowed_tool_names(). Same server, same tool definitions, two different capability sets handed out at construction time, and a test that asserts add_note never shows up in the read-only list catches the day someone adds a write tool and forgets to leave it out.

Harden the endpoint against DNS rebinding

A server bound to your own machine still needs to check who’s asking. The MCP transport spec is explicit about all three of these for streamable-HTTP: validate Origin on every connection, bind to loopback rather than 0.0.0.0 when running locally, and authenticate (MCP spec: streamable HTTP transport). TransportSecuritySettings takes both allowed_hosts for the Host header and allowed_origins for the Origin header a browser sends on a cross-origin request; the token gate above is the third leg. DNS rebinding is the attack “it’s only local” tends to skip: an attacker registers a domain, points your browser at it, then re-resolves that same domain to 127.0.0.1 after the page loads, and the same-origin policy no longer protects your loopback port because the hostname never changed from the browser’s point of view (Wikipedia: DNS rebinding). An empty or missing allowed_hosts closes the Host-header half of that gap by rejecting every request with an unrecognized Host before it reaches a tool; a non-browser client like a locally-launched agent carries no Origin at all, which is exactly why the token still has to hold up on its own.

Use it and verify it

The two decisions worth defending, mount order and the host/token gate, are cheap to check against the real objects instead of eyeballing the code:

# verify.py
import json
from starlette.routing import Mount, Route
from starlette.testclient import TestClient
from transport import TOKEN, build_app

app = build_app(allowed_hosts=["fitness.home.local", "testserver"])

routes = app.router.routes
mcp_idx = next(i for i, r in enumerate(routes) if isinstance(r, Mount) and r.path == "/mcp")
catchall_idx = next(i for i, r in enumerate(routes) if isinstance(r, Route) and "{full_path" in r.path)
print(f"mount before catch-all: {mcp_idx < catchall_idx}")

HDRS = {"Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "Host": "fitness.home.local"}
init_body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
    "params": {"protocolVersion": "2025-06-18", "capabilities": {},
               "clientInfo": {"name": "demo", "version": "1"}}})

with TestClient(app) as client:
    r = client.post("/mcp/", content=init_body, headers=HDRS)
    print(f"no token: {r.status_code}")

    bad_host = {**HDRS, "Authorization": f"Bearer {TOKEN}", "Host": "evil.example.com"}
    r = client.post("/mcp/", content=init_body, headers=bad_host)
    print(f"bad host: {r.status_code}")

    good = {**HDRS, "Authorization": f"Bearer {TOKEN}"}
    r = client.post("/mcp/", content=init_body, headers=good)
    print(f"token + allowed host: {r.status_code}")

Running it against the real server built above:

mount before catch-all: True
no token: 401
bad host: 421
token + allowed host: 200

A follow-up tools/call for add_note over that same session comes back {"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"saved (1 total)"}],"isError":false}}, a single JSON reply, not an SSE stream, because json_response=True was set on the session manager.

Gotchas

Three of the bugs a design and quality-gate pass on this caught were only visible by running the code, not by reading it.

Skipping manager.run() in the lifespan looks harmless because the app still starts. The first request to /mcp/ then raises RuntimeError: Task group is not initialized, every time, because mounting the handler never started its session manager. The fix is the async with manager.run(): yield shown above, not a retry.

An allowed_hosts list that doesn’t contain the exact host your client’s requests will carry turns into a wall of 421 responses, even with a correct token and correct routes. localhost and 127.0.0.1 are not the same string as fitness.home.local:8765, and the check is literal.

The sneakiest one is a catch-all route registered before the MCP mount. Nothing errors. A GET /mcp/ just returns whatever the catch-all serves instead of ever reaching the session manager, and the failure looks like “the client can’t find the server” rather than “the server has a routing bug.” Starlette matches routes in registration order, so more specific routes and mounts need to come before general ones (Starlette: routing). A route-order assertion like the one in verify.py catches this before a client ever connects.

Sources

Changelog

  • Make the fitness MCP the primary interface (#21) (#22) (dbe865a)