# mcp-from-scratch

A from-scratch implementation of the Model Context Protocol (MCP): raw JSON-RPC 2.0 server, HTTP and stdio transports, a minimal ReAct agent loop, and a trajectory-grading evaluation harness. Built with stdlib and FastAPI only — no LangChain, no MCP SDK, no third-party agent frameworks.

## Why this exists

Most "MCP" code on the internet is `npm install` and five lines of config. When my agent's tool call fails at 3 a.m., I want to be able to read the spec and fix it, not grep through four layers of SDK abstractions. This repo is the smallest version that teaches you what JSON-RPC actually looks like on the wire — and then gives you an eval harness to grade what your agent did with it.

```
┌──────────┐    HTTP/stdio    ┌─────────────┐
│  Client  │ ───────────────► │ MCP Server  │
└──────────┘                  │  (raw JSON- │
                              │   RPC 2.0)  │
                              └──────┬──────┘
                                     │ dispatch
                              ┌──────┴──────┐
                              │ Tool/Resource│
                              │   Registry   │
                              └──────┬──────┘
                                     │
                              ┌──────┴──────┐
                              │ search_docs  │
                              │ (corpus.json)│
                              └─────────────┘
```

## What's in here

- **`src/mcp_from_scratch/jsonrpc.py`** — JSON-RPC 2.0 codec: `JsonRpcRequest`, `JsonRpcResponse`, `JsonRpcError`, encode/decode helpers, full error-code coverage (`-32700` parse, `-32600` invalid request, `-32601` method not found, `-32602` invalid params, `-32603` internal).
- **`src/mcp_from_scratch/transport_http.py`** — FastAPI HTTP transport, single `POST /rpc` endpoint, line-delimited JSON over `Content-Type: application/json`.
- **`src/mcp_from_scratch/transport_stdio.py`** — Stdio line-delimited JSON transport for piping agents.
- **`src/mcp_from_scratch/server.py`** — MCP server core: tool registry, resource registry, full dispatch over the five MCP methods (`initialize`, `tools/list`, `tools/call`, `resources/list`, `resources/read`).
- **`src/mcp_from_scratch/tools/search_docs.py`** — A single tool: `search_docs(query) -> list[dict]` over the bundled corpus.
- **`src/mcp_from_scratch/resources/docs.py`** — A single resource: `docs://{slug}` returning doc content.
- **`src/mcp_from_scratch/agent_loop.py`** — A minimal ReAct agent loop as an explicit state machine. `Thought → Act → Observe`, with a deterministic stub LLM (keyword router) and a real LLM hook for production. `max_steps` budget.
- **`src/mcp_from_scratch/eval/trajectory.py`** — `Trajectory` and `Step` data structures.
- **`src/mcp_from_scratch/eval/graders.py`** — Grader suite: `exact_match`, `substring_match` (`contains`), `tool_called`, `step_count`, `llm_judge`.
- **`src/mcp_from_scratch/eval/harness.py`** — `run_eval(agent_fn, test_set, graders) -> EvalReport`, plus a CLI runner.
- **`data/eval_set.json`** — Four-task eval suite: Python dataclasses, JSON-RPC spec, MCP tools, async I/O.
- **`examples/run_eval.py`** — Demo runner that emits `eval_report.json`.

## Quickstart

```bash
git clone https://github.com/aditya0si/mcp-from-scratch
cd mcp-from-scratch
pip install -e ".[dev]"

# Run unit tests
pytest

# Run the eval harness against the bundled demo agent
python examples/run_eval.py
# → writes eval_report.json with per-task scores
```

## Try the MCP server

```bash
# Start the HTTP transport
python -m mcp_from_scratch.transport_http
# → listens on http://127.0.0.1:8765/rpc
```

A `curl` roundtrip — full MCP `initialize` then `tools/list`:

```bash
curl -s http://127.0.0.1:8765/rpc \
  -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"client":"demo","version":"0.1.0"}}'
# {"jsonrpc":"2.0","id":1,"result":{"server":"mcp-from-scratch","version":"0.1.0","capabilities":{"tools":true,"resources":true}}}

curl -s http://127.0.0.1:8765/rpc \
  -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list"}'
# {"jsonrpc":"2.0","id":2,"result":[{"name":"search_docs","description":"...","inputSchema":{...}}]}
```

