nate.swenson_
Local Fitness

Make MCP the front door: one server, a read-only fence, and a localhost that still needs locks

· 9 min read ·v0.3.0

Shipped

Until v0.3.0 the useful part of local-fitness lived in the wrong place. My Garmin data sits in SQLite, a Claude agent writes a daily briefing, and all of that ran server-side in a brief loop that no AI client could reach. Opening Claude to ask “how did I sleep this week” did not really work, because the data and the coaching had no front door.

This release makes MCP that front door. There is a coach prompt that assembles the daily snapshot and persona in one round-trip, fitness://schema and fitness://brief/latest resources, a one-call daily_snapshot tool, miles formatted server-side, and for the first time a write surface so observations and manual workouts can feed the training-load model. The feature is “talk to my app over MCP.” The part worth writing about is how to expose your own app this way without duplicating your server, without letting the read path mutate data, and without trusting “it’s only on my machine” as a security model. Here is the end-to-end build.

Setup: the server and its tools

The Model Context Protocol is a standard surface of tools, resources, and prompts that any compatible client can reach (Anthropic: Introducing MCP). The official Python SDK ships FastMCP, which turns plain functions into that surface with decorators (MCP Python SDK). Start with the dependencies.

python -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]" uvicorn starlette

Now define the read surface. A tool performs an action and returns structured data; a resource is addressable, read-only context the client can load by URI. The snapshot tool does the join across your SQLite tables once, and the schema resource lets a client discover the shape before it asks anything.

# fitness/server.py
from mcp.server.fastmcp import FastMCP
from fitness import db  # your own data layer over SQLite

# db.snapshot_for(day) returns a row exposing the attributes used below:
#   day, resting_hr, hrv_ms, sleep_minutes, distance_meters, training_load
# (a dataclass, sqlite3.Row, or any object with those fields). Substitute
# your own join over whatever tables hold sleep, HRV, and workout data.

def register_read_tools(mcp: FastMCP) -> None:
    @mcp.tool()
    def daily_snapshot(day: str | None = None) -> dict:
        """One-call snapshot: sleep, HRV, load, and last workout for a day."""
        row = db.snapshot_for(day)            # default: most recent day
        return {
            "day": row.day,
            "resting_hr": row.resting_hr,
            "hrv_ms": row.hrv_ms,
            "sleep_hours": round(row.sleep_minutes / 60, 1),
            "distance_miles": round(row.distance_meters / 1609.34, 2),  # format server-side
            "training_load": row.training_load,
        }

    @mcp.resource("fitness://schema")
    def schema() -> str:
        """The table shapes a client should know before querying."""
        return db.schema_as_markdown()

    @mcp.resource("fitness://brief/latest")
    def latest_brief() -> str:
        return db.latest_brief_markdown()

    @mcp.prompt(title="Coach")
    def coach(day: str | None = None) -> str:
        snap = daily_snapshot(day)
        return f"You are my training coach. Today's snapshot:\n{snap}\nGive me one focus for today."

Formatting miles inside daily_snapshot is deliberate. The client should never have to know your meters-to-miles convention, and every consumer gets the same answer because the conversion lives in one place.

One server, two transports

The easy mistake when adding a new way to reach your tools is to stand up a second server for the new transport. That immediately gives you two copies of every schema and handler to keep in sync. MCP supports multiple transports against the same server precisely so you do not have to (MCP: Streamable HTTP transport). stdio is what a local desktop client launches as a subprocess; streamable-HTTP is the network-reachable endpoint. Both are doors into the same room.

Build the server once, then choose how to serve it. For HTTP you mount the SDK’s already-wired ASGI app into Starlette, wrapping the session manager in a lifespan so it starts and stops cleanly.

# fitness/transports.py
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from fitness.server import register_read_tools
from fitness.writes import register_write_tools
from mcp.server.fastmcp import FastMCP

def build_full_server() -> FastMCP:
    mcp = FastMCP("fitness", json_response=True)
    register_read_tools(mcp)
    register_write_tools(mcp)
    return mcp

def http_app(mcp: FastMCP) -> Starlette:
    @contextlib.asynccontextmanager
    async def lifespan(app: Starlette):
        async with mcp.session_manager.run():
            yield
    # streamable_http_app() already serves on /mcp, so mount it at the root.
    # Mounting at "/mcp" would double-prefix the path to /mcp/mcp (a 404).
    return Starlette(routes=[Mount("/", app=mcp.streamable_http_app())], lifespan=lifespan)

if __name__ == "__main__":
    # stdio: the client spawns this process and speaks over stdin/stdout.
    build_full_server().run(transport="stdio")

One definition of daily_snapshot, reachable two ways. A new tool shows up on both transports at once, and there is no second implementation to drift.

Fence the write surface structurally

Adding write tools to the same server the brief loop uses created a real risk. A brief generation should never mutate my data, but now the tools that can mutate it live in the same process. The weak fix is a comment that says “the brief code just will not call the write tools.” A comment survives exactly until some future code, or some agent, ignores it.

The durable fix is to make the write tools unreachable from the brief path, so the rule is enforced by structure rather than discipline. This is the principle of least privilege applied at construction time: give the brief loop the minimum capability it needs and nothing more (OWASP: Least Privilege Principle). I split registration into read and write halves, then build a separate read-only server instance that never registers the write half.

# fitness/writes.py
from mcp.server.fastmcp import FastMCP
from fitness import db

def register_write_tools(mcp: FastMCP) -> None:
    @mcp.tool()
    def log_observation(note: str) -> dict:
        """Append a coach/athlete observation."""
        return {"id": db.insert_observation(note)}

    @mcp.tool()
    def add_manual_workout(kind: str, minutes: int, distance_miles: float = 0.0) -> dict:
        """Record a workout that Garmin did not capture; feeds training load."""
        return {"id": db.insert_workout(kind, minutes, distance_miles)}

