stamp-mcp

An MCP server for NTP time and clock drift. Zero dependencies, fully synchronous, raw JSON-RPC 2.0 over stdio -- no MCP library, no asyncio.

A single NTP query tells you where your clock is right now. Drift history tells you where it is going: a clock that is consistently 200ms fast and accelerating is a different problem than one that is stable at 200ms fast. get_time takes the measurement; every call appends to a local log; get_drift reads the log and reports the trend.

Install

pip install stamp-mcp

Use with Claude Desktop

Edit ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "stamp": {
      "command": "stamp-mcp"
    }
  }
}

Or run the module directly:

{
  "mcpServers": {
    "stamp": {
      "command": "python3",
      "args": ["-m", "stamp_mcp"]
    }
  }
}

Tools

• get_time -- one NTP query: UTC time, this clock's offset in ms, network delay, stratum. Appends the sample to the drift log. Optional argument: server (default time.cloudflare.com).
• get_drift -- analyzes the drift log: sample count, timespan, current/mean/stddev offset, drift rate in ms/day (least-squares fit), first-half vs. second-half rates, and a verdict: stable, drifting, or accelerating. Optional argument: server to filter samples.

The drift log

Every get_time call appends one JSON line to ~/.stamp/drift.jsonl (override with STAMP_DRIFT_LOG). The file is capped at 10,000 samples. Call get_time periodically -- a cron job, a heartbeat, or just asking Claude "check the clock" now and then -- and get_drift turns the accumulated offsets into a trend.

How it works

The whole server is stamp_mcp/server.py:

• JSON-RPC 2.0 over stdio -- initialize, tools/list, tools/call, ping, notifications, and the standard error codes (-32700, -32601).
• Raw NTP with real offset math -- a 48-byte NTPv3 packet over UDP 123 carrying our transmit timestamp; the response's receive (t2) and transmit (t3) timestamps are unpacked with struct.unpack("!II", ...) as 64-bit fixed point, and offset/delay follow RFC 5905: offset = ((t2-t1)+(t3-t4))/2, delay = (t4-t1)-(t3-t2).
• Two rules that matter: stdout is the protocol channel (log to stderr only), and flush() after every write (subprocess stdout is block-buffered).

Hosted endpoint (streamable HTTP)

The same dispatch() also serves MCP's streamable-HTTP transport via stamp_mcp/http_server.py -- still zero dependencies (http.server):

stamp-mcp-http                      # binds 127.0.0.1:8000
STAMP_PORT=9000 stamp-mcp-http      # custom port
STAMP_TOKEN=secret stamp-mcp-http   # require "Authorization: Bearer secret"

• POST /mcp -- JSON-RPC requests (single or batch); notifications get 202, requests get 200 application/json.
• GET /mcp -- 405 (no SSE streams; nothing server-initiated exists).
• GET /health -- 200 for proxies and monitors.

TLS is terminated by a reverse proxy, not Python. The production layout is two containers on one AWS instance, wired by docker-compose.yml:

