Vishesh 'ironeagle' Bangotra
9929f05334
- collect.py: copy repo MCP bundles to ROOT/<name>/mcp (instead of ROOT/mcp/<name>); add nginx location guard to deny /<name>/mcp/ - config.py: bundle roots now point to <repo>/mcp and standards/mcp for static bundles - Rename core.py -> mcp_server.py and update imports in main.py and registry.py - Dockerfile: per-repo MCP COPYs under /app/<name>/mcp, drop blog copy; add ports 8008/8009 in .drone.yml - Restore standards/mcp/libraries.json from git for registry server
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MCPBundle:
|
|
root: Path
|
|
name: str
|
|
port: int
|
|
kind: str = "docs" # "docs" -> base MCPServer; "registry" -> StandardsMCPServer
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepoDoc:
|
|
name: str
|
|
description: str | None
|
|
lib: str | None
|
|
api: str | None
|
|
wiki: str | None
|
|
mcp_bundle: str | None # source-relative path in the repo (collect use)
|
|
mcp_server: str | None # MCP server / namespace name
|
|
mcp_port: int | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServiceConfig:
|
|
name: str
|
|
html_port: int
|
|
health_port: int
|
|
repos: tuple[RepoDoc, ...]
|
|
mcp_static: dict[str, MCPBundle] = field(default_factory=dict)
|
|
|
|
@property
|
|
def mcp_servers(self) -> dict[str, MCPBundle]:
|
|
"""Return every MCP server: one per repo that declares a bundle, plus
|
|
the hand-authored static servers (for example the standards registry)."""
|
|
servers: dict[str, MCPBundle] = {
|
|
r.name: MCPBundle(root=Path(r.name) / "mcp", name=r.mcp_server, port=r.mcp_port)
|
|
for r in self.repos
|
|
if r.mcp_port is not None
|
|
}
|
|
servers.update(self.mcp_static)
|
|
return servers
|
|
|
|
|
|
def load_config(path: Path | str = "config.yml") -> ServiceConfig:
|
|
p = Path(path)
|
|
data = yaml.safe_load(p.read_text(encoding="utf-8"))
|
|
service = data.get("service") or {}
|
|
repos: list[RepoDoc] = []
|
|
for entry in data.get("repos", []):
|
|
docs = entry.get("docs") or {}
|
|
mcp = docs.get("mcp")
|
|
repos.append(RepoDoc(
|
|
name=entry["name"],
|
|
description=entry.get("description"),
|
|
lib=docs.get("lib"),
|
|
api=docs.get("api"),
|
|
wiki=docs.get("wiki"),
|
|
mcp_bundle=mcp["bundle"] if mcp else None,
|
|
mcp_server=mcp["server"] if mcp else None,
|
|
mcp_port=mcp["port"] if mcp else None,
|
|
))
|
|
|
|
mcp_static: dict[str, MCPBundle] = {}
|
|
for entry in data.get("mcp_static", []):
|
|
mcp_static[entry["name"]] = MCPBundle(
|
|
root=Path("mcp") / entry["bundle"],
|
|
name=entry.get("server") or entry["name"],
|
|
port=entry["port"],
|
|
kind=entry.get("kind", "docs"),
|
|
)
|
|
|
|
return ServiceConfig(
|
|
name=service.get("name", "aetos-docs"),
|
|
html_port=service.get("html_port", 9000),
|
|
health_port=service.get("health_port", 8000),
|
|
repos=tuple(repos),
|
|
mcp_static=mcp_static,
|
|
)
|