refund-mcp-server

Two related pieces:

• Task 1 — MCPServer.py: a minimal, strictly-validated MCP server over stdio. Documented below.

• Task 2 — gateway.py: an HTTP/JSON-RPC reverse proxy that authenticates callers and authorizes individual tools/call requests before they reach a downstream MCP server. See Task 2 — MCP Security Gateway.

Task 1 — refund-mcp-server

A minimal, strictly-validated MCP server (Python, official mcp SDK + Pydantic v2) exposing two tools:

• get_customer_record(customer_id: "CUST-XXXXX")

• trigger_refund(customer_id: "CUST-XXXXX", amount: float > 0, reason: str >= 10 chars)

Run it

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python MCPServer.py

It speaks JSON-RPC over stdio, so you normally won't run it directly — point an MCP client (Claude Desktop, mcp dev, your own ClientSession, etc.) at python /path/to/MCPServer.py.

Related MCP server: pogo-tb buyer MCP bridge

Test it

pip install pytest pytest-asyncio
pytest -v client.py

client.py spins up the real server as a subprocess (over stdio, just like a real client would) and covers:

• both tools are listed correctly,

• happy-path calls for both tools,

• a "customer not found" business outcome (still a successful JSON-RPC response, since the request itself was well-formed),

• every validation edge case (negative/zero/boolean/string amount, short reason, malformed customer_id, unknown extra fields) raising a real JSON-RPC -32602 INVALID_PARAMS error,

• an unknown tool name raising -32601 METHOD_NOT_FOUND,

• a raw-subprocess check that every line on stdout is valid, parseable JSON (the "stdout isolation" requirement).

Design decisions worth calling out

1. Strict Pydantic schemas

class TriggerRefundInput(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    customer_id: str = Field(..., pattern=r"^CUST-[A-Z0-9]{5}$")
    amount: float = Field(..., gt=0)
    reason: str = Field(..., min_length=10)

• strict=True turns off Pydantic's normal type coercion, so "10.0" is rejected for amount instead of silently parsed into 10.0.

• extra="forbid" rejects any field not in the schema (e.g. a client accidentally — or maliciously — passing an extra admin: true field).

• bool is a subclass of int in Python, so a plain float field would otherwise accept True/False as 1.0/0.0. A field_validator in mode="before" explicitly rejects booleans for amount.

• customer_id format is enforced by a regex (CUST- + 5 uppercase alphanumeric characters) on both tools. I assumed XXXXX means 5 alphanumeric characters based on the ticket; if the real spec is digits-only, this is a one-line change ([A-Z0-9]{5} → \d{5}).

2. STDIO isolation

• Logging setup is pulled out into its own module, logging_config.py, rather than living inline in MCPServer.py. It exposes get_logger(name), which configures the root logger to stderr exactly once (safe to call from multiple modules) and returns a standard logging.Logger. This is the one and only place logging output is configured — a future module (e.g. a separated business-logic layer) just does from logging_config import get_logger and gets stderr-only logging for free, with no risk of a second, differently-configured logger sneaking in.

• No print() calls exist anywhere in MCPServer.py.

• As defense-in-depth, logging_config.py also monkey-patches builtins.print at configure-time to force file=sys.stderr, so even an accidental future print("debug:", x) anywhere in the process can't corrupt the JSON-RPC channel on stdout.

• test_stdout_is_pure_json_rpc in client.py proves this empirically: it talks to the server as a raw subprocess (no client library abstraction) and asserts every line emitted on stdout is valid JSON.

3. JSON-RPC error-code compliance (the subtle part)

The installed SDK version (mcp==1.30.0) implements @server.call_tool() such that it catches every exception raised inside your handler (including a manually-raised McpError) and converts it into a successful JSON-RPC response whose result is CallToolResult(isError=True, ...). That's the MCP spec's recommended pattern for "the tool ran but the operation failed" (so an LLM caller sees the failure and can retry/self-correct) — but it is not a protocol-level JSON-RPC error, and this task explicitly asks for "standard MCP JSON-RPC error codes."

To get real top-level JSON-RPC errors, this server registers its own tools/call handler directly:

server.request_handlers[types.CallToolRequest] = _handle_call_tool

This is the same extension point @server.call_tool() uses internally, so it's a supported (if low-level) integration path rather than a hack. Inside _handle_call_tool, raising McpError(ErrorData(code=..., message=...)) propagates up to Server._handle_request, which specifically catches McpError and returns its .error as a genuine JSON-RPC error object:

A well-formed request for a customer that doesn't exist (e.g. get_customer_record("CUST-99999")) is not treated as a protocol error — the request was valid, so it returns a normal, successful result with found: false in the payload. Conflating "malformed request" with "legitimate not-found" is a common MCP anti-pattern this implementation avoids on purpose.

4. Mock data

Customer records are an in-memory dict (CUST-00001, CUST-00002) purely so the tools have something to operate on; swap _CUSTOMERS for a real data-access layer in production.

Task 2 — MCP Security Gateway Proxy

A lightweight HTTP/JSON-RPC reverse proxy that sits between an AI agent and a downstream MCP server, turns a Bearer token into a role, and refuses privileged tools/call requests without ever contacting the downstream.

 agent ──HTTP POST──▶ gateway.py ──HTTP POST──▶ mock_downstream.py
                          │
                          ├─ Authorization: Bearer <token> ─▶ Principal(subject, role)
                          ├─ tools/list  ─▶ forwarded transparently
                          ├─ tools/call  ─▶ params.name inspected:
                          │                 admin_*  requires role=admin,
                          │                 otherwise −32001, downstream untouched
                          └─ everything else ─▶ forwarded

Files

Run it

pip install -r requirements.txt

python mock_downstream.py --port 8081     # terminal 1: downstream MCP server
python gateway.py --port 8080             # terminal 2: the gateway

Demo tokens: admin-token → role admin, viewer-token → role viewer.

# viewer calling a privileged tool -> intercepted, downstream never called
curl -s localhost:8080/mcp \
  -H 'Authorization: Bearer viewer-token' -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call",
       "params":{"name":"admin_reset_key","arguments":{"key_id":"KEY-PRIMARY"}}}'
