Building an MCP Server

# script_id: day_040_mcp_servers/minimal_server
# A one-tool MCP server over stdio. Install: uv add "mcp[cli]"   Env: ORBIT_DATA (default datasets/orbit)
import os
from pathlib import Path

from mcp.server import MCPServer

DOCS = Path(os.environ.get("ORBIT_DATA", "datasets/orbit")) / "docs"
mcp = MCPServer("atlas-mini", version="0.1.0")

@mcp.tool()
def search_docs(query: str, k: int = 3) -> list[str]:
    """Find Orbit documentation pages by keyword. Returns up to k page ids, best first."""
    words = query.lower().split()
    scores = {p.stem: sum(p.read_text().lower().count(w) for w in words) for p in DOCS.glob("*.md")}
    return sorted((d for d in scores if scores[d]), key=scores.get, reverse=True)[:k]

if __name__ == "__main__":
    mcp.run()  # stdio is the default transport

# script_id: day_040_mcp_servers/minimal_client
# Launches minimal_server.py over stdio and calls its tool. Install: uv add "mcp[cli]"   Env: ORBIT_DATA
import asyncio
import os
import sys
from pathlib import Path

from mcp import Client, StdioServerParameters

SERVER = Path(__file__).with_name("minimal_server.py")

async def main() -> None:
    # stdio servers inherit only a safe default env (PATH, HOME, ...), so pass what they need
    env = {"ORBIT_DATA": os.environ.get("ORBIT_DATA", "datasets/orbit")}
    params = StdioServerParameters(command=sys.executable, args=[str(SERVER)], env=env)
    async with Client(params) as client:
        tools = await client.list_tools()
        print([t.name for t in tools.tools])
        print(tools.tools[0].input_schema)
        result = await client.call_tool("search_docs", {"query": "lease"})
        print(result.structured_content)

asyncio.run(main())

['search_docs']
{'type': 'object', 'properties': {'query': {'title': 'Query', 'type': 'string'}, 'k': {'default': 3, 'title': 'K', 'type': 'integer'}}, 'required': ['query'], 'title': 'search_docsArguments'}
{'result': ['leases-and-heartbeats', 'troubleshooting-duplicates', 'environment-variables-2x']}

# script_id: day_040_mcp_servers/break_it
# Three v1 habits that break on SDK v2. Install: uv add "mcp[cli]"
from mcp.server import MCPServer

mcp = MCPServer("atlas", "Search Orbit docs")  # meant as a description...
print(f"title={mcp.title!r} version={mcp.version!r}")  # ...it became the title

try:
    @mcp.tool  # missing ()
    def ping() -> str:
        return "pong"
except TypeError as err:
    print(f"TypeError: {err}")

try:
    MCPServer("atlas", port=8000)  # v1 habit: transport settings on the constructor
except TypeError as err:
    print(f"TypeError: {err}")

title='Search Orbit docs' version=''
TypeError: The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool
TypeError: MCPServer.__init__() got an unexpected keyword argument 'port'

# script_id: day_040_mcp_servers/traversal_demo
# Same naive code as a resource and as a tool; only one is protected. Install: uv add "mcp[cli]"   Env: ORBIT_DATA
import asyncio
import os
from pathlib import Path

from mcp import Client, MCPError
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

DOCS = Path(os.environ.get("ORBIT_DATA", "datasets/orbit")) / "docs"
DOC_IDS = frozenset(p.stem for p in DOCS.glob("*.md"))
demo = MCPServer("traversal-demo")

@demo.resource("orbit://docs/{doc_id}")
def naive_resource(doc_id: str) -> str:
    return (DOCS / f"{doc_id}.md").read_text()  # BAD: trusts doc_id

@demo.tool()
def naive_get_doc(doc_id: str) -> str:
    """Return one doc page."""
    return (DOCS / f"{doc_id}.md").read_text()  # BAD: the same line, now inside a tool

@demo.tool()
def safe_get_doc(doc_id: str) -> str:
    """Return one doc page by its id."""
    if doc_id not in DOC_IDS:  # allowlist: only ids we know ever reach the filesystem
        raise ToolError(f"Unknown doc_id {doc_id!r}. Call search_docs to find ids.")
    return (DOCS / f"{doc_id}.md").read_text()