# fitness/brief.py
from fitness.server import register_read_tools
from mcp.server.fastmcp import FastMCP

def build_read_only_server() -> FastMCP:
    """The server the brief loop is allowed to touch. Write tools are never registered."""
    mcp = FastMCP("fitness-read")
    register_read_tools(mcp)
    return mcp                      # log_observation / add_manual_workout do not exist here

def run_brief() -> str:
    mcp = build_read_only_server()  # handed only the read-only capability set
    # _tool_manager is a private SDK attribute; the public async equivalent is
    # `await mcp.list_tools()`. get_tool returns a Tool descriptor, not data.
    snapshot_tool = mcp._tool_manager.get_tool("daily_snapshot")
    return synthesize(snapshot_tool)  # synthesize() is your own reader; no path here reaches a write tool

The brief loop cannot call a write tool because the instance it holds never registered one. The safe thing is the only available thing.

Secure a localhost server

A server bound to your own machine still needs locks. The MCP transport spec is explicit: servers MUST validate the Origin header to prevent DNS rebinding, SHOULD bind to 127.0.0.1 rather than 0.0.0.0, and SHOULD authenticate (MCP: Streamable HTTP transport). DNS rebinding is the attack a “it’s only local” server tends to skip: a page you visit resolves an attacker-controlled hostname to 127.0.0.1, sidestepping the same-origin policy so the browser can talk to your local port (DNS rebinding). Validating Origin and binding to loopback closes that door.

My quality gate caught two execution-only gotchas here that source review missed. One was the missing Origin check. The other was a catch-all route registered before the MCP mount, which quietly shadowed it so nothing connected at all. Order the middleware and the mount with both in mind.

# fitness/secure.py
import os, hmac
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

# Browser Origin headers carry the port, so list the exact origin a client uses,
# e.g. http://localhost:8765, not just the scheme+host.
ALLOWED_ORIGINS = {
    "http://localhost", "http://127.0.0.1",
    "http://localhost:8765", "http://127.0.0.1:8765",
}

# BaseHTTPMiddleware buffers responses and breaks SSE streaming; it is safe here
# only because the server runs with json_response=True (no SSE stream to interrupt).
class GuardMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        origin = request.headers.get("origin")
        if origin is not None and origin not in ALLOWED_ORIGINS:
            return JSONResponse({"error": "bad origin"}, status_code=403)  # DNS-rebinding guard
        # Read the token lazily so an unset env var fails the request, not import.
        token = os.environ.get("FITNESS_MCP_TOKEN", "")
        auth = request.headers.get("authorization", "")
        if not hmac.compare_digest(auth, f"Bearer {token}"):
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

if __name__ == "__main__":
    import uvicorn
    from fitness.transports import build_full_server, http_app
    app = http_app(build_full_server())
    app.add_middleware(GuardMiddleware)               # runs before the MCP mount handles the request
    uvicorn.run(app, host="127.0.0.1", port=8765)     # loopback only, not 0.0.0.0

stdio mode skips the token because the client owns the subprocess and there is no network surface. HTTP mode gets the bearer gate plus the Origin and loopback defenses.

Use it from a client

For a desktop MCP client, point it at the stdio entry point. The client launches the process and speaks the protocol over its stdin and stdout (MCP: Connect to local servers).

{
  "mcpServers": {
    "fitness": {
      "command": "python",
      "args": ["-m", "fitness.transports"],
      "env": { "FITNESS_DB": "/home/me/fitness.db" }
    }
  }
}

After restarting the client, the coach prompt and the daily_snapshot tool appear. Asking “what’s my focus today?” triggers a tools/call, and the tool returns the same structured snapshot you defined, for example:

{
  "day": "2026-06-17",
  "resting_hr": 48,
  "hrv_ms": 72,
  "sleep_hours": 7.4,
  "distance_miles": 5.2,
  "training_load": "moderate"
}

The client now reads fitness://schema to learn the shape, then calls the tool, with no SQL on the client side and no second API to maintain.

Verify it

The two decisions worth defending are testable. Assert that the brief server cannot reach a write tool, and that a bad Origin is rejected. Both became regression tests so the fence and the guard cannot silently rot.

# tests/test_mcp_safety.py
from starlette.testclient import TestClient
from fitness.brief import build_read_only_server
from fitness.transports import build_full_server, http_app
from fitness.secure import GuardMiddleware

def tool_names(mcp) -> set[str]:
    return {t.name for t in mcp._tool_manager.list_tools()}

def test_brief_server_has_no_write_tools():
    names = tool_names(build_read_only_server())
    assert "daily_snapshot" in names
    assert "log_observation" not in names
    assert "add_manual_workout" not in names

def test_bad_origin_is_rejected(monkeypatch):
    monkeypatch.setenv("FITNESS_MCP_TOKEN", "test-token")
    app = http_app(build_full_server())
    app.add_middleware(GuardMiddleware)
    client = TestClient(app)
    resp = client.post("/mcp", headers={"origin": "http://evil.example"})
    assert resp.status_code == 403

The first test fails the moment someone registers a write tool on the brief path. The second fails if the Origin guard is removed or the catch-all route shadows the mount again.

Next

This is the setup for a bigger inversion. Once the agent can both read and write everything over MCP, the synthesis no longer needs to run inside the server at all; the brief loop can move out to an agent that talks to the same tools any other client uses. That becomes the agent-first rearchitecture in v0.5.0.

Sources

Changelog

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