Spaces:
Paused
Paused
File size: 2,722 Bytes
6b3874e | 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 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | """Bonus: expose the audited tool registry as a real MCP server.
The `mcp` python package installed cleanly in this environment, so the
registry can be served over stdio to any MCP client:
.venv/bin/python -m sensorflow.retro.tools.mcp_server
Design notes:
- Every MCP tool call still flows through ToolRegistry.call(), so the audit
trail, schema validation, timeouts, and write-authorization gate apply
identically to MCP clients and in-process callers.
- The write tool (create_evaluation_case) is exposed but the registry will
deny it unless the client passes policy_authorization=true in arguments —
the MCP layer cannot bypass the boundary.
- If `mcp` is not installed this module raises a clear ImportError; the
registry itself remains the MCP-style boundary (documented in registry.py).
"""
from __future__ import annotations
from typing import Any, Dict
from sensorflow.retro.tools.builtin import build_registry
def _import_server_class():
"""Support both the modern (MCPServer) and older (FastMCP) APIs."""
try:
from mcp.server.mcpserver import MCPServer
return MCPServer
except ImportError:
pass
try:
from mcp.server.fastmcp import FastMCP
return FastMCP
except ImportError as exc: # honest failure; registry remains the boundary
raise ImportError(
"the 'mcp' package is not installed (or has an unknown server "
"API); the ToolRegistry remains the MCP-style boundary "
"(pip install mcp to enable this server)") from exc
def build_mcp_server():
"""Construct (but do not run) the MCP server wrapping the registry."""
server_cls = _import_server_class()
registry = build_registry(analysis_id="mcp-session")
server = server_cls(name="sensorflow-retro")
def make_handler(tool_name: str):
def handler(arguments: Dict[str, Any] | None = None) -> Dict[str, Any]:
args = dict(arguments or {})
authorized = bool(args.pop("policy_authorization", False))
res = registry.call(tool_name, args, policy_authorization=authorized)
return res.model_dump()
return handler
for spec in registry.specs():
server.add_tool(
make_handler(spec.name),
name=spec.name,
description=(f"{spec.description} "
f"[read_only={spec.read_only}, timeout={spec.timeout_s}s; "
f"args: single 'arguments' object matching the input "
f"schema; write tools need policy_authorization=true]"))
return server, registry
if __name__ == "__main__":
srv, _ = build_mcp_server()
srv.run() # stdio transport
|