PlantAI-rag-new/tooling/cap/fake_mcp_server.py

#!/usr/bin/env python3
"""fake_mcp_server.py —— 自检用假 MCP server（stdio，一个 echo 工具）。

用官方 mcp SDK 2.x 的 lowlevel Server + add_request_handler 实现；
若 mcp SDK 未安装则退化为纯 stdio JSON-RPC 手工实现（保底）。
S1 端到端自检用它验证「热加载 + 工具调用」全链。
"""
from __future__ import annotations

import json
import sys


def _serve_manual() -> int:
    """无 mcp SDK 保底：JSON-RPC over stdio（initialize/tools-list/tools-call）。"""
    for line in sys.stdin:
        try:
            message = json.loads(line)
        except Exception:
            continue
        method = message.get("method")
        msg_id = message.get("id")
        if method == "initialize":
            reply = {
                "jsonrpc": "2.0",
                "id": msg_id,
                "result": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {"tools": {}},
                    "serverInfo": {"name": "fake-echo", "version": "0.1.0"},
                },
            }
        elif method == "notifications/initialized":
            continue
        elif method == "tools/list":
            reply = {
                "jsonrpc": "2.0",
                "id": msg_id,
                "result": {
                    "tools": [
                        {
                            "name": "echo",
                            "description": "回显文本",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "text": {"type": "string", "description": "要回显的文本"}
                                },
                                "required": ["text"],
                            },
                        }
                    ]
                },
            }
        elif method == "tools/call":
            params = message.get("params") or {}
            arguments = params.get("arguments") or {}
            text = str(arguments.get("text", ""))
            reply = {
                "jsonrpc": "2.0",
                "id": msg_id,
                "result": {
                    "content": [{"type": "text", "text": f"echo: {text}"}],
                    "isError": False,
                },
            }
        else:
            reply = {
                "jsonrpc": "2.0",
                "id": msg_id,
                "error": {"code": -32601, "message": f"unknown method {method}"},
            }
        sys.stdout.write(json.dumps(reply, ensure_ascii=False) + "\n")
        sys.stdout.flush()
    return 0


def _serve_mcp() -> int:
    from mcp.server.lowlevel import Server
    from mcp.server.stdio import stdio_server
    import mcp.types as types
    import anyio

    server = Server("fake-echo")

    async def handle_list_tools(ctx, params: types.PaginatedRequestParams):
        return types.ListToolsResult(
            tools=[
                types.Tool(
                    name="echo",
                    description="回显文本",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "text": {"type": "string", "description": "要回显的文本"}
                        },
                        "required": ["text"],
                    },
                )
            ]
        )

    async def handle_call_tool(ctx, params: types.CallToolRequestParams):
        arguments = params.arguments or {}
        text = str(arguments.get("text", ""))
        return types.CallToolResult(
            content=[types.TextContent(type="text", text=f"echo: {text}")],
            is_error=False,
        )

    server.add_request_handler("tools/list", types.PaginatedRequestParams, handle_list_tools)
    server.add_request_handler("tools/call", types.CallToolRequestParams, handle_call_tool)

    async def run() -> None:
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream, server.create_initialization_options())

    anyio.run(run)
    return 0


def main() -> int:
    try:
        import mcp  # noqa: F401
    except ImportError:
        return _serve_manual()
    return _serve_mcp()


if __name__ == "__main__":
    raise SystemExit(main())