added mcp server code

This commit is contained in:
2026-01-21 17:08:55 +05:30
parent 5370a7faa2
commit 751bbe8949
5 changed files with 177 additions and 3 deletions

View File

@@ -192,6 +192,40 @@ def build(mkdocs_yml: Path) -> None:
click.echo("MkDocs build completed")
# ---------------------------------------------------------------------
# serve-mcp
# ---------------------------------------------------------------------
@cli.command(name="serve-mcp")
def serve_mcp() -> None:
"""
Serve MCP documentation from the local mcp_docs directory.
"""
from docforge.servers import MCPServer
mcp_root = Path.cwd() / "mcp_docs"
if not mcp_root.exists():
raise click.ClickException("mcp_docs directory not found")
required = [
mcp_root / "index.json",
mcp_root / "nav.json",
mcp_root / "modules",
]
for path in required:
if not path.exists():
raise click.ClickException(f"Invalid MCP bundle, missing: {path.name}")
server = MCPServer(
mcp_root=mcp_root,
name="doc-forge-mcp",
)
server.run()
# ---------------------------------------------------------------------
# serve
# ---------------------------------------------------------------------

View File

@@ -0,0 +1,5 @@
from .mcp_server import MCPServer
__all__ = [
"MCPServer",
]

View File

@@ -0,0 +1,73 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Literal
from mcp.server.fastmcp import FastMCP
class MCPServer:
"""
MCP server for serving a pre-built MCP documentation bundle.
"""
def __init__(self, mcp_root: Path, name: str) -> None:
self.mcp_root = mcp_root
self.app = FastMCP(name)
self._register_resources()
self._register_tools()
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _read_json(self, path: Path) -> Any:
if not path.exists():
return {
"error": "not_found",
"path": str(path),
}
return json.loads(path.read_text(encoding="utf-8"))
# ------------------------------------------------------------------
# MCP resources
# ------------------------------------------------------------------
def _register_resources(self) -> None:
@self.app.resource("docs://index")
def index():
return self._read_json(self.mcp_root / "index.json")
@self.app.resource("docs://nav")
def nav():
return self._read_json(self.mcp_root / "nav.json")
@self.app.resource("docs://module/{module}")
def module(module: str):
return self._read_json(
self.mcp_root / "modules" / f"{module}.json"
)
# ------------------------------------------------------------------
# MCP tools (optional / diagnostic)
# ------------------------------------------------------------------
def _register_tools(self) -> None:
@self.app.tool()
def ping() -> str:
return "pong"
# ------------------------------------------------------------------
# Server lifecycle
# ------------------------------------------------------------------
def run(self, transport: Literal["stdio", "sse", "streamable-http"] = "streamable-http") -> None:
"""
Start the MCP server.
Args:
transport: MCP transport (default: streamable-http)
"""
self.app.run(transport=transport)

View File

@@ -1,7 +1,7 @@
from pathlib import Path
from typing import Callable
import json
import pytest
from pathlib import Path
from click.testing import CliRunner
@@ -65,3 +65,47 @@ def mock_mkdocs_serve(monkeypatch):
fake_serve,
)
return lambda: called["value"]
@pytest.fixture
def fake_mcp_docs(tmp_path: Path) -> Path:
"""
Create a minimal valid MCP bundle in mcp_docs/.
"""
mcp_root = tmp_path / "mcp_docs"
modules_dir = mcp_root / "modules"
modules_dir.mkdir(parents=True)
(mcp_root / "index.json").write_text(
json.dumps({"project": "test", "type": "docforge-model"}),
encoding="utf-8",
)
(mcp_root / "nav.json").write_text(
json.dumps([]),
encoding="utf-8",
)
(modules_dir / "test.mod.json").write_text(
json.dumps({"module": "test.mod", "content": {}}),
encoding="utf-8",
)
return mcp_root
@pytest.fixture
def mock_mcp_server_run(monkeypatch):
"""
Mock MCPServer.run so no real server is started.
"""
called = {"value": False}
def fake_run(self, transport="streamable-http"):
called["value"] = True
monkeypatch.setattr(
"docforge.servers.MCPServer.run",
fake_run,
)
return lambda: called["value"]

View File

@@ -0,0 +1,18 @@
from docforge.cli.main import cli
def test_serve_mcp(
cli_runner,
fake_mcp_docs,
mock_mcp_server_run,
monkeypatch,
):
monkeypatch.chdir(fake_mcp_docs.parent)
result = cli_runner.invoke(
cli,
["serve-mcp"],
)
assert result.exit_code == 0
assert mock_mcp_server_run()