# 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-2--mcp-security-gateway-proxy).

---

# 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

```bash
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`.

## Test it

```bash
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

```python
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:

```python
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:

| Condition                          | JSON-RPC code           |
|-------------------------------------|--------------------------|
| Unknown tool name                   | `-32601 METHOD_NOT_FOUND` |
| Schema/format validation failure    | `-32602 INVALID_PARAMS`   |
| Unhandled exception in business logic | `-32603 INTERNAL_ERROR` |

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

| File | Role |
|---|---|
| `gateway.py` | The proxy: ASGI app, request handler, header plumbing, forwarding, response merging, CLI. |
| `gateway_auth.py` | `Authorization: Bearer <token>` → `Principal(subject, role)`. |
| `gateway_policy.py` | Pure authorization logic: which requests may be forwarded. |
| `gateway_jsonrpc.py` | JSON-RPC 2.0 wire-format parsing and error construction. |
| `mock_downstream.py` | Mock MCP server over HTTP. No auth of its own, on purpose. |
| `test_gateway.py` | 92 tests covering the whole path. |

## Run it

```bash
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`.

```bash
# 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:

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

## Test it

```bash
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

| Condition | JSON-RPC code | HTTP |
|---|---|---|
| `admin_*` tool called by a non-admin role | `-32001 Unauthorized Tool Call` | 403 |
| Missing / malformed / unknown bearer token | `-32000 Unauthenticated` | 401 |
| Downstream unreachable or timed out | `-32003 Bad gateway` | 502 |
| Body is not valid JSON | `-32700 Parse error` | 400 |
| Not a valid JSON-RPC 2.0 request / empty batch | `-32600 Invalid Request` | 400 |
| `tools/call` without a usable `params.name` | `-32602 Invalid params` | 200 |

`-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:

```python
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.