Compare commits
1 Commits
main
...
53868e7fc7
| Author | SHA1 | Date | |
|---|---|---|---|
| 53868e7fc7 |
176
generate_mcp.py
176
generate_mcp.py
@@ -1,176 +0,0 @@
|
|||||||
"""
|
|
||||||
Generate a fully-contained MCP bundle by reading Python docstrings directly.
|
|
||||||
|
|
||||||
Uses Griffe (the same library mkdocstrings-python uses internally),
|
|
||||||
but without MkDocs, Markdown, or HTML.
|
|
||||||
|
|
||||||
Output:
|
|
||||||
mcp/
|
|
||||||
├── index.json
|
|
||||||
├── nav.json
|
|
||||||
└── modules/
|
|
||||||
└── package.module.json
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import json
|
|
||||||
import argparse
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import Iterable
|
|
||||||
|
|
||||||
from griffe import GriffeLoader, ModulesCollection, LinesCollection, AliasResolutionError
|
|
||||||
|
|
||||||
|
|
||||||
PROJECT_ROOT = Path(__file__).resolve().parent
|
|
||||||
DEFAULT_PACKAGE_ROOT = "mail_intake"
|
|
||||||
DEFAULT_MCP_DIR = PROJECT_ROOT / "mcp"
|
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
|
||||||
# Utilities
|
|
||||||
# -------------------------
|
|
||||||
|
|
||||||
def iter_modules(package_root: Path, package_name: str) -> Iterable[str]:
|
|
||||||
for py in package_root.rglob("*.py"):
|
|
||||||
rel = py.relative_to(package_root.parent)
|
|
||||||
if py.name == "__init__.py":
|
|
||||||
yield ".".join(rel.parent.parts)
|
|
||||||
else:
|
|
||||||
yield ".".join(rel.with_suffix("").parts)
|
|
||||||
|
|
||||||
|
|
||||||
def serialize_object(obj) -> dict:
|
|
||||||
"""Convert a Griffe object into MCP-friendly JSON (alias-safe)."""
|
|
||||||
|
|
||||||
try:
|
|
||||||
docstring = obj.docstring.value if obj.docstring else None
|
|
||||||
except AliasResolutionError:
|
|
||||||
docstring = None
|
|
||||||
|
|
||||||
data = {
|
|
||||||
"name": obj.name,
|
|
||||||
"kind": obj.kind.value,
|
|
||||||
"path": obj.path,
|
|
||||||
"docstring": docstring,
|
|
||||||
}
|
|
||||||
|
|
||||||
# Signature may also trigger alias resolution
|
|
||||||
try:
|
|
||||||
if hasattr(obj, "signature") and obj.signature:
|
|
||||||
data["signature"] = str(obj.signature)
|
|
||||||
except AliasResolutionError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Recurse into members, but never allow alias failure to bubble up
|
|
||||||
members = {}
|
|
||||||
if hasattr(obj, "members"):
|
|
||||||
for name, member in obj.members.items():
|
|
||||||
if name.startswith("_"):
|
|
||||||
continue
|
|
||||||
try:
|
|
||||||
members[name] = serialize_object(member)
|
|
||||||
except AliasResolutionError:
|
|
||||||
continue
|
|
||||||
|
|
||||||
if members:
|
|
||||||
data["members"] = members
|
|
||||||
|
|
||||||
return data
|
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
|
||||||
# MCP generation
|
|
||||||
# -------------------------
|
|
||||||
|
|
||||||
def build_mcp(package_root: Path, package_name: str, mcp_root: Path) -> None:
|
|
||||||
modules_dir = mcp_root / "modules"
|
|
||||||
modules_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
|
|
||||||
loader = GriffeLoader(
|
|
||||||
search_paths=[str(package_root.parent)],
|
|
||||||
modules_collection=ModulesCollection(),
|
|
||||||
lines_collection=LinesCollection(),
|
|
||||||
)
|
|
||||||
|
|
||||||
nav = []
|
|
||||||
count = 0
|
|
||||||
|
|
||||||
for module in sorted(iter_modules(package_root, package_name)):
|
|
||||||
loader.load(module)
|
|
||||||
mod = loader.modules_collection[module]
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"module": module,
|
|
||||||
"content": serialize_object(mod),
|
|
||||||
}
|
|
||||||
|
|
||||||
out = modules_dir / f"{module}.json"
|
|
||||||
out.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
out.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
||||||
|
|
||||||
nav.append({
|
|
||||||
"module": module,
|
|
||||||
"resource": f"docs://module/{module}",
|
|
||||||
})
|
|
||||||
|
|
||||||
count += 1
|
|
||||||
|
|
||||||
(mcp_root / "nav.json").write_text(
|
|
||||||
json.dumps(nav, indent=2),
|
|
||||||
encoding="utf-8",
|
|
||||||
)
|
|
||||||
|
|
||||||
(mcp_root / "index.json").write_text(
|
|
||||||
json.dumps(
|
|
||||||
{
|
|
||||||
"project": package_name,
|
|
||||||
"type": "docstrings-direct",
|
|
||||||
"modules_count": count,
|
|
||||||
"source": "griffe",
|
|
||||||
},
|
|
||||||
indent=2,
|
|
||||||
),
|
|
||||||
encoding="utf-8",
|
|
||||||
)
|
|
||||||
|
|
||||||
print(f"MCP generated at: {mcp_root}")
|
|
||||||
|
|
||||||
|
|
||||||
# -------------------------
|
|
||||||
# CLI
|
|
||||||
# -------------------------
|
|
||||||
|
|
||||||
def main() -> None:
|
|
||||||
parser = argparse.ArgumentParser(
|
|
||||||
description="Generate MCP directly from Python docstrings (Griffe-based)",
|
|
||||||
)
|
|
||||||
|
|
||||||
parser.add_argument(
|
|
||||||
"--package-root",
|
|
||||||
default=DEFAULT_PACKAGE_ROOT,
|
|
||||||
help="Root Python package name",
|
|
||||||
)
|
|
||||||
|
|
||||||
parser.add_argument(
|
|
||||||
"--mcp-dir",
|
|
||||||
type=Path,
|
|
||||||
default=DEFAULT_MCP_DIR,
|
|
||||||
help="Output MCP directory",
|
|
||||||
)
|
|
||||||
|
|
||||||
args = parser.parse_args()
|
|
||||||
|
|
||||||
package_root = PROJECT_ROOT / args.package_root
|
|
||||||
if not package_root.exists():
|
|
||||||
raise FileNotFoundError(package_root)
|
|
||||||
|
|
||||||
build_mcp(
|
|
||||||
package_root=package_root,
|
|
||||||
package_name=args.package_root,
|
|
||||||
mcp_root=args.mcp_dir,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
||||||
@@ -1,32 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
import json
|
|
||||||
from mcp.server.fastmcp import FastMCP
|
|
||||||
|
|
||||||
MCP_ROOT = Path("mcp")
|
|
||||||
|
|
||||||
mcp = FastMCP("aetoskia-mail-intake-docs")
|
|
||||||
|
|
||||||
def read_json(path: Path):
|
|
||||||
if not path.exists():
|
|
||||||
return {"error": "not_found", "path": str(path)}
|
|
||||||
return json.loads(path.read_text())
|
|
||||||
|
|
||||||
@mcp.resource("docs://index")
|
|
||||||
def index():
|
|
||||||
return read_json(MCP_ROOT / "index.json")
|
|
||||||
|
|
||||||
@mcp.resource("docs://nav")
|
|
||||||
def nav():
|
|
||||||
return read_json(MCP_ROOT / "nav.json")
|
|
||||||
|
|
||||||
@mcp.resource("docs://module/{module}")
|
|
||||||
def module(module: str):
|
|
||||||
return read_json(MCP_ROOT / "modules" / f"{module}.json")
|
|
||||||
|
|
||||||
@mcp.tool()
|
|
||||||
def ping() -> str:
|
|
||||||
return "Pong! (fake tool executed)"
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
# FastMCP owns the HTTP server
|
|
||||||
mcp.run(transport="streamable-http")
|
|
||||||
Reference in New Issue
Block a user