# HTTP 403
# {"jsonrpc":"2.0","id":1,"error":{"code":-32001,
#   "message":"Unauthorized Tool Call: tool 'admin_reset_key' requires role admin,
#              caller has role 'viewer'.",
#   "data":{"tool":"admin_reset_key","required_roles":["admin"],"role":"viewer"}}}

# same call as admin -> forwarded and executed
curl -s localhost:8080/mcp \
  -H 'Authorization: Bearer admin-token' -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/call",
       "params":{"name":"admin_reset_key","arguments":{"key_id":"KEY-PRIMARY"}}}'
# HTTP 200 -> {"jsonrpc":"2.0","id":2,"result":{...,"status":"key_reset"}}

# what actually reached the downstream
curl -s localhost:8081/_received

GET /healthz reports liveness and the configured upstream.

Configuration

python gateway.py --host 127.0.0.1 --port 8080 \
                  --upstream http://127.0.0.1:8081/mcp \
                  --path /mcp --timeout 30 \
                  [--filter-tools-list]

Tokens come from the MCP_GATEWAY_TOKENS environment variable — a JSON object of {"<token>": "<role>"} or {"<token>": {"sub": "...", "role": "..."}} — falling back to the built-in demo pair (with a warning) when unset:

export MCP_GATEWAY_TOKENS='{"tok-abc":{"sub":"svc-etl","role":"admin"}}'

Test it

pytest -q test_gateway.py            # 92 passed
pytest -q test_gateway.py client.py  # 111 passed (both tasks)

Note that Task 1's tests live in client.py, which a bare pytest does not collect — its filename doesn't match the default test_*.py pattern, so it has to be named explicitly.

The gateway and the mock downstream are both mounted in-process over httpx.ASGITransport, so the full path — HTTP headers → JSON-RPC parsing → auth → policy → forwarding → response merging — runs for real without binding a port. The load-bearing assertion throughout is mock_downstream.RECEIVED: an intercepted request must never appear in it. Coverage includes every authentication failure mode, every malformed-envelope shape, id echoing (including 0, "", null and float ids), batch splitting and merging, notification handling, role spoofing, and downstream unavailability.

Error codes

-32000/-32001/-32003 sit in the -32000..-32099 range the JSON-RPC spec reserves for implementation-defined server errors. Everything else is relayed verbatim from the downstream (e.g. its own -32601 for an unknown tool) — the gateway does not invent verdicts that aren't its to make.

Design decisions worth calling out

1. The security decision is a pure function

gateway_policy.authorize(principal, request) -> Decision does no I/O and knows nothing about HTTP or proxying. Reading one small file tells you exactly which requests can reach the downstream, and every rule is directly unit-testable without a server. The rule the brief asks for is one row of data rather than an if buried in the request path:

TOOL_PREFIX_RULES = (
    ("admin_", frozenset({"admin"})),
)

Adding finance_* → {finance, admin} is a data change, not a code change.

2. Deny means the downstream is never contacted

An unauthorized tools/call is answered entirely from the gateway. This is asserted in the tests via mock_downstream.RECEIVED, which is the only assertion that actually proves interception happened — checking the response body alone would still pass if the request had been forwarded and the answer rewritten.

3. Fail closed on undecidable requests

If a tools/call arrives without a string params.name, the gateway cannot evaluate the policy, so it returns -32602 locally instead of forwarding a request whose authorization status is unknowable. A malformed envelope is never a reason to pass something through.

4. Transparent when it can be, surgical when it can't

