ai-demo/app/tools/base_mcp_server.py

"""
MCP Server 基类
提供通用的 MCP 服务器基础设施

设计说明：
- McpServer: 通用 MCP 服务器基类
  - 子类在 __init__ 中通过 self.tools = [...] 赋值工具实例
  - 提供工具列表展示、调用处理、服务器启动等通用逻辑

使用示例：
    class MyServer(McpServer):
        def __init__(self):
            super().__init__("my-mcp", "1.0.0")
            # 直接赋值工具实例列表
            self.tools = [MyTool1(), MyTool2()]
    
    # 启动服务器
    server = MyServer()
    await server.run_stdio()
"""
import asyncio
from typing import Any, Dict, List, Optional

from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import (
    Tool,
    TextContent,
    ImageContent,
    EmbeddedResource,
    ServerCapabilities,
)


class BaseMcpServer:
    """
    通用 MCP 服务器基类

    子类在 __init__ 中通过 self.tools 属性赋值工具实例列表。
    封装了工具列表展示、调用处理、结果格式化、服务器启动等通用逻辑。
    """

    def __init__(self, server_name: str, version: str = "1.0.0"):
        """
        初始化 MCP 服务器

        Args:
            server_name: 服务器名称
            version: 服务器版本
        """
        self.server_name = server_name
        self.version = version
        self.server = Server(server_name)

        # 工具列表（子类在 __init__ 中赋值）
        self.tools: List[Any] = []

        # 注册处理器
        self.server.list_tools()(self._handle_list_tools)
        self.server.call_tool()(self._handle_call_tool)

    async def _handle_list_tools(self) -> list[Tool]:
        """列出所有可用的工具"""
        return [
            Tool(
                name=tool.name,
                description=tool.description,
                inputSchema=tool.parameters_schema,
            )
            for tool in self.tools
        ]

    async def _handle_call_tool(
        self, name: str, arguments: Optional[dict[str, Any]] = None
    ) -> list[TextContent | ImageContent | EmbeddedResource]:
        """处理工具调用"""
        if not arguments:
            arguments = {}

        # 查找工具
        tool = next((t for t in self.tools if t.name == name), None)
        if not tool:
            return [TextContent(type="text", text=f"未知工具：{name}")]

        try:
            result = tool.execute(**arguments)
            response_text = self._format_tool_result(result)
            return [TextContent(type="text", text=response_text)]

        except ValueError as e:
            return [TextContent(type="text", text=f"❌ 参数错误：{str(e)}")]
        except RuntimeError as e:
            return [TextContent(type="text", text=f"❌ 执行失败：{str(e)}")]
        except Exception as e:
            return [TextContent(type="text", text=f"❌ 未知错误：{str(e)}")]

    @staticmethod
    def _format_tool_result(result: Any) -> str:
        """
        格式化工具执行结果

        支持多种返回格式：
        - 包含 message/data 属性的对象
        - 字典（包含 message/data 键）
        - 其他类型（直接转为字符串）
        """
        response_text = ""

        # 提取消息
        message = getattr(result, 'message', None)
        if message:
            response_text += f"✅ {message}\n\n"
        elif isinstance(result, dict) and 'message' in result:
            response_text += f"✅ {result['message']}\n\n"

        # 提取数据
        data = getattr(result, 'data', None)
        if data is None and isinstance(result, dict):
            data = result.get('data')

        if data:
            response_text += "详情：\n"
            if isinstance(data, dict):
                for key, value in data.items():
                    response_text += f"  - {key}: {value}\n"
            else:
                response_text += f"  {data}\n"

        # 如果没有消息也没有数据，直接返回结果的字符串形式
        if not response_text:
            response_text = str(result)

        return response_text

    async def run_stdio(self):
        """
        启动服务器，使用标准输入输出通信
        """
        from mcp.server.stdio import stdio_server

        async with stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name=self.server_name,
                    server_version=self.version,
                    capabilities=ServerCapabilities(
                        tools={}  # 启用工具能力
                    ),
                ),
            )