feat: mcp-py 接入工具总线 sundynix.tools.py.* (与 Go 同契约)

第 5b 层 Python 算法工具挂上 NATS，core NATS request-reply + 队列组，与 Go 侧 ServeTool 字节级同契约({tool,args,task_id}/{ok,content,error})。 - mcp_gateway: nats-py 连接(无限重连) + queue subscribe(mcp-py-workers) + 按工具名路由工具 echo / run_code(Docker桩) / parse_document(MinerU桩) / secure_sandbox(gVisor桩) - Makefile: 新增 mcp-py-setup(venv + pip install -e .)，mcp-py 缺 venv 自动 setup e2e 补上 TestToolCallRoundTrip - 验证: live 对 NATS 容器跑通 4 类调用(含未知工具错误)；3 个 Go e2e PASS Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 12:58:38 +08:00
parent e5fa0ae36c
commit a67604f4b7
2 changed files with 104 additions and 12 deletions
@@ -1,4 +1,4 @@
-.PHONY: infra infra-down devnats demo e2e gateway dispatcher mcp-go mcp-py desktop tidy
+.PHONY: infra infra-down devnats demo e2e gateway dispatcher mcp-go mcp-py mcp-py-setup desktop tidy

 infra:        ## 启动基础设施 (NATS / Postgres / Redis / Milvus / Neo4j)
 	docker compose up -d
@@ -12,8 +12,8 @@ devnats:      ## 启动内嵌 JetStream NATS（无 Docker 本地联调）
 demo:         ## 一键演示 Gateway→NATS→Dispatcher 任务流（无需 Docker）
 	bash scripts/demo.sh

-e2e:          ## 跑共享 bus 的端到端测试（内嵌 NATS）
-	cd sundynix-shared && go test ./bus/ -run 'TestTaskRoundTrip|TestTokenStreamRoundTrip' -v
+e2e:          ## 跑共享 bus 的端到端测试（内嵌 NATS）：任务流 / 工具调用 / Token 流
+	cd sundynix-shared && go test ./bus/ -run 'TestTaskRoundTrip|TestToolCallRoundTrip|TestTokenStreamRoundTrip' -v

 gateway:
 	cd sundynix-gateway && go run ./cmd/server
@@ -24,8 +24,12 @@ dispatcher:
 mcp-go:
 	cd sundynix-mcp-go && go run ./cmd/server

-mcp-py:
-	cd sundynix-mcp-py && python -m sundynix_mcp_py.main
+mcp-py-setup: ## 创建 venv 并安装 mcp-py 依赖（首次运行 mcp-py 前执行）
+	cd sundynix-mcp-py && python3 -m venv .venv && .venv/bin/pip install -q -e .
+
+mcp-py:       ## 运行 Python 算法型 MCP 工具服务（缺 venv 则自动 setup）
+	cd sundynix-mcp-py && [ -x .venv/bin/python ] || $(MAKE) mcp-py-setup
+	cd sundynix-mcp-py && .venv/bin/python -m sundynix_mcp_py.main

 desktop:
 	cd sundynix-desktop && wails dev
@@ -1,24 +1,112 @@
-"""MCP 协议网关：注册算法型工具并经 NATS 分发调用。"""
+"""MCP 协议网关：注册算法型工具并经 NATS 分发调用。
+
+与 Go 侧 sundynix-shared/bus 的 ServeTool 同契约（core NATS request-reply）：
+  请求体  {"tool": str, "args": {...}, "task_id": str}
+  应答体  {"ok": bool, "content": str, "error": str}
+订阅 sundynix.tools.py.>，队列组 mcp-py-workers（多副本负载均衡）。
+"""

 from __future__ import annotations

+import asyncio
+import json
+import logging
+import os
+
+import nats
+
 from .interpreter import CodeInterpreter
 from .mineru import MultimodalParser
 from .sandbox import SecureSandbox