Then call the tool:

```bash
curl -s http://127.0.0.1:8765/rpc \
  -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"search_docs","arguments":{"query":"python json-rpc"}}}'
```

## How the agent loop works

State machine, no LangChain. Each step is explicit:

```python
class AgentLoop:
    def step(self, task: str) -> Step:
        thought = self.llm.think(self.history, task)
        action = self.router.parse(thought)        # tool_call | final_answer
        if action.type == "final_answer":
            return Step(thought=thought, action=action, observation="")
        observation = self.tools.call(action.tool, action.arguments)
        return Step(thought=thought, action=action, observation=observation)
```

Loop terminates on `final_answer` or after `max_steps`. The LLM is an injected callable — for tests we use a deterministic stub (keyword router), for production we wire in any function-calling LLM.

## Eval harness — what's graded

Each test runs the agent against a task, captures the full `Trajectory`, then scores it with the grader list:

| Grader | What it checks |
|---|---|
| `exact_match` | Final answer equals expected (case/whitespace normalized) |
| `substring_match` (alias `contains`) | Final answer contains an expected substring |
| `tool_called` | A specific tool name appears in the trajectory |
| `step_count` | Total steps ≤ a maximum |
| `llm_judge` | An LLM rates the final answer 1-5 against the expected |

`run_eval(agent_fn, test_set, graders=[...])` returns:

```json
{
  "total_tests": 4,
  "passed_tests": 4,
  "pass_rate": 1.0,
  "per_test": [
    {
      "id": "eval-python-dataclass",
      "task": "...",
      "passed": true,
      "final_answer": "...",
      "trajectory": { "steps": [...], "tool_calls": [...] },
      "grader_results": [
        {"grader": "exact_match", "passed": true, "score": 1.0, "reason": "..."},
        {"grader": "tool_called", "passed": true, "score": 1.0, "reason": "..."},
        {"grader": "step_count", "passed": true, "score": 1.0, "reason": "..."}
      ]
    }
  ],
  "aggregate": { "avg_steps": 1.75, "avg_tool_calls": 0.75 }
}
```

Run `python examples/run_eval.py` to regenerate.

## Tests

29 unit tests across five files:

```
tests/test_jsonrpc.py          6 tests
tests/test_server.py           7 tests
tests/test_transport_http.py   6 tests
tests/test_agent_loop.py       4 tests
tests/test_eval_harness.py     6 tests
```

All passing on Python 3.11+. CI runs them on every PR.

## Project layout

```
mcp-from-scratch/
├── data/
│   ├── corpus.json            # ~20 short docs the tool searches over
│   └── eval_set.json          # 4-task evaluation set
├── examples/
│   └── run_eval.py            # demo runner
├── src/mcp_from_scratch/
│   ├── agent_loop.py
│   ├── eval/
│   │   ├── graders.py
│   │   ├── harness.py
│   │   └── trajectory.py
│   ├── jsonrpc.py
│   ├── resources/
│   │   └── docs.py
│   ├── server.py
│   ├── tools/
│   │   └── search_docs.py
│   ├── transport_http.py
│   └── transport_stdio.py
└── tests/
    ├── test_agent_loop.py
    ├── test_eval_harness.py
    ├── test_jsonrpc.py
    ├── test_server.py
    └── test_transport_http.py
```

## Why no LangChain / SDK

Because the point is to understand what's underneath. After reading this code, when you use the official MCP SDK or LangGraph you'll know exactly what JSON-RPC message is on the wire, why the tool call failed, and where the dispatch lives. You'll also be able to debug production agents without grepping four layers of abstraction.

## Roadmap

- [ ] Real LLM integration examples (OpenAI function calling, Anthropic, Groq)
- [ ] Streaming responses over SSE
- [ ] Resource subscriptions and notifications
- [ ] Pluggable tool registry (load from YAML)
- [ ] More graders: cost, latency, token-budget
- [ ] CI workflow with GitHub Actions

## License

MIT.