async def main() -> None:
    async with Client(demo) as client:
        try:
            await client.read_resource("orbit://docs/..%2FREADME")
        except MCPError as err:
            print("resource:", err)
        leaked = await client.call_tool("naive_get_doc", {"doc_id": "../README"})
        print("naive tool:", leaked.content[0].text.splitlines()[0])
        blocked = await client.call_tool("safe_get_doc", {"doc_id": "../README"})
        print("safe tool:", blocked.is_error, blocked.content[0].text)

asyncio.run(main())

resource: Unknown resource: orbit://docs/..%2FREADME
naive tool: # The Orbit dataset
safe tool: True Error executing tool safe_get_doc: Unknown doc_id '../README'. Call search_docs to find ids.

# script_id: day_040_mcp_servers/signature_sketch
# fragment
@mcp.tool(annotations=READ_ONLY)
def search_docs(
    query: Annotated[str, Field(min_length=2, max_length=200)],
    k: Annotated[int, Field(ge=1, le=10)] = 5,            # -> "minimum": 1, "maximum": 10
    applies_to: Literal["2.x", "3.x", "all"] | None = None,  # -> "enum": [...]
) -> list[DocHit]: ...

Error executing tool search_docs: 1 validation error for search_docsArguments
k
  Input should be less than or equal to 10 [type=less_than_equal, input_value=50, input_type=int]

# script_id: day_040_mcp_servers/readonly_sketch
# fragment
conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)               # the file refuses writes
row = conn.execute("SELECT title FROM issues WHERE id = ?", (issue_id,)).fetchone()
# conn.execute(f"SELECT COUNT(*) FROM {table}")  <- never: identifiers can't be parameters

# script_id: day_040_mcp_servers/atlas_mcp_server
# Atlas MCP server: Orbit docs search + cited answers, read-only, stdio or Streamable HTTP.
# Install: uv add "mcp[cli]" openai pydantic
# Env:     ORBIT_DATA (default datasets/orbit) · ATLAS_MODEL (default gpt-6-luna)
#          OPENAI_API_KEY (optional: without it, `ask` returns retrieval-only, degraded=True)
# Run:     python atlas_mcp_server.py          # stdio, for a host that launches it
#          python atlas_mcp_server.py --http   # http://127.0.0.1:8000/mcp
import math
import os
import re
import sys
from collections import Counter
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
DOCS_DIR = (DATA_DIR / "docs").resolve()
MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
WORD = re.compile(r"[a-z0-9]+")

def parse_page(path: Path) -> dict:
    """Split the front matter (key: value lines between '---') from the Markdown body."""
    _, head, body = path.read_text(encoding="utf-8").split("---\n", 2)
    meta = {}
    for line in head.splitlines():
        key, _, value = line.partition(":")
        meta[key.strip()] = value.split(" #")[0].strip().strip('"')  # drop comments, quotes
    return meta | {"body": body, "tf": Counter(WORD.findall(body.lower()))}

PAGES = {p.stem: parse_page(p) for p in sorted(DOCS_DIR.glob("*.md"))}
DOC_IDS: frozenset[str] = frozenset(PAGES)  # the allowlist: no other id ever touches the disk
DF = Counter(term for page in PAGES.values() for term in page["tf"])  # pages containing each term

def snippet(body: str, terms: list[str], width: int = 240) -> str:
    """First prose paragraph that mentions a query term, flattened to one line."""
    for para in (p.strip() for p in body.split("\n\n")):
        if not para.startswith(("#", "`", "|")) and any(t in para.lower() for t in terms):
            return " ".join(para.split())[:width]
    return " ".join(body.split())[:width]

# script_id: day_040_mcp_servers/atlas_mcp_server
from pydantic import BaseModel, Field

def keyword_search(query: str, k: int, applies_to: str | None = None) -> list[dict]:
    """Keyword scorer: rare words weigh more (IDF), and a word in the title adds a bonus."""
    terms = WORD.findall(query.lower())
    scored = []
    for doc_id, page in PAGES.items():
        if applies_to and page["applies_to"] not in (applies_to, "all"):
            continue
        score = sum(
            (page["tf"][t] + 5 * (t in page["title"].lower())) * math.log(len(PAGES) / DF[t])
            for t in terms if DF[t]
        )
        if score > 0:
            scored.append((score, doc_id))
    top = sorted(scored, reverse=True)[:k]
    return [PAGES[d] | {"doc_id": d, "snippet": snippet(PAGES[d]["body"], terms)} for _, d in top]

class DocHit(BaseModel):
    doc_id: str
    title: str
    section: str
    snippet: str