• Whole envelope allowed and no response rewriting needed → the original request bytes are forwarded unmodified and the downstream's status, body and headers are relayed as-is. No re-serialisation means no chance of the proxy altering semantics.

• Otherwise (a partially-denied batch, or --filter-tools-list) the gateway re-serialises just the permitted subset as a batch, then merges the downstream responses back with its own locally-generated errors, restoring the client's original request order.

5. Batches are handled member-by-member

A JSON-RPC batch is not one authorization decision, it's n. A viewer sending [get_customer_record, admin_reset_key, tools/list] gets three responses in order — two results and one -32001 — while the downstream sees a two-member batch it was allowed to execute. Notifications (no id) are honoured properly: a denied notification gets no response body (HTTP 204), because answering one would violate the spec.

6. The gateway is the trust boundary, and says so downstream

Forwarded requests carry X-MCP-Gateway-Subject, X-MCP-Gateway-Role and a correlating X-MCP-Gateway-Request-Id. Any client-supplied header in the x-mcp-gateway-* namespace is stripped before forwarding, so a caller cannot forge its own role by setting the header the downstream trusts — there's a test for exactly that escalation attempt. Hop-by-hop headers (RFC 7230 §6.1) are dropped in both directions, plus content-length (the body may be rewritten) and content-encoding (httpx hands back decoded bytes).

7. Bearer parsing follows the RFCs, not intuition

The scheme is compared case-insensitively (bEaReR admin-token works, per RFC 7235) while the token stays case-sensitive. A 401 always carries a WWW-Authenticate challenge (RFC 6750). Token lookup runs secrets.compare_digest against every configured token rather than a dict hit, so response timing doesn't leak which prefix was right.

8. HTTP status codes carry the verdict too

JSON-RPC over HTTP conventionally answers 200 and puts the failure in the body. That's honoured for ordinary application errors, but a security gateway is far easier to monitor and alert on when a refusal is visible at the HTTP layer, so single-request refusals also get a semantic status (403/401/502/400). The JSON-RPC error object is present in every case regardless, so a strict JSON-RPC client loses nothing. Mixed batches stay 200, since no single status can describe them.

9. Ids are echoed exactly, including the awkward ones

0, "", null and float ids all round-trip unchanged — they're easy to break with a truthiness check. "id": null (a request that must be answered with a null id) is distinguished from an absent id (a notification that must not be answered at all) via an explicit MISSING sentinel rather than None. Even authentication failures echo the id where the body permits, which is safe because an unauthenticated request is never forwarded.

10. tools/list filtering is opt-in

The brief specifies that tools/list is forwarded transparently, so by default it is — a viewer sees the admin_* tools in the listing and is refused only at call time. Because "Tool Filtering" is arguably the stronger posture (an agent that can't see a tool won't waste a turn on it, and won't be tempted by it), --filter-tools-list redacts unauthorized tools from the response. It's a flag rather than the default precisely because it deviates from the stated requirement.

11. The mock downstream trusts everyone, deliberately

mock_downstream.py has no auth and will happily rotate an API key for anyone who asks. That's what makes the tests meaningful: it stands in for a real MCP server whose privileged tools are only as safe as the gateway in front of them, so any request that reaches it proves the gateway let it through. It reuses Task 1's TOOL_REGISTRY and adds admin_reset_key and admin_list_audit_log so there is something worth guarding.

12. Auditability

Every decision is logged at INFO/WARNING with a per-request correlation id, the principal, the method, the tool and the outcome:

[7611d51f] DENY  bob@example.com (viewer) method=tools/call tool=admin_reset_key code=-32001
[28a3b360] ALLOW alice@example.com (admin) method=tools/call tool=admin_reset_key
[28a3b360] upstream http://127.0.0.1:8081/mcp -> 200 (12.3 ms, 1 forwarded, 0 intercepted)

Logging reuses Task 1's logging_config.py, so everything stays on stderr.

Known limitations

Called out rather than hidden — all are deliberate scope choices:

• Static tokens. Real deployments verify a signed JWT (or call an introspection endpoint) and read the role from a claim. That swap is confined to TokenStore.resolve(); nothing downstream of it changes.

• One role per principal. Principal.role is a single string because the brief describes exactly that. A frozenset of roles with roles & required_roles would be the natural extension.

• Bodies are buffered, not streamed. MCP's Streamable HTTP transport can reply with text/event-stream. A fully-allowed request relays such a body byte-for-byte with its original content type, but it is buffered rather than streamed, so the client sees it only once complete. The merge and filter paths need a JSON body outright; an SSE body there is logged and treated as empty. Streaming-aware merging is a larger piece of work.

• No session/initialize state. The gateway is stateless per request and does not track MCP session ids or enforce initialization ordering.

• No rate limiting or argument inspection. The policy looks at params.name, not params.arguments — it would not stop an admin-authorized trigger_refund for an implausible amount.