• stamp -- the server, built from Dockerfile, exposed only to the internal compose network. Drift log persists in the stamp-data volume.
• caddy -- official Caddy image, terminates HTTPS at stamp-mcp.terradev.cloud (automatic Let's Encrypt once DNS points at the instance) and reverse-proxies to stamp:8000.

deploy/ also has a non-container path (stamp-mcp.service systemd unit, deploy.sh) if you ever want to run it bare-metal.

Continuous deploy

.github/workflows/deploy.yml runs on every push to main: it uses AWS SSM Run Command (no SSH, no inbound connectivity needed -- the SSM agent dials out) to git pull on the instance and run docker compose up -d --build.

Required repo secrets (Settings -> Secrets -> Actions):

• AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY -- IAM creds allowed to ssm:SendCommand + ssm:GetCommandInvocation on the instance
• AWS_REGION -- e.g. us-east-1
• AWS_INSTANCE_ID -- e.g. i-0123456789abcdef0
• STAMP_TOKEN -- bearer token clients must send

Instance prerequisites: an IAM instance profile with AmazonSSMManagedInstanceCore, the SSM agent (preinstalled on Amazon Linux), docker + the compose plugin, git, and a DNS A record for stamp-mcp.terradev.cloud pointing at the instance. The security group must allow inbound 80/443 (ACME + HTTPS) and outbound UDP 123 (NTP).

Then point an MCP client at https://stamp-mcp.terradev.cloud/mcp with Authorization: Bearer <STAMP_TOKEN>.

Development

client.py is a test harness that plays the role of an MCP host -- it spawns the server and performs the real handshake, printing every raw frame:

python3 client.py                          # tests server.py (stage 1)
python3 client.py server_atomic.py         # tests the standalone artifact
python3 client.py stamp_mcp/server.py      # tests the package

server.py and server_atomic.py are the from-scratch learning artifacts; stamp_mcp/ is the packaged, published server.

Publishing

The GitHub Action in .github/workflows/publish-mcp.yml runs on version tags (git tag v0.3.0 && git push origin v0.3.0) and does two things:

1. Publishes the package to PyPI -- requires a PYPI_API_TOKEN repository secret (or configure Trusted Publishing on PyPI and remove the password line).
2. Publishes metadata to the MCP Registry -- uses mcp-publisher with GitHub OIDC (id-token: write), no secret needed. The server name io.github.theoddden/stamp is bound to the GitHub account; the mcp-name HTML comment at the top of this README is the PyPI ownership verification marker.

---

Appendix: how this was built, stage by stage

Build an MCP server with no library, one concept at a time. By the end you will have written every line yourself and the official MCP SDK becomes a convenience you could discard.

Stage 1 -- raw JSON-RPC over stdio (DONE, verified)

• server.py -- the entire protocol in ~100 lines: sys.stdin -> json -> dispatch -> sys.stdout -> flush().
• client.py -- plays the role of Claude Desktop. Spawns the server and performs the real handshake, printing every raw frame.

Run it:

python3 client.py

Things to notice in the output:

• initialize returns protocolVersion, capabilities, serverInfo.
• notifications/initialized has no id and gets no response.
• tools/list returns the manifest; inputSchema is plain JSON Schema.
• tools/call results are {"content": [{"type": "text", ...}]}.
• Unknown method -> JSON-RPC error -32601.
• Unknown tool -> a normal result with isError: true (so the model can read the failure and recover).
• Malformed JSON -> -32700.

Two rules that will bite you if ignored:

1. stdout is the protocol channel. One stray print() corrupts the stream. Log to stderr only.
2. flush() after every write. As a subprocess, stdout is block-buffered; without flush the host thinks the server is dead.

Stage 2 -- asyncio

Rewrite the stdin loop as an async coroutine:

• async def main() + asyncio.run(main())
• Read stdin without blocking the loop: loop.run_in_executor(None, sys.stdin.readline) or asyncio.StreamReader hooked to stdin via loop.connect_read_pipe.
• await each handler.

The payoff comes in stage 4 -- for now it is the same server with a different engine.

Stage 3 -- real NTP

Replace the stub get_time with a real query.

• First pass: pip install ntplib, then ntplib.NTPClient().request('pool.ntp.org', version=3).
• Second pass (optional, illuminating): delete ntplib and write the UDP query yourself. NTPv3 packet = 48 bytes, first byte 0x1B (LI=0, VN=3, Mode=3), rest zeros. Send to port 123, read 48 bytes back, unpack the transmit timestamp (bytes 40-43, seconds since 1900) with struct.unpack('!I', ...). Subtract 2208988800 to get Unix time. ntplib is ~200 lines of exactly this -- read its source once.

Stage 4 -- blocking vs. the event loop

The lesson you learn by breaking it:

1. Call ntplib directly inside your async def handler.
2. While a slow NTP server is being queried, send a ping from the client. Watch it hang -- the single-threaded event loop is frozen.
3. Fix it: await loop.run_in_executor(None, blocking_ntp_call). Blocking work goes to the thread pool; async work gets awaited.

Rule of thumb: ntplib, requests, file I/O = blocking. aiohttp, httpx (async mode), asyncpg = not blocking.

Connecting to Claude Desktop

Edit ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "ntp-scratch": {
      "command": "/usr/bin/python3",
      "args": ["/Users/theowolfenden/CascadeProjects/mcp-from-scratch/server.py"]
    }
  }
}

Restart Claude Desktop, then ask it "what tools do you have?" -- get_time should appear. If it doesn't, check the logs at ~/Library/Logs/Claude/mcp*.log -- a stray print or missing flush is the usual culprit.

The wire protocol, in one glance

>>> {"jsonrpc":"2.0","id":1,"method":"initialize","params":{...}}
<<< {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05",...}}
>>> {"jsonrpc":"2.0","method":"notifications/initialized"}     (no reply)
>>> {"jsonrpc":"2.0","id":2,"method":"tools/list"}
<<< {"jsonrpc":"2.0","id":2,"result":{"tools":[...]}}
>>> {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"get_time","arguments":{}}}
<<< {"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"..."}]}}

That is the whole thing. Everything else is plumbing.