class Citation(BaseModel):  # Day 15's shape
    doc_id: str
    quote: str = Field(description="A sentence copied exactly from the cited page.")

class Draft(BaseModel):  # what the model must return
    answer: str
    citations: list[Citation]

class Answer(Draft):  # what the tool returns: the draft plus one honesty flag
    degraded: bool = False  # True = no model call was made; citations are raw retrieval

# script_id: day_040_mcp_servers/atlas_mcp_server
# gpt-6-luna, USD per 1M input/output tokens; as of 2026-11, verify at developers.openai.com/api/docs/pricing
LUNA_PRICE = (0.10, 0.50)

def log_cost(usage) -> None:
    """Compact stand-in for Day 12's CostTracker: one line per call, on stderr."""
    usd = (usage.input_tokens * LUNA_PRICE[0] + usage.output_tokens * LUNA_PRICE[1]) / 1e6
    print(f"[atlas] ask {usage.input_tokens}+{usage.output_tokens} tok ${usd:.5f}", file=sys.stderr)

from mcp.server.mcpserver.exceptions import ToolError

def answer_question(question: str, hits: list[dict]) -> Answer:
    """Compact stand-in for R2's ask() (Days 28 and 34). Swap in your own through the seam."""
    if not hits or not os.environ.get("OPENAI_API_KEY"):
        cites = [Citation(doc_id=h["doc_id"], quote=h["snippet"]) for h in hits]
        return Answer(answer="No model call made; these are the closest passages.",
                      citations=cites, degraded=True)
    from openai import OpenAI  # imported here so the server starts without a key
    pages = "\n\n".join(f"<page id='{h['doc_id']}'>\n{h['body'][:4000]}\n</page>" for h in hits)
    resp = OpenAI().responses.parse(
        model=MODEL, text_format=Draft, input=f"{pages}\n\nQuestion: {question}",
        instructions="Answer only from the pages. Cite each claim with its page id and an exact "
                     "quote. If the pages don't answer the question, say so and cite nothing.",
        reasoning={"effort": "none"}, max_output_tokens=800)  # no reasoning: a small cap is safe
    log_cost(resp.usage)
    if resp.output_parsed is None:  # refusal, or cut off at max_output_tokens
        raise ToolError(f"Model returned no answer (status={resp.status}). Try a narrower question.")
    allowed = {h["doc_id"] for h in hits}  # drop citations to pages we never sent
    cites = [c for c in resp.output_parsed.citations if c.doc_id in allowed]
    return Answer(answer=resp.output_parsed.answer, citations=cites)

# script_id: day_040_mcp_servers/atlas_mcp_server
from difflib import get_close_matches
from typing import Annotated, Literal

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ResourceNotFoundError
from mcp.types import ToolAnnotations

READ_ONLY = ToolAnnotations(read_only_hint=True, destructive_hint=False, open_world_hint=False)

mcp = MCPServer(
    "atlas",
    version="1.0.0",
    instructions="Atlas knows Orbit, a Python background-jobs library. Use search_docs to "
    "find pages, ask for a cited answer, and the orbit://docs/{doc_id} resource for a full "
    "page. Every tool is read-only.",
)

def safe_doc_path(doc_id: str) -> Path:
    """Allowlist first, path check second: only known ids reach the filesystem."""
    if doc_id not in DOC_IDS:
        close = [d for d in sorted(DOC_IDS) if doc_id and doc_id in d]
        close = close or get_close_matches(doc_id, DOC_IDS, n=1, cutoff=0.5)
        hint = f" Closest: '{close[0]}'." if close else " Call search_docs to find ids."
        raise ResourceNotFoundError(f"Unknown doc_id '{doc_id}'.{hint}")
    path = (DOCS_DIR / f"{doc_id}.md").resolve()
    if not path.is_relative_to(DOCS_DIR):  # unreachable today; guards a future, looser allowlist
        raise ResourceNotFoundError(f"doc_id '{doc_id}' is outside the docs folder.")
    return path

@mcp.resource("orbit://docs/{doc_id}", mime_type="text/markdown")
def read_doc(doc_id: str) -> str:
    """One Orbit documentation page as Markdown, front matter included."""
    return safe_doc_path(doc_id).read_text(encoding="utf-8")

