mcp_server/                    # MCP Server（独立微服务）
│   └── server.py                  # FastMCP + Streamable HTTP，暴露电商工具


    ── mcp_client/                # MCP Client（同步封装）
│   │   ├── client.py              # MCPClient：后台线程管理异步连接
│   │   └── converter.py           # MCP Tool schema → OpenAI function calling 格式

│   ├── config/
│   │   └── settings.py            # 配置管理（从 .env 读取，含 MCP / RAG / Multi-Agent / Memory / Skill / Evaluation 配置）


        ├── tools/                 # 电商工具集（Function Calling）
    │   │   │   ├── manager.py         # ToolManager：统一管理本地 + MCP 工具（支持 allowed_tools 过滤）

mcp配置

MCP 服务器配置 (servers_config.json)

MCP服务，函数注册为mcp工具(server.py)

STDIO 模式（本地进程）：

MCP Client (Agent) -----stdin/stdout----- MCP Server(weather.py) 

SSE 模式（远程服务）：

MCP Client (Agent) -----HTTP/SSE----- 远程MCP服务器 

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("CalculatorServer")

"""
MCP Server：通过 Streamable HTTP 暴露工具服务。
启动方式：python mcp_server/server.py
默认监听：http://127.0.0.1:9123/mcp

mcp = FastMCP("ecom-tools", host="127.0.0.1", port=9123)
"""

@mcp.tool()
async def calculate(expression: str) -> str:
    """
    计算数学表达式。
    :param expression: 数学表达式，如 "2 + 3 * 4"
    :return: 计算结果
    """
    try:
        result = eval(expression)  # 注意：生产环境需要安全处理
        return f"结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

if __name__ == "__main__":
    mcp.run(transport='stdio')

1. @mcp.tool() 装饰器：将普通函数注册为 MCP 工具，LLM 可以自动发现和调用

2. 函数文档字符串：非常重要！LLM 通过文档字符串理解工具的功能

3. transport='stdio'：通过标准输入/输出与客户端通信

api服务(api_server.py)

    # 1. 读取 MCP 服务器配置，初始化连接
    servers_cfg = Configuration.load_servers()
    
    # 2. 同时连接 多个MCP 服务器并自动获取所有可用的工具
    mcp_client = MultiServerMCPClient(servers_cfg)
    tools = await mcp_client.get_tools()
    
    # 3. 初始化 ReAct Agent
    model = ChatTongyi(model=cfg.model, streaming=False)
    checkpointer = InMemorySaver()  # 内存中保存对话历史
    
    agent = create_react_agent(
        model=model, 
        tools=tools,
        prompt=prompt,
        checkpointer=checkpointer
    )
    
    yield  # 应用运行中
    
    # 4. 应用关闭时清理资源
    await mcp_client.cleanup()

工具管理

"""ToolManager：统一管理本地工具和 MCP 工具。

当 MCP 启用时，通过 Streamable HTTP 连接 MCP Server 获取工具；
当 MCP 未启用或连接失败时，退回本地工具。
"""

import json
from typing import Optional

from app.agent.tools.registry import TOOL_DEFINITIONS as LOCAL_TOOL_DEFINITIONS
from app.agent.tools.registry import execute_tool as local_execute_tool


class ToolManager:
    """聚合本地工具和 MCP 工具，提供统一的工具定义和调度接口。"""

    def __init__(
        self,
        use_mcp: bool = False,
        mcp_server_url: str = "",
        allowed_tools: Optional[set] = None,
    ):
        self._mcp_client = None
        self._tool_source: dict[str, str] = {}
        self._tool_defs: list[dict] = []

        if use_mcp and mcp_server_url:
            self._init_mcp(mcp_server_url)
        else:
            self._init_local()

        if allowed_tools is not None:
            self._filter_tools(allowed_tools)

    def _init_local(self):
        """只加载本地工具。"""
        self._tool_defs = list(LOCAL_TOOL_DEFINITIONS)
        for td in self._tool_defs:
            self._tool_source[td["function"]["name"]] = "local"

    def _init_mcp(self, server_url: str):
        """连接 MCP Server 加载工具；失败时降级到本地工具。"""
        from app.mcp_client import MCPClient

        try:
            self._mcp_client = MCPClient(server_url)
            mcp_tools = self._mcp_client.connect()
            print(f"🔗 [MCP] 已连接 {server_url}，发现 {len(mcp_tools)} 个工具")

            mcp_names = set()
            for td in mcp_tools:
                name = td["function"]["name"]
                mcp_names.add(name)
                self._tool_source[name] = "mcp"
            self._tool_defs = list(mcp_tools)

            for td in LOCAL_TOOL_DEFINITIONS:
                name = td["function"]["name"]
                if name not in mcp_names:
                    self._tool_defs.append(td)
                    self._tool_source[name] = "local"

        except Exception as e:
            print(f"⚠️  [MCP] 连接失败 ({e})，降级使用本地工具")
            if self._mcp_client:
                self._mcp_client.close()
                self._mcp_client = None
            self._init_local()

    def _filter_tools(self, allowed: set):
        """只保留白名单中的工具，用于子 Agent 工具隔离。"""
        self._tool_defs = [
            d for d in self._tool_defs
            if d["function"]["name"] in allowed
        ]
        self._tool_source = {
            k: v for k, v in self._tool_source.items()
            if k in allowed
        }

    @property
    def tool_definitions(self) -> list[dict]:
        return self._tool_defs

    def execute_tool(self, name: str, arguments: dict) -> str:
        """根据工具来源分发调用。"""
        source = self._tool_source.get(name)

        if source == "mcp" and self._mcp_client:
            return self._mcp_client.call_tool(name, arguments)

        if source == "local":
            return local_execute_tool(name, arguments)

        return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False)

    def close(self):
        """清理 MCP 连接。"""
        if self._mcp_client:
            self._mcp_client.close()
            self._mcp_client = None

很全的项目

GitHub - HuaiNan54321/ecom-service-agent · GitHub