File size: 1,836 Bytes
1fde9f6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """M3D module SDK.
A module = a module.json manifest + an MCP server exposing the same tools
in-process (used by the DAG runner) and over stdio/HTTP (used by external
agents and agentic harnesses). One class, one interface, every runtime.
Built on MCP Python SDK v2 (pip install mcp==2.2.0):
from mcp.server import MCPServer; @mcp.tool() registers a typed tool.
"""
from __future__ import annotations
import json
from typing import Any
from mcp.server import MCPServer
class M3DModule:
def __init__(self, manifest: dict):
assert "id" in manifest and "version" in manifest, "manifest needs id + version"
assert "license" in manifest, "every M3D module declares its license graph"
self.manifest = manifest
self.server = MCPServer(manifest["id"])
self._tools: dict[str, Any] = {}
def tool(self, name: str | None = None):
"""Register a typed function as both an M3D tool and an MCP tool."""
def deco(fn):
tool_name = name or fn.__name__
self._tools[tool_name] = fn
self.server.add_tool(fn, name=tool_name) # MCP v2 registration
return fn
return deco
def run_tool(self, name: str, args: dict) -> Any:
"""Direct in-process call used by the DAG runner (isolation principle)."""
if name not in self._tools:
raise KeyError(
f"module {self.manifest['id']} has no tool {name!r}; "
f"available: {sorted(self._tools)}"
)
return self._tools[name](**args)
def serve(self, transport: str = "stdio") -> None:
"""Serve over MCP — stdio/sse/streamable-http, any agentic harness can attach."""
self.server.run(transport=transport)
def manifest_json(self) -> str:
return json.dumps(self.manifest, indent=2) |