# script_id: day_040_mcp_servers/atlas_mcp_server
@mcp.tool(annotations=READ_ONLY)
def search_docs(
    query: Annotated[str, Field(min_length=2, max_length=200, description="Keywords, e.g. 'lease'")],
    k: Annotated[int, Field(ge=1, le=10, description="How many pages to return")] = 5,
    applies_to: Literal["2.x", "3.x", "all"] | None = None,
) -> list[DocHit]:
    """Find Orbit documentation pages by keyword, best match first.

    Use this to locate pages before reading or citing them. applies_to='3.x' keeps pages
    for 3.x plus version-agnostic ones; omit it to search everything. Returns [] on no match.
    """
    if not WORD.search(query.lower()):
        raise ToolError(f"Query {query!r} has no searchable words. Use keywords like 'retry backoff'.")
    hits = keyword_search(query, k, applies_to)
    return [DocHit(doc_id=h["doc_id"], title=h["title"], section=h["section"], snippet=h["snippet"])
            for h in hits]

@mcp.tool(annotations=READ_ONLY)
def ask(question: Annotated[str, Field(min_length=3, max_length=500)]) -> Answer:
    """Answer a question about Orbit from its docs, with citations (doc_id + exact quote).

    Makes one model call (well under $0.001 on gpt-6-luna). Prefer search_docs when page ids
    are enough. degraded=true means no model was available and the citations are raw passages.
    """
    # Adapter seam: finished Capstone R2? Replace keyword_search/answer_question with your
    # Index.hybrid_search() and ask(). Keep the tool name, arguments and Answer shape: Day 61 depends on them.
    return answer_question(question, keyword_search(question, k=3))

if __name__ == "__main__":
    if "--http" in sys.argv:  # loopback only: no auth until Day 74
        print(f"Atlas MCP server {mcp.version} on http://127.0.0.1:8000/mcp", file=sys.stderr)
        mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
    else:
        mcp.run()  # stdio: stdout belongs to the protocol, so log to stderr

$ ORBIT_DATA=datasets/orbit python atlas_mcp_server.py --http
Atlas MCP server 1.0.0 on http://127.0.0.1:8000/mcp
INFO:     Started server process [41873]
INFO:     Waiting for application startup.
StreamableHTTP session manager started
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

# script_id: day_040_mcp_servers/test_atlas_mcp_server
# Contract and behaviour tests, in-memory. Install: uv add --dev pytest   Env: ORBIT_DATA
# Run: uv run pytest test_atlas_mcp_server.py -q   (UPDATE_SNAPSHOTS=1 rewrites tool_contract.json)
import json
import os
from pathlib import Path

import pytest
from mcp import Client, MCPError

from atlas_mcp_server import mcp

CONTRACT = Path(__file__).with_name("tool_contract.json")

@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"

@pytest.mark.anyio
async def test_search_returns_typed_hits() -> None:
    async with Client(mcp, raise_exceptions=True) as client:
        result = await client.call_tool("search_docs", {"query": "lease heartbeat", "k": 2})
    hits = result.structured_content["result"]
    assert not result.is_error and len(hits) == 2
    assert hits[0]["doc_id"] == "leases-and-heartbeats"

@pytest.mark.anyio
@pytest.mark.parametrize("args, expected", [
    ({"query": "lease", "k": 50}, "less than or equal to 10"),
    ({"query": "lease", "applies_to": "4.x"}, "'2.x', '3.x' or 'all'"),
])
async def test_bad_arguments_become_tool_errors(args: dict, expected: str) -> None:
    async with Client(mcp, raise_exceptions=True) as client:
        result = await client.call_tool("search_docs", args)
    assert result.is_error and expected in result.content[0].text

# script_id: day_040_mcp_servers/test_atlas_mcp_server
@pytest.mark.anyio
async def test_unknown_doc_suggests_a_real_id() -> None:
    async with Client(mcp, raise_exceptions=True) as client:
        with pytest.raises(MCPError, match="Closest: 'retries-and-backoff'"):
            await client.read_resource("orbit://docs/retries")

@pytest.mark.anyio
async def test_tool_contract_matches_snapshot() -> None:
    async with Client(mcp) as client:
        tools = (await client.list_tools()).tools
    current = {t.name: {"input": t.input_schema, "output": t.output_schema} for t in tools}
    if os.environ.get("UPDATE_SNAPSHOTS") or not CONTRACT.exists():
        CONTRACT.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n")
    expected = json.loads(CONTRACT.read_text())
    assert current == expected, "Tool contract changed: update the snapshot AND bump the version"

$ ORBIT_DATA=datasets/orbit uv run pytest test_atlas_mcp_server.py -q
.....                                                                    [100%]
5 passed in 0.37s