Coverage for src / documint_mcp / mcp_server.py: 0%
194 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-29 10:41 -0400
1"""
2Documint MCP Server -- exposes Documint tools to AI coding agents.
4Uses the high-level FastMCP API from the MCP Python SDK (>=1.0).
6Transports:
7 stdio (default) -- for Claude Code / local IDE integrations
8 sse -- for cloud-based agents (Manus, etc.)
9 streamable-http -- modern streamable HTTP transport
11Usage:
12 # stdio (default, for Claude Code)
13 documint-mcp
15 # SSE transport on port 8100
16 DOCUMINT_MCP_TRANSPORT=sse DOCUMINT_MCP_PORT=8100 documint-mcp
18 # Or via CLI flag
19 documint-mcp --transport sse --port 8100
21ASGI mounting (for embedding in the FastAPI control-plane):
22 from documint_mcp.mcp_server import mcp
23 app.mount("/mcp-sse", mcp.sse_app())
25Configure in Claude Code ~/.claude/claude.json:
26{
27 "mcpServers": {
28 "documint": {
29 "command": "uvx",
30 "args": ["documint-mcp"],
31 "env": { "DOCUMINT_API_KEY": "dm_live_xxxx" }
32 }
33 }
34}
35"""
37from __future__ import annotations
39import argparse
40import base64
41import json
42import logging
43import os
44import re
45import urllib.parse
46from collections.abc import AsyncIterator
47from contextlib import asynccontextmanager
48from pathlib import Path
49from typing import Any
51import httpx
52from mcp.server.fastmcp import FastMCP
54_logger = logging.getLogger("documint.mcp")
55DEFAULT_API_BASE_URL = "https://api-production-285b.up.railway.app"
57# ---- API client -------------------------------------------------------------
60class DocumintClient:
61 """Thin async HTTP wrapper around the Documint REST API."""
63 def __init__(self, api_key: str, base_url: str = DEFAULT_API_BASE_URL) -> None:
64 self.api_key = api_key
65 self.base_url = base_url.rstrip("/")
66 self.headers = {
67 "Authorization": f"Bearer {api_key}",
68 "Content-Type": "application/json",
69 }
70 self._http = httpx.AsyncClient(
71 base_url=self.base_url,
72 headers=self.headers,
73 timeout=30,
74 )
76 async def get(self, path: str) -> Any:
77 resp = await self._http.get(path)
78 resp.raise_for_status()
79 return resp.json()
81 async def post(self, path: str, body: dict[str, Any] | None = None) -> Any:
82 resp = await self._http.post(path, json=body or {})
83 resp.raise_for_status()
84 return resp.json()
86 async def close(self) -> None:
87 await self._http.aclose()
90# ---- Lifespan (manages the HTTP client) -------------------------------------
92_client: DocumintClient | None = None
95def _get_client() -> DocumintClient:
96 """Return the shared DocumintClient, creating it on first call."""
97 global _client
98 if _client is None:
99 api_key = os.environ.get("DOCUMINT_API_KEY", "")
100 if not api_key:
101 raise RuntimeError("DOCUMINT_API_KEY environment variable not set")
102 base_url = os.environ.get("DOCUMINT_API_URL", DEFAULT_API_BASE_URL)
103 _client = DocumintClient(api_key=api_key, base_url=base_url)
104 return _client
107@asynccontextmanager
108async def _lifespan(server: FastMCP) -> AsyncIterator[None]: # noqa: ARG001
109 """Startup / shutdown hook -- ensures the HTTP client is closed cleanly."""
110 yield
111 if _client is not None:
112 await _client.close()
115# ---- Helpers -----------------------------------------------------------------
118def _pid(project_id: str) -> str:
119 """URL-encode a project ID for path segments."""
120 return urllib.parse.quote(project_id, safe="")
123def _paginate(data: Any, key: str, offset: int = 0, limit: int = 20) -> str:
124 """Extract a list from *data*, paginate, and return JSON."""
125 if isinstance(data, list):
126 items = data
127 elif isinstance(data, dict):
128 items = data.get(key, [])
129 else:
130 items = []
131 total = len(items)
132 items = items[offset : offset + limit]
133 return json.dumps({"total": total, "offset": offset, "limit": limit, "items": items})
136def _json(data: Any) -> str:
137 return json.dumps(data)
140# ---- FastMCP server ----------------------------------------------------------
142mcp = FastMCP(
143 "documint",
144 instructions=(
145 "Documint keeps documentation in sync with source code. "
146 "Typical workflow: list_projects -> get_findings -> get_patch -> approve_patch."
147 ),
148 lifespan=_lifespan,
149)
152# ---- Tools (15 total -- same names and schemas as before) --------------------
155@mcp.tool(
156 name="documint_list_projects",
157 description=(
158 "List all Documint projects connected to this API key. "
159 "Returns an array of project objects, each with: id (use as project_id in other tools), "
160 "name, slug, github_repo (owner/name), and onboarding_status. "
161 "Call this first when you need to discover project IDs. "
162 "Typical workflow: list_projects -> get_findings -> get_patch -> approve_patch."
163 ),
164)
165async def documint_list_projects(offset: int = 0, limit: int = 20) -> str:
166 """List all Documint projects connected to this API key."""
167 client = _get_client()
168 data = await client.get("/projects")
169 return _paginate(data, "projects", offset, limit)
172@mcp.tool(
173 name="documint_list_artifacts",
174 description=(
175 "List all artifact definitions for a project. Each artifact maps source files to a documentation file "
176 "(e.g., 'src/memory.py' -> 'docs/api/memory.md'). "
177 "Returns an array with: artifact_key (use in documint_get_mint), title, artifact_type "
178 "(api_reference | sdk_guides | mcp_reference | changelog | migration_notes), "
179 "doc_paths (docs being tracked), and source_patterns (source globs being watched). "
180 "Use artifact_key with documint_get_mint to read the structured machine-readable docs."
181 ),
182)
183async def documint_list_artifacts(project_id: str, offset: int = 0, limit: int = 20) -> str:
184 """List artifact definitions for a project."""
185 client = _get_client()
186 data = await client.get(f"/projects/{_pid(project_id)}/artifacts")
187 return _paginate(data, "artifacts", offset, limit)
190@mcp.tool(
191 name="documint_get_findings",
192 description=(
193 "List open documentation drift findings for a project. A finding means source code changed "
194 "but the corresponding documentation was not updated. "
195 "Returns an array of finding objects with: id (finding_id for other tools), artifact_id, "
196 "severity (high | medium | low), summary (human-readable description of the drift, e.g. "
197 "'add_memory() gained a new required parameter `ttl` not documented'), "
198 "has_breaking_changes (bool), changed_symbols (list of changed function/class names), "
199 "and suggested_actions. "
200 "Use finding.id with documint_get_patch and documint_approve_patch."
201 ),
202)
203async def documint_get_findings(project_id: str) -> str:
204 """List open documentation drift findings for a project."""
205 client = _get_client()
206 data = await client.get(f"/projects/{_pid(project_id)}/findings")
207 return _json(data)
210@mcp.tool(
211 name="documint_check_drift",
212 description=(
213 "Trigger a fresh documentation drift scan for a project. Compares current source code AST "
214 "signatures (SHA-256 of symbol hashes via tree-sitter) against tracked documentation. "
215 "This is an async operation: returns a job object with job_id and status. "
216 "The scan typically completes in 5-30 seconds. "
217 "After triggering, call documint_get_findings to retrieve the results -- findings are "
218 "updated once the scan job completes. "
219 "Use this when you want up-to-date drift data after a recent code change."
220 ),
221)
222async def documint_check_drift(project_id: str) -> str:
223 """Trigger a fresh documentation drift scan."""
224 client = _get_client()
225 data = await client.post(f"/projects/{_pid(project_id)}/rescan")
226 return _json(data)
229@mcp.tool(
230 name="documint_get_patch",
231 description=(
232 "Get (or generate on demand) the AI-drafted documentation patch for a specific drift finding. "
233 "If no patch exists yet, triggers the 4-step AI chain: "
234 "(1) Haiku structured diff report, (2) stale section detection, "
235 "(3) Sonnet surgical patch generation, (4) Haiku verification. "
236 "Returns a patch object with: id, preview_markdown (the full updated documentation), "
237 "summary, rationale, confidence_score (0.0-1.0), ai_provider, chain_steps_used, "
238 "and citations (source files used). "
239 "Always read preview_markdown before calling documint_approve_patch so you can "
240 "show the user what will be committed."
241 ),
242)
243async def documint_get_patch(project_id: str, finding_id: str) -> str:
244 """Get (or generate) the AI-drafted documentation patch for a drift finding."""
245 client = _get_client()
246 pid = _pid(project_id)
247 fid = urllib.parse.quote(finding_id, safe="")
248 data = await client.get(f"/projects/{pid}/findings/{fid}/patch")
249 return _json(data)
252@mcp.tool(
253 name="documint_approve_patch",
254 description=(
255 "Approve an AI-drafted patch and open a GitHub pull request with the documentation fix. "
256 "The PR is created on the project's connected GitHub repository targeting the default branch. "
257 "Returns a pull_request object with: id, branch_name, title, url (GitHub PR URL), and state. "
258 "IMPORTANT: Always call documint_get_patch first and present preview_markdown to the user "
259 "for review before calling this tool. Approval is irreversible -- it creates a real PR. "
260 "Requires the project to have a connected GitHub installation."
261 ),
262)
263async def documint_approve_patch(project_id: str, finding_id: str) -> str:
264 """Approve an AI-drafted patch and open a GitHub PR."""
265 client = _get_client()
266 pid = _pid(project_id)
267 fid = urllib.parse.quote(finding_id, safe="")
268 data = await client.post(f"/projects/{pid}/findings/{fid}/approve")
269 return _json(data)
272@mcp.tool(
273 name="documint_get_mint",
274 description=(
275 "Get the .mint file (machine-readable documentation) for a specific artifact. "
276 "The .mint format is a structured JSON representation optimised for LLM consumption: "
277 "it includes symbol inventory (all documented functions/classes with their signatures), "
278 "section graph, cross-references, and verification metadata. "
279 "Use this when you need to understand what an artifact currently documents "
280 "without reading raw markdown. "
281 "artifact_key comes from documint_list_artifacts."
282 ),
283)
284async def documint_get_mint(project_id: str, artifact_key: str) -> str:
285 """Get the .mint file for a specific artifact."""
286 client = _get_client()
287 pid = _pid(project_id)
288 akey = urllib.parse.quote(artifact_key, safe="")
289 data = await client.get(f"/projects/{pid}/artifacts/{akey}/mint")
290 return _json(data)
293@mcp.tool(
294 name="documint_get_claude_md",
295 description=(
296 "Get the CLAUDE.md context file for a project. "
297 "CLAUDE.md is a curated system-prompt insert that helps Claude-family LLMs understand "
298 "the project's conventions, key modules, and coding patterns. "
299 "Returns an object with a 'content' field containing the markdown text. "
300 "Recommended: load this at the start of any session working on this project's codebase."
301 ),
302)
303async def documint_get_claude_md(project_id: str) -> str:
304 """Get the CLAUDE.md context file for a project."""
305 client = _get_client()
306 data = await client.get(f"/projects/{_pid(project_id)}/claude-md")
307 return _json(data)
310@mcp.tool(
311 name="documint_get_agents_md",
312 description=(
313 "Get the AGENTS.md guide for a project. "
314 "AGENTS.md describes the AI agent roles, tool permissions, escalation rules, "
315 "and task boundaries for this project's multi-agent setup. "
316 "Returns an object with a 'content' field containing the markdown text. "
317 "Load this when orchestrating sub-agents or deciding which agent should handle a task."
318 ),
319)
320async def documint_get_agents_md(project_id: str) -> str:
321 """Get the AGENTS.md guide for a project."""
322 client = _get_client()
323 data = await client.get(f"/projects/{_pid(project_id)}/agents-md")
324 return _json(data)
327@mcp.tool(
328 name="documint_get_llms_txt",
329 description=(
330 "Get the llms.txt index file for a project (per the llmstxt.org specification). "
331 "llms.txt is a plain-text index of all project documentation with short summaries "
332 "and URLs, optimised for LLM context windows. Use it to discover what documentation "
333 "exists and decide which pages to fetch for a user's question. "
334 "Returns an object with a 'content' field containing the plain text."
335 ),
336)
337async def documint_get_llms_txt(project_id: str) -> str:
338 """Get the llms.txt index file for a project."""
339 client = _get_client()
340 data = await client.get(f"/projects/{_pid(project_id)}/llms-txt")
341 return _json(data)
344@mcp.tool(
345 name="documint_get_symbol_diff",
346 description=(
347 "Get a detailed symbol-level diff for a specific drift finding. "
348 "Shows exactly what changed in the source code signatures vs. what the documentation says, "
349 "including: symbol_name, change_type (ADDED | REMOVED | PARAM_ADDED | PARAM_REMOVED | "
350 "RETURN_CHANGED | SIGNATURE_CHANGED), severity (BREAKING | ADDITIVE | COSMETIC), "
351 "and before/after signature strings. "
352 "Use this before generating or reviewing a patch to understand the exact technical delta. "
353 "More precise than the finding summary alone."
354 ),
355)
356async def documint_get_symbol_diff(project_id: str, finding_id: str) -> str:
357 """Get a symbol-level diff for a drift finding."""
358 client = _get_client()
359 pid = _pid(project_id)
360 fid = urllib.parse.quote(finding_id, safe="")
361 data = await client.get(f"/projects/{pid}/findings/{fid}/symbol-diff")
362 return _json(data)
365@mcp.tool(
366 name="documint_status",
367 description=(
368 "Get a health summary for a project. "
369 "Returns: drift_status (clean | drifted | unknown), open_findings_count (int), "
370 "last_scan_at (ISO timestamp of most recent drift scan), "
371 "onboarding_status (connected | pending | error), "
372 "and a short human-readable health_message. "
373 "Use this at the start of a session to decide whether to run documint_check_drift "
374 "or go straight to documint_get_findings."
375 ),
376)
377async def documint_status(project_id: str) -> str:
378 """Get a health summary for a project."""
379 client = _get_client()
380 data = await client.get(f"/projects/{_pid(project_id)}/status")
381 return _json(data)
384@mcp.tool(
385 name="documint_watch_artifact",
386 description=(
387 "Returns the current drift status, last symbol hash, and staleness age for an artifact. "
388 "Use this to check if documentation is stale before proposing changes. "
389 "Returns: {artifact_id, verification_status, last_symbol_hash, stale_age_hours, "
390 "findings_count, last_checked}"
391 ),
392)
393async def documint_watch_artifact(project_id: str, artifact_id: str) -> str:
394 """Get drift status for a specific artifact."""
395 client = _get_client()
396 pid = _pid(project_id)
397 aid = urllib.parse.quote(artifact_id, safe="")
398 trace = await client.get(f"/projects/{pid}/artifacts/{aid}/trace")
399 result = {
400 "artifact_id": trace.get("id") if isinstance(trace, dict) else artifact_id,
401 "verification_status": trace.get("verification_status") if isinstance(trace, dict) else None,
402 "latest_source_ref": (
403 trace.get("latest_source_revision", {}).get("ref")
404 if isinstance(trace, dict) and trace.get("latest_source_revision")
405 else None
406 ),
407 "latest_doc_ref": (
408 trace.get("latest_doc_revision", {}).get("ref")
409 if isinstance(trace, dict) and trace.get("latest_doc_revision")
410 else None
411 ),
412 "is_stale": (
413 trace.get("verification_status") == "stale"
414 if isinstance(trace, dict)
415 else None
416 ),
417 "doc_paths": trace.get("doc_paths") if isinstance(trace, dict) else None,
418 "source_paths": trace.get("source_paths") if isinstance(trace, dict) else None,
419 }
420 return _json(result)
423@mcp.tool(
424 name="documint_get_coverage",
425 description=(
426 "Returns documentation coverage report for a project -- what percentage of symbols "
427 "are documented per artifact. "
428 "Returns: {project_id, artifacts: [{artifact_id, coverage_pct, documented, total}], "
429 "overall_pct}"
430 ),
431)
432async def documint_get_coverage(project_id: str) -> str:
433 """Get documentation coverage report for a project."""
434 client = _get_client()
435 data = await client.get(f"/projects/{_pid(project_id)}/coverage")
436 return _json(data)
439# ---- Resources (4 URI schemes -- dynamic per-project) ------------------------
440#
441# The FastMCP @resource() decorator uses URI *templates* with {param} syntax.
442# These are registered as MCP ResourceTemplates and resolved on read.
445@mcp.resource(
446 "claude-md://{project_id}",
447 name="CLAUDE.md",
448 description="CLAUDE.md context file for a project (curated system-prompt insert for Claude-family LLMs).",
449 mime_type="text/markdown",
450)
451async def read_claude_md(project_id: str) -> str:
452 """Read the CLAUDE.md resource for a project."""
453 client = _get_client()
454 pid = _pid(project_id)
455 data = await client.get(f"/projects/{pid}/claude-md")
456 return data.get("content", _json(data)) if isinstance(data, dict) else _json(data)
459@mcp.resource(
460 "agents-md://{project_id}",
461 name="AGENTS.md",
462 description="AGENTS.md guide describing AI agent roles, permissions, and task boundaries.",
463 mime_type="text/markdown",
464)
465async def read_agents_md(project_id: str) -> str:
466 """Read the AGENTS.md resource for a project."""
467 client = _get_client()
468 pid = _pid(project_id)
469 data = await client.get(f"/projects/{pid}/agents-md")
470 return data.get("content", _json(data)) if isinstance(data, dict) else _json(data)
473@mcp.resource(
474 "llms-txt://{project_id}",
475 name="llms.txt",
476 description="llms.txt index of all project documentation with summaries and URLs (llmstxt.org spec).",
477 mime_type="text/plain",
478)
479async def read_llms_txt(project_id: str) -> str:
480 """Read the llms.txt resource for a project."""
481 client = _get_client()
482 pid = _pid(project_id)
483 data = await client.get(f"/projects/{pid}/llms-txt")
484 return data.get("content", _json(data)) if isinstance(data, dict) else _json(data)
487@mcp.resource(
488 "mint-binary://{project_id}/{artifact_id}",
489 name=".mint binary",
490 description="Binary .mint artifact file for a project (base64-encoded).",
491 mime_type="application/octet-stream",
492)
493async def read_mint_binary(project_id: str, artifact_id: str) -> str: # noqa: ARG001 (project_id required by URI template)
494 """Read a local .mint binary file for an artifact."""
495 if not re.fullmatch(r"[A-Za-z0-9_-]+", artifact_id):
496 return _json({"error": "Invalid artifact_id: must contain only alphanumeric characters, hyphens, and underscores."})
497 docs_root = os.environ.get("DOCUMINT_DOCS_ROOT", ".")
498 expected_parent = (Path(docs_root) / ".documint").resolve()
499 mint_path = Path(docs_root) / ".documint" / f"{artifact_id}.mint"
500 if not mint_path.resolve().is_relative_to(expected_parent):
501 return _json({"error": "Invalid artifact path: path traversal detected."})
502 if mint_path.exists():
503 raw = mint_path.read_bytes()
504 return _json({
505 "artifact_id": artifact_id,
506 "encoding": "base64",
507 "data": base64.b64encode(raw).decode(),
508 })
509 return _json({"error": "No .mint file found. Run 'documint publish' first."})
512# ---- Prompt (documint workflow guide) ----------------------------------------
515@mcp.prompt(
516 name="documint_workflow",
517 title="Documint Drift-Fix Workflow",
518 description="Step-by-step guide for using Documint to find and fix documentation drift.",
519)
520def documint_workflow(project_id: str = "") -> str:
521 """Return a workflow prompt for fixing documentation drift."""
522 pid_hint = f" (project: {project_id})" if project_id else ""
523 return (
524 f"Follow this workflow to fix documentation drift{pid_hint}:\n\n"
525 "1. Call `documint_list_projects` to discover available projects and their IDs.\n"
526 "2. Call `documint_status` with the project_id to check whether the project has drift.\n"
527 "3. If drift_status is 'unknown' or stale, call `documint_check_drift` to trigger a fresh scan.\n"
528 "4. Call `documint_get_findings` to list all open drift findings.\n"
529 "5. For each finding, call `documint_get_symbol_diff` to understand what changed.\n"
530 "6. Call `documint_get_patch` to get (or generate) the AI-drafted fix.\n"
531 "7. Review the patch's `preview_markdown` with the user.\n"
532 "8. If approved, call `documint_approve_patch` to open a GitHub PR.\n\n"
533 "Tips:\n"
534 "- Use `documint_get_claude_md` at session start for project context.\n"
535 "- Use `documint_get_coverage` to understand documentation completeness.\n"
536 "- Use `documint_watch_artifact` to check staleness of specific artifacts.\n"
537 )
540# ---- ASGI mounting helper for FastAPI integration ----------------------------
543def mount_on_fastapi(app: Any, path: str = "/mcp-sse") -> None:
544 """Mount the Documint MCP server as SSE on an existing FastAPI/Starlette app.
546 Usage in server.py:
547 from documint_mcp.mcp_server import mount_on_fastapi
548 app = create_app()
549 mount_on_fastapi(app, "/mcp-sse")
551 Clients connect via:
552 SSE endpoint: GET {path}/sse
553 Message endpoint: POST {path}/messages/
554 """
555 sse_starlette = mcp.sse_app()
556 app.mount(path, sse_starlette)
559def get_streamable_http_app() -> Any:
560 """Return a Starlette ASGI app for streamable-http transport.
562 Usage:
563 from documint_mcp.mcp_server import get_streamable_http_app
564 app.mount("/mcp", get_streamable_http_app())
565 """
566 return mcp.streamable_http_app()
569# ---- Entry point -------------------------------------------------------------
572def main() -> None:
573 """CLI entry point for the Documint MCP server.
575 Supports transport selection via --transport flag or DOCUMINT_MCP_TRANSPORT env var.
576 """
577 parser = argparse.ArgumentParser(
578 prog="documint-mcp",
579 description="Documint MCP server for AI coding agents",
580 )
581 parser.add_argument(
582 "--transport",
583 choices=["stdio", "sse", "streamable-http"],
584 default=os.environ.get("DOCUMINT_MCP_TRANSPORT", "stdio"),
585 help="Transport protocol (default: stdio, or set DOCUMINT_MCP_TRANSPORT env var)",
586 )
587 parser.add_argument(
588 "--host",
589 default=os.environ.get("DOCUMINT_MCP_HOST", "127.0.0.1"),
590 help="Host to bind for SSE/HTTP transport (default: 127.0.0.1)",
591 )
592 parser.add_argument(
593 "--port",
594 type=int,
595 default=int(os.environ.get("DOCUMINT_MCP_PORT", "8100")),
596 help="Port for SSE/HTTP transport (default: 8100)",
597 )
598 args = parser.parse_args()
600 # Configure FastMCP settings for network transports
601 if args.transport in ("sse", "streamable-http"):
602 mcp.settings.host = args.host
603 mcp.settings.port = args.port
605 _logger.info(
606 "Starting Documint MCP server (transport=%s, host=%s, port=%d)",
607 args.transport,
608 args.host,
609 args.port,
610 )
612 mcp.run(transport=args.transport)
615if __name__ == "__main__":
616 main()