diff --git a/docs/mcp.md b/docs/mcp.md index 10cdeb0..9519077 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -1,3 +1,52 @@ -# mcp +# MCP server -_Written in the docs phase._ +The catalog ships as an MCP server so agents can search the docs, read the +reference code, execute the examples, and get pattern recommendations with +honest verdicts attached. + +## Connect + +From a checkout: + +```bash +claude mcp add design-patterns -- uv run --directory /path/to/python-design-patterns python-design-patterns-mcp +``` + +Once published to PyPI: + +```bash +claude mcp add design-patterns -- uvx python-design-patterns-mcp +``` + +Remote/HTTP (Streamable HTTP on `/mcp`): + +```bash +python-design-patterns-mcp --http --host 127.0.0.1 --port 8734 +``` + +## Tools + +| Tool | What it does | +|---|---| +| `list_patterns(group?, verdict?)` | Catalog listing, filterable | +| `get_pattern(pattern_id, variant?)` | Full prose + example source (`naive`/`pythonic`/`real_world`/`all`) | +| `search_patterns(query, limit?)` | BM25 full-text search over names, aliases, problems, symptoms, prose | +| `run_example(pattern_id, variant)` | Executes the vendored example in a sandboxed subprocess; returns real stdout | +| `recommend_pattern(problem_statement, limit?)` | Ranked candidates with caveats; `prefer-alternative` verdicts tell you what to write instead | + +## Resources + +- `catalog://index` — the whole catalog as JSON +- `pattern:///` — one pattern's prose +- `pattern:////` — one example's source + +## Prompts + +`refactor_toward(pattern_id, code)` · `explain_pattern(pattern_id, audience?)` · `choose_pattern(problem)` + +## Sandbox contract + +`run_example` executes only files resolved from the catalog index — the +`(id, variant)` pair is a dictionary lookup, never joined into a path. The +subprocess runs `python -I` in a temp cwd with a scrubbed environment, a 10s +timeout, and 64KB output caps. There is no arbitrary-code-execution tool. diff --git a/pyproject.toml b/pyproject.toml index 2ab0d70..104c09a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ classifiers = [ "Topic :: Education", ] dependencies = [ + "mcp>=2.1.1", "pyyaml>=6.0", ] @@ -25,6 +26,9 @@ dependencies = [ Homepage = "https://github.com/SuperElectron/python-design-patterns" Reference = "https://python-patterns.guide/" +[project.scripts] +python-design-patterns-mcp = "design_patterns_mcp.server:main" + [dependency-groups] dev = [ "pytest>=8.0", @@ -40,7 +44,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["src/design_patterns"] +packages = ["src/design_patterns", "src/design_patterns_mcp"] [tool.ruff] line-length = 100 diff --git a/src/design_patterns_mcp/__init__.py b/src/design_patterns_mcp/__init__.py new file mode 100644 index 0000000..07b6c88 --- /dev/null +++ b/src/design_patterns_mcp/__init__.py @@ -0,0 +1 @@ +"""MCP server exposing the pattern catalog to agents.""" diff --git a/src/design_patterns_mcp/sandbox.py b/src/design_patterns_mcp/sandbox.py new file mode 100644 index 0000000..a997abd --- /dev/null +++ b/src/design_patterns_mcp/sandbox.py @@ -0,0 +1,63 @@ +"""Sandboxed execution of catalog example files -- and nothing else. + +The contract: only paths resolved from the catalog index are runnable. +The (id, variant) pair is looked up, never joined into a path, so there is +no traversal and no arbitrary-file execution surface. +""" + +from __future__ import annotations + +import subprocess +import sys +import tempfile +from dataclasses import dataclass + +from design_patterns.catalog import Catalog + +TIMEOUT_SECONDS = 10 +MAX_OUTPUT_BYTES = 64 * 1024 + + +@dataclass(frozen=True) +class RunResult: + exit_code: int + stdout: str + stderr: str + timed_out: bool = False + + +def run_example(catalog: Catalog, pattern_id: str, variant: str) -> RunResult: + """Execute one vendored example in a subprocess and capture its output.""" + pattern = catalog.get(pattern_id) # KeyError for unknown ids -- by design + variants = pattern.variants() + if variant not in variants: + raise KeyError(f"{pattern_id} has no variant {variant!r} (has: {sorted(variants)})") + path = variants[variant] # resolved by the catalog, never by the caller + + repo_root = pattern.path.parents[2] + module = f"patterns.{pattern.group}.{pattern.slug}.{variant}" + # -I ignores PYTHONPATH by design, so the repo root (resolved by the + # catalog, never by the caller) is injected in the bootstrap itself. + bootstrap = ( + f"import sys, runpy; sys.path.insert(0, {str(repo_root)!r}); " + f"runpy.run_module({module!r}, run_name='__main__')" + ) + with tempfile.TemporaryDirectory() as scratch_cwd: + try: + completed = subprocess.run( + [sys.executable, "-I", "-c", bootstrap], + cwd=scratch_cwd, + env={}, # scrubbed environment + capture_output=True, + timeout=TIMEOUT_SECONDS, + text=True, + ) + except subprocess.TimeoutExpired as exc: + out = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "") + return RunResult(exit_code=-1, stdout=out[:MAX_OUTPUT_BYTES], stderr="", timed_out=True) + assert path.is_file() + return RunResult( + exit_code=completed.returncode, + stdout=completed.stdout[:MAX_OUTPUT_BYTES], + stderr=completed.stderr[:MAX_OUTPUT_BYTES], + ) diff --git a/src/design_patterns_mcp/search.py b/src/design_patterns_mcp/search.py new file mode 100644 index 0000000..93ef3b6 --- /dev/null +++ b/src/design_patterns_mcp/search.py @@ -0,0 +1,78 @@ +"""BM25 search over the pattern catalog. Pure stdlib, no network.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass + +from design_patterns.catalog import Catalog, Pattern + +_TOKEN = re.compile(r"[a-z0-9']+") + +# Field weights: a hit in the problem statement or symptoms should count +# for more than a hit deep in the prose. +_WEIGHTS = {"name": 4.0, "aliases": 4.0, "problem": 3.0, "symptoms": 3.0, "prose": 1.0} + + +def _tokenize(text: str) -> list[str]: + return _TOKEN.findall(text.lower()) + + +def _document(pattern: Pattern) -> list[str]: + tokens: list[str] = [] + fields = { + "name": pattern.name, + "aliases": " ".join(pattern.aliases), + "problem": pattern.problem, + "symptoms": " ".join(pattern.symptoms), + "prose": pattern.prose, + } + for field, text in fields.items(): + weight = int(_WEIGHTS[field]) + tokens.extend(_tokenize(text) * weight) + return tokens + + +@dataclass(frozen=True) +class Hit: + pattern: Pattern + score: float + + +class SearchIndex: + """A small BM25 index over every unit's frontmatter and prose.""" + + K1 = 1.5 + B = 0.75 + + def __init__(self, catalog: Catalog) -> None: + self._patterns = catalog.patterns + self._docs = [_document(p) for p in self._patterns] + self._doc_lens = [len(d) for d in self._docs] + self._avg_len = sum(self._doc_lens) / len(self._docs) + self._freqs = [{t: doc.count(t) for t in set(doc)} for doc in self._docs] + self._df: dict[str, int] = {} + for freq in self._freqs: + for term in freq: + self._df[term] = self._df.get(term, 0) + 1 + + def _idf(self, term: str) -> float: + n, df = len(self._docs), self._df.get(term, 0) + return math.log(1 + (n - df + 0.5) / (df + 0.5)) + + def search(self, query: str, limit: int = 5) -> list[Hit]: + terms = _tokenize(query) + hits: list[Hit] = [] + for i, pattern in enumerate(self._patterns): + score = 0.0 + for term in terms: + tf = self._freqs[i].get(term, 0) + if tf == 0: + continue + norm = self.K1 * (1 - self.B + self.B * self._doc_lens[i] / self._avg_len) + score += self._idf(term) * tf * (self.K1 + 1) / (tf + norm) + if score > 0: + hits.append(Hit(pattern, round(score, 3))) + hits.sort(key=lambda h: h.score, reverse=True) + return hits[:limit] diff --git a/src/design_patterns_mcp/server.py b/src/design_patterns_mcp/server.py new file mode 100644 index 0000000..29a2077 --- /dev/null +++ b/src/design_patterns_mcp/server.py @@ -0,0 +1,205 @@ +"""The python-design-patterns MCP server. + +Tools, resources, and prompts over the pattern catalog. Run over stdio by +default (``python-design-patterns-mcp``) or streamable HTTP (``--http``). +""" + +from __future__ import annotations + +import argparse +from typing import Any + +from mcp.server import MCPServer + +from design_patterns.catalog import Catalog, Pattern, load_catalog +from design_patterns_mcp.sandbox import run_example as _run_example +from design_patterns_mcp.search import SearchIndex + +_catalog: Catalog = load_catalog() +_index = SearchIndex(_catalog) + +mcp = MCPServer( + "python-design-patterns", + instructions=( + "Design patterns in Python: 32 units covering all 23 GoF patterns, " + "Python-native patterns, and modern additions. Each unit has prose, a " + "naive (GoF-literal) example, a pythonic example, a real_world stdlib " + "sighting, and an honest verdict. Start with search_patterns or " + "recommend_pattern; verdicts of 'prefer-alternative' tell you what to " + "write instead." + ), +) + + +def _summary(pattern: Pattern) -> dict[str, Any]: + return { + "id": pattern.id, + "name": pattern.name, + "problem": pattern.problem, + "verdict": pattern.verdict, + } + + +def _detail(pattern: Pattern, include_source: str | None) -> dict[str, Any]: + detail: dict[str, Any] = { + **_summary(pattern), + "aliases": list(pattern.aliases), + "guide_url": pattern.guide_url, + "symptoms": list(pattern.symptoms), + "caveats": list(pattern.caveats), + "stdlib_sightings": list(pattern.stdlib_sightings), + "variants": sorted(pattern.variants()), + "prose": pattern.prose, + } + if include_source: + variants = pattern.variants() + wanted = sorted(variants) if include_source == "all" else [include_source] + detail["source"] = {name: variants[name].read_text() for name in wanted if name in variants} + return detail + + +@mcp.tool() +def list_patterns(group: str | None = None, verdict: str | None = None) -> list[dict[str, Any]]: + """List catalog patterns, optionally filtered by group (creational, + structural, behavioral, python, principle, modern) or verdict + (pythonic, use-with-care, prefer-alternative).""" + patterns = _catalog.patterns + if group is not None: + patterns = tuple(p for p in patterns if p.group == group) + if verdict is not None: + patterns = tuple(p for p in patterns if p.verdict == verdict) + return [_summary(p) for p in patterns] + + +@mcp.tool() +def get_pattern(pattern_id: str, variant: str | None = None) -> dict[str, Any]: + """Fetch one pattern's full documentation. pattern_id is '/' + (e.g. 'structural/decorator'). variant: 'naive', 'pythonic', 'real_world', + or 'all' to include example source code.""" + try: + pattern = _catalog.get(pattern_id) + except KeyError: + known = ", ".join(_catalog.ids()) + raise ValueError(f"unknown pattern {pattern_id!r}; known ids: {known}") from None + return _detail(pattern, variant) + + +@mcp.tool() +def search_patterns(query: str, limit: int = 5) -> list[dict[str, Any]]: + """Full-text search across pattern names, aliases, problems, symptoms, + and prose. Returns the best matches with scores.""" + return [{**_summary(h.pattern), "score": h.score} for h in _index.search(query, limit)] + + +@mcp.tool() +def run_example(pattern_id: str, variant: str) -> dict[str, Any]: + """Execute one of a pattern's vendored example files ('naive', 'pythonic', + 'real_world') in a sandboxed subprocess and return its real output.""" + result = _run_example(_catalog, pattern_id, variant) + return { + "exit_code": result.exit_code, + "stdout": result.stdout, + "stderr": result.stderr, + "timed_out": result.timed_out, + } + + +@mcp.tool() +def recommend_pattern(problem_statement: str, limit: int = 3) -> list[dict[str, Any]]: + """Describe a design problem in plain words; get ranked candidate patterns, + each with its caveats and verdict attached. A 'prefer-alternative' verdict + means the pythonic variant shows what to write instead.""" + recommendations = [] + for hit in _index.search(problem_statement, limit): + p = hit.pattern + rec = { + **_summary(p), + "score": hit.score, + "caveats": list(p.caveats), + "stdlib_sightings": list(p.stdlib_sightings), + } + if p.verdict == "prefer-alternative": + rec["note"] = ( + f"The guide's honest answer is usually not {p.name}: " + f"see this unit's pythonic.py for what to write instead." + ) + recommendations.append(rec) + return recommendations + + +@mcp.resource("catalog://index") +def catalog_index() -> str: + """The whole catalog as JSON: every pattern's metadata and variants.""" + return _catalog.to_json() + + +@mcp.resource("pattern://{group}/{slug}") +def pattern_doc(group: str, slug: str) -> str: + """One pattern's README prose.""" + return _catalog.get(f"{group}/{slug}").prose + + +@mcp.resource("pattern://{group}/{slug}/{variant}") +def pattern_source(group: str, slug: str, variant: str) -> str: + """One pattern's example source (naive | pythonic | real_world).""" + pattern = _catalog.get(f"{group}/{slug}") + variants = pattern.variants() + if variant not in variants: + raise KeyError(f"{pattern.id} has no variant {variant!r}") + return variants[variant].read_text() + + +@mcp.prompt() +def refactor_toward(pattern_id: str, code: str) -> str: + """Ask for a refactor of the given code toward one catalog pattern.""" + pattern = _catalog.get(pattern_id) + caveats = "\n".join(f"- {c}" for c in pattern.caveats) + return ( + f"Refactor the following code toward the {pattern.name} pattern " + f"({pattern.id}), as done in this catalog's pythonic variant.\n" + f"Verdict for this pattern: {pattern.verdict}. Honor these caveats:\n" + f"{caveats}\n\nCode:\n```python\n{code}\n```" + ) + + +@mcp.prompt() +def explain_pattern(pattern_id: str, audience: str = "an intermediate Python developer") -> str: + """Ask for an explanation of one pattern, tuned to an audience.""" + pattern = _catalog.get(pattern_id) + return ( + f"Explain the {pattern.name} pattern to {audience}. Problem it solves: " + f"{pattern.problem} Use the catalog's naive-vs-pythonic contrast, state " + f"the verdict ({pattern.verdict}) plainly, and show where the stdlib " + f"already uses it ({', '.join(pattern.stdlib_sightings)})." + ) + + +@mcp.prompt() +def choose_pattern(problem: str) -> str: + """Ask which pattern (if any!) fits a described problem.""" + return ( + f"A developer describes this problem:\n\n{problem}\n\n" + "Using the python-design-patterns catalog (search_patterns / " + "recommend_pattern), name the best-fitting pattern or say plainly that " + "no pattern is needed. If the top candidate's verdict is " + "'prefer-alternative', recommend the alternative its pythonic variant " + "shows instead." + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description="python-design-patterns MCP server") + parser.add_argument( + "--http", action="store_true", help="serve streamable HTTP instead of stdio" + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8734) + args = parser.parse_args() + if args.http: + mcp.run(transport="streamable-http", host=args.host, port=args.port) + else: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 0000000..d0db7a5 --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,108 @@ +"""MCP server integration tests: driven through an in-memory client session.""" + +import json + +from mcp import Client +from mcp.types import TextResourceContents + +from design_patterns_mcp.server import mcp + + +class TestTools: + async def test_list_patterns_returns_whole_catalog(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("list_patterns", {}) + assert result.structured_content is not None + patterns = result.structured_content["result"] + assert len(patterns) == 32 + assert {"id", "name", "problem", "verdict"} <= patterns[0].keys() + + async def test_list_patterns_filters(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("list_patterns", {"group": "creational"}) + assert result.structured_content is not None + ids = [p["id"] for p in result.structured_content["result"]] + assert len(ids) == 5 and all(i.startswith("creational/") for i in ids) + + async def test_get_pattern_with_source(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "get_pattern", {"pattern_id": "structural/decorator", "variant": "pythonic"} + ) + assert result.structured_content is not None + detail = result.structured_content + assert detail["verdict"] == "pythonic" + assert "functools" in detail["source"]["pythonic"] + + async def test_get_pattern_unknown_id_names_the_catalog(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool("get_pattern", {"pattern_id": "nope/nothing"}) + assert result.is_error + + async def test_search_finds_singleton_from_symptoms(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "search_patterns", {"query": "only one shared config instance"} + ) + assert result.structured_content is not None + ids = [h["id"] for h in result.structured_content["result"]] + assert "creational/singleton" in ids + + async def test_run_example_returns_real_output(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "run_example", {"pattern_id": "creational/singleton", "variant": "pythonic"} + ) + assert result.structured_content is not None + run = result.structured_content + assert run["exit_code"] == 0 and not run["timed_out"] + assert "module global is shared" in run["stdout"] + + async def test_recommend_attaches_caveats_and_alternative_note(self) -> None: + async with Client(mcp) as client: + result = await client.call_tool( + "recommend_pattern", + {"problem_statement": "I want a class with only one instance, a singleton"}, + ) + assert result.structured_content is not None + recs = result.structured_content["result"] + singleton = next(r for r in recs if r["id"] == "creational/singleton") + assert singleton["verdict"] == "prefer-alternative" + assert "pythonic.py" in singleton["note"] + assert singleton["caveats"] + + +class TestResources: + async def test_catalog_index_resource(self) -> None: + async with Client(mcp) as client: + result = await client.read_resource("catalog://index") + contents = result.contents[0] + assert isinstance(contents, TextResourceContents) + assert len(json.loads(contents.text)) == 32 + + async def test_pattern_doc_and_source_templates(self) -> None: + async with Client(mcp) as client: + doc = await client.read_resource("pattern://behavioral/iterator") + first = doc.contents[0] + assert isinstance(first, TextResourceContents) + assert "# Iterator" in first.text + + src = await client.read_resource("pattern://behavioral/iterator/naive") + first_src = src.contents[0] + assert isinstance(first_src, TextResourceContents) + assert "__next__" in first_src.text + + +class TestPrompts: + async def test_prompts_are_listed_and_render(self) -> None: + async with Client(mcp) as client: + listed = await client.list_prompts() + names = {p.name for p in listed.prompts} + assert {"refactor_toward", "explain_pattern", "choose_pattern"} <= names + + prompt = await client.get_prompt( + "refactor_toward", + {"pattern_id": "creational/singleton", "code": "class Config: pass"}, + ) + text = prompt.messages[0].content + assert "prefer-alternative" in str(text) diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py new file mode 100644 index 0000000..57b99e2 --- /dev/null +++ b/tests/test_sandbox.py @@ -0,0 +1,48 @@ +"""Sandbox contract: only catalog files run; timeouts and bad ids refuse.""" + +import pytest + +from design_patterns.catalog import load_catalog +from design_patterns_mcp.sandbox import run_example + +CATALOG = load_catalog() + + +class TestSandbox: + def test_runs_a_real_example(self) -> None: + result = run_example(CATALOG, "structural/flyweight", "pythonic") + assert result.exit_code == 0 + assert "shares" in result.stdout + assert not result.timed_out + + def test_unknown_pattern_id_is_refused(self) -> None: + with pytest.raises(KeyError): + run_example(CATALOG, "../../etc/passwd", "naive") + + def test_unknown_variant_is_refused(self) -> None: + with pytest.raises(KeyError, match="no variant"): + run_example(CATALOG, "structural/flyweight", "__init__") + + def test_traversal_shaped_variant_is_refused(self) -> None: + with pytest.raises(KeyError): + run_example(CATALOG, "structural/flyweight", "../../../tmp/evil") + + def test_failing_example_reports_not_raises(self) -> None: + # every current example exits 0; simulate by checking the API shape + result = run_example(CATALOG, "behavioral/command", "real_world") + assert isinstance(result.exit_code, int) + assert isinstance(result.stderr, str) + + +class TestSearchIndex: + def test_symptom_search_hits_the_right_unit(self) -> None: + from design_patterns_mcp.search import SearchIndex + + index = SearchIndex(CATALOG) + top = index.search("undo redo history snapshot", limit=3) + assert top and top[0].pattern.id in {"behavioral/memento", "behavioral/command"} + + def test_no_match_returns_empty(self) -> None: + from design_patterns_mcp.search import SearchIndex + + assert SearchIndex(CATALOG).search("zzzqqqxxx") == []