+log = logging.getLogger("mcp_py")
+
+# 与 contract.SubjectToolsPyAll / QueueToolsPy 保持一致。
+SUBJECT_PY_ALL = "sundynix.tools.py.>"
+QUEUE_PY = "mcp-py-workers"
+

 class McpGateway:
    def __init__(self) -> None:
        self.sandbox = SecureSandbox()       # gVisor / KataVM + Static Code Guard
        self.parser = MultimodalParser()     # MinerU / PaddleOCR
        self.interpreter = CodeInterpreter() # Docker 隔离沙箱
+        self._nc: nats.NATS | None = None
+        self._tools: dict[str, callable] = {}

    async def register_tools(self) -> None:
-        # TODO: 向 MCP server 注册工具 schema
-        ...
-
-    async def serve(self) -> None:
-        # TODO: 连接 NATS，订阅 sundynix.tools.py.*，按 MCP JSON-RPC 路由
-        import asyncio
+        """注册工具名 → 处理协程。"""
+        self._tools = {
+            "echo": self._echo,
+            "run_code": self._run_code,
+            "parse_document": self._parse_document,
+            "secure_sandbox": self._secure_sandbox,
+        }

+    async def serve(self, url: str | None = None) -> None:
+        url = url or os.getenv("NATS_URL", "nats://localhost:4222")
+        # 容忍服务先于 NATS 启动：无限重连 + 等待间隔。
+        self._nc = await nats.connect(
+            url,
+            allow_reconnect=True,
+            max_reconnect_attempts=-1,
+            reconnect_time_wait=1,
+            connect_timeout=5,
+        )
+        log.info("[mcp_py] connected %s", url)
+        await self._nc.subscribe(SUBJECT_PY_ALL, queue=QUEUE_PY, cb=self._on_call)
+        log.info(
+            "[mcp_py] tools ready on %s (queue=%s): %s",
+            SUBJECT_PY_ALL, QUEUE_PY, ", ".join(self._tools),
+        )
        await asyncio.Event().wait()
+
+    async def _on_call(self, msg) -> None:
+        """解析 ToolCall → 路由到工具 → Respond ToolResult。"""
+        try:
+            req = json.loads(msg.data)
+        except Exception as e:  # noqa: BLE001
+            await self._reply(msg, ok=False, error=f"bad tool call: {e}")
+            return
+        tool = req.get("tool", "")
+        task_id = req.get("task_id", "")
+        args = req.get("args") or {}
+        log.info("[mcp_py] tool=%s task=%s args=%s", tool, task_id, args)
+        fn = self._tools.get(tool)
+        if fn is None:
+            await self._reply(msg, ok=False, error=f"unknown tool: {tool}")
+            return
+        try:
+            content = await fn(args)
+            await self._reply(msg, ok=True, content=content)
+        except Exception as e:  # noqa: BLE001
+            await self._reply(msg, ok=False, error=str(e))
+
+    @staticmethod
+    async def _reply(msg, *, ok: bool, content: str = "", error: str = "") -> None:
+        payload = json.dumps({"ok": ok, "content": content, "error": error})
+        await msg.respond(payload.encode())
+
+    # ---- 工具实现（算法层目前为桩，但调用链路做真）----
+
+    async def _echo(self, args: dict) -> str:
+        return str(args.get("text", ""))
+
+    async def _run_code(self, args: dict) -> str:
+        code = str(args.get("code", ""))
+        result = await self.interpreter.execute(code)  # Docker 隔离沙箱（桩）
+        return f"[run_code] Docker 隔离执行(桩) stdout={result.get('stdout','')!r}"
+
+    async def _parse_document(self, args: dict) -> str:
+        path = str(args.get("path", ""))
+        result = await self.parser.parse(path)  # MinerU / PaddleOCR（桩）
+        return f"[parse_document] MinerU 解析(桩) path={result.get('path','')!r} blocks={len(result.get('blocks', []))}"
+
+    async def _secure_sandbox(self, args: dict) -> str:
+        code = str(args.get("code", ""))
+        out = await self.sandbox.run(code)  # gVisor/KataVM（桩，含静态守卫）
+        return f"[secure_sandbox] gVisor 隔离执行(桩) out={out!r}"
+
+    async def close(self) -> None:
+        if self._nc is not None:
+            await self._nc.drain()