MCP 多 Agent 连接与工具注册精华指南

专注：MCP Server 连接管理 + 工具注册/发现/选择机制
基于：MCP Spec v2025-11-25 + Python SDK v2.0.0 + LangGraph 集成

---

一、连接管理（Connection）

1.1 连接类型

类型	协议	场景	特点
stdio	标准输入/输出	本地进程	简单、安全、无网络开销
Streamable HTTP	HTTP + SSE	远程服务	支持流式响应、可水平扩展
HTTP (无流)	HTTP	简单远程调用	最基础实现

1.2 连接池（Connection Pool）

# 自定义连接池实现（参考 src/mcp/client.py）
class ConnectionPool:
    def __init__(self, max_connections: int = 10, timeout: int = 30):
        self.max_connections = max_connections
        self.timeout = timeout
        self._pools: Dict[str, aiohttp.ClientSession] = {}

    async def get_session(self, server_url: str) -> aiohttp.ClientSession:
        """获取或创建 HTTP 会话（每 Server 一个）"""
        with self._lock:
            if server_url not in self._pools:
                connector = aiohttp.TCPConnector(
                    limit=self.max_connections,
                    limit_per_host=self.max_connections
                )
                session = aiohttp.ClientSession(
                    connector=connector,
                    timeout=aiohttp.ClientTimeout(total=self.timeout)
                )
                self._pools[server_url] = session
            return self._pools[server_url]

    async def close_all(self):
        """关闭所有连接"""
        tasks = [session.close() for session in self._pools.values()]
        await asyncio.gather(*tasks, return_exceptions=True)
        self._pools.clear()

关键设计点：

• Per-Server Session：每个 MCP Server 独立维护连接池
• 超时控制：默认 30s，可根据工具类型调整（IO 密集型可设更长）
• 优雅关闭：应用退出时关闭所有连接

1.3 标准 MCP Client 连接

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamablehttp_client

# stdio 模式
server_params = StdioServerParameters(
    command="python",
    args=["my_server.py"],
    env={"DEBUG": "1"}
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()

# Streamable HTTP 模式
async with streamablehttp_client("https://api.example.com/mcp") as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()

1.4 MultiServerMCPClient（推荐）

from langchain_mcp_adapters.client import MultiServerMCPClient

# 同时连接多个 Server
async with MultiServerMCPClient({
    "database": {
        "command": "python",
        "args": ["-m", "mcp_server_db"],
        "transport": "stdio"
    },
    "filesystem": {
        "command": "python",
        "args": ["-m", "mcp_server_fs", "/workspace"],
        "transport": "stdio"
    },
    "analytics": {
        "url": "https://analytics.example.com/mcp",
        "transport": "streamable_http",
        "headers": {"Authorization": "Bearer xxx"}  # 支持认证
    }
}) as client:
    # 获取所有工具
    all_tools = client.get_tools()

    # 按 Server 过滤
    db_tools = [t for t in all_tools if "database" in t.tags]

---

二、工具注册（Tool Registration）

2.1 Server 端注册

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MyToolServer")

# 方式 1：装饰器注册（推荐）
@mcp.tool()
async def query_database(sql: str, database: str = "main") -> str:
    """执行 SQL 查询（只读）。

    Args:
        sql: SQL 查询语句
        database: 数据库名称，默认 main
    """
    result = await db.execute(sql)
    return json.dumps(result)

# 方式 2：动态注册
@mcp.tool(
    name="custom_tool",
    description="自定义工具",
    tags={"custom", "utility"}
)
async def custom_handler(param: str) -> str:
    return f"Result: {param}"

# 方式 3：带输出 Schema
@mcp.tool()
async def get_user(user_id: int) -> dict:
    """获取用户信息。"""
    user = await db.get_user(user_id)
    return {
        "id": user.id,
        "name": user.name,
        "email": user.email
    }

# 启动 Server
mcp.run(transport="stdio")

2.2 工具定义结构

# MCPToolDefinition（参考 src/models/base.py）
class MCPToolDefinition(BaseModel):
    name: str                          # 工具名称（全局唯一）
    description: str                   # 工具描述（LLM 理解的关键）
    server_url: str                    # Server URL
    tool_name: str                     # Server 端工具名
    parameters: List[MCPToolParameter] # 参数定义
    capabilities: List[str]            # 能力标签（用于选择）
    applicable_tasks: List[TaskType]   # 适用任务类型
    priority: int                      # 优先级（1-10）
    timeout_seconds: int               # 超时时间
    cache_enabled: bool                # 是否启用缓存

class MCPToolParameter(BaseModel):
    name: str                          # 参数名
    type: str                          # 类型：string/number/boolean/object/array
    description: str                   # 参数描述
    required: bool = True              # 是否必需
    default: Any = None                # 默认值
    enum: Optional[List[str]] = None   # 枚举值

2.3 工具注册表（Registry）

# 参考 src/mcp/registry.py
class MCPToolRegistry:
    def __init__(self):
        self.tools: Dict[str, MCPToolDefinition] = {}
        self._category_index: Dict[ToolCategory, Set[str]] = {}
        self._task_index: Dict[TaskType, Set[str]] = {}
        self._capability_index: Dict[str, Set[str]] = {}

    def register_tool(self, tool_def: MCPToolDefinition):
        """注册工具并建立索引"""
        with self._lock:
            self.tools[tool_def.name] = tool_def
            # 建立多维度索引
            self._update_indexes(tool_def)

    def _update_indexes(self, tool: MCPToolDefinition):
        # 类别索引
        if tool.category not in self._category_index:
            self._category_index[tool.category] = set()
        self._category_index[tool.category].add(tool.name)

        # 任务索引
        for task_type in tool.applicable_tasks:
            if task_type not in self._task_index:
                self._task_index[task_type] = set()
            self._task_index[task_type].add(tool.name)

        # 能力索引
        for capability in tool.capabilities:
            if capability not in self._capability_index:
                self._capability_index[capability] = set()
            self._capability_index[capability].add(tool.name)

    def get_tools_for_task(self, task_type: TaskType) -> List[MCPToolDefinition]:
        """获取适用于特定任务的工具（按优先级排序）"""
        tool_names = self._task_index.get(task_type, set())
        tools = [self.tools[name] for name in tool_names if name in self.tools]
        return sorted(tools, key=lambda t: t.priority, reverse=True)

---

三、工具发现与选择（Tool Discovery & Selection）

3.1 动态发现（标准 MCP）

async with ClientSession(read, write) as session:
    await session.initialize()

    # 发现所有工具
    tools_result = await session.list_tools()
    tools = tools_result.tools

    for tool in tools:
        print(f"{tool.name}: {tool.description}")
        print(f"  Schema: {tool.inputSchema}")

    # 调用工具
    result = await session.call_tool(
        "query_database",
        {"sql": "SELECT * FROM users LIMIT 10"}
    )

3.2 智能选择（4 策略）

# 参考 src/mcp/selector.py
class ToolSelector:
    def __init__(self, registry: MCPToolRegistry):
        self.registry = registry
        self._keyword_cache: Dict[str, List[str]] = {}

    def select_tools_for_task(
        self,
        task_type: TaskType,
        user_input: str,
        max_tools: int = 3,
        context: Optional[Dict] = None
    ) -> List[MCPToolSelection]:
        """多策略工具选择"""
        candidates = []
        seen_tools = set()

        # 策略 1：基于任务类型
        task_tools = self.registry.get_tools_for_task(task_type)
        for tool in task_tools:
            if tool.name not in seen_tools:
                confidence = self._calculate_task_relevance(tool, task_type, user_input)
                candidates.append((tool, confidence))
                seen_tools.add(tool.name)

        # 策略 2：基于关键词匹配
        keyword_tools = self._select_by_keywords(user_input)
        for tool_name in keyword_tools:
            if tool_name not in seen_tools:
                tool = self.registry.get_tool(tool_name)
                if tool:
                    confidence = self._calculate_keyword_relevance(tool, user_input)
                    candidates.append((tool, confidence))
                    seen_tools.add(tool_name)

        # 策略 3：基于能力匹配
        capability_tools = self._select_by_capabilities(task_type, user_input)
        for tool_name in capability_tools:
            if tool_name not in seen_tools:
                tool = self.registry.get_tool(tool_name)
                if tool:
                    confidence = self._calculate_capability_relevance(tool, task_type, user_input)
                    candidates.append((tool, confidence))
                    seen_tools.add(tool_name)

        # 策略 4：基于上下文（历史记录）
        if context:
            context_tools = self._select_by_context(context)
            for tool_name in context_tools:
                if tool_name not in seen_tools:
                    tool = self.registry.get_tool(tool_name)
                    if tool:
                        confidence = self._calculate_context_relevance(tool, context)
                        candidates.append((tool, confidence))
                        seen_tools.add(tool_name)

        # 排序并返回 Top-N
        candidates.sort(key=lambda x: x[1], reverse=True)
        return [
            MCPToolSelection(
                tool_name=tool.name,
                reason=self._generate_selection_reason(tool, confidence),
                confidence=confidence,
                parameters=self._suggest_parameters(tool, task_type, user_input)
            )
            for tool, confidence in candidates[:max_tools]
        ]

    def _calculate_task_relevance(self, tool, task_type, user_input) -> float:
        """任务类型相关性评分"""
        base_score = 0.8 if task_type in tool.applicable_tasks else 0.2
        priority_bonus = (tool.priority - 1) / 9 * 0.2
        input_relevance = self._calculate_input_relevance(tool, user_input)
        return min(1.0, base_score + priority_bonus + input_relevance)

    def _calculate_keyword_relevance(self, tool, user_input) -> float:
        """关键词匹配评分"""
        input_lower = user_input.lower()
        total_keywords = 0
        matched_keywords = 0

        for pattern, tool_names in self.keyword_mappings.items():
            if tool.name in tool_names:
                keywords = pattern.split('|')
                total_keywords += len(keywords)
                for keyword in keywords:
                    if keyword.strip().lower() in input_lower:
                        matched_keywords += 1

        if total_keywords == 0:
            return 0.0
        keyword_score = matched_keywords / total_keywords
        return round(min(1.0, 0.3 + keyword_score * 0.7), 3)

3.3 关键词映射配置

# 中英文关键词映射
keyword_mappings = {
    # 数据分析
    "数据|分析|统计|报表|图表|趋势|data|analysis|statistics|report|chart|trend": [
        "data_analysis_tool"
    ],
    # 代码分析
    "代码|编程|审查|质量|bug|调试|security|code|programming|review|quality|debug": [
        "code_analysis_tool"
    ],
    # 搜索
    "搜索|查找|查询|信息|网页|web|search|find|query|information|webpage": [
        "web_search_tool"
    ],
    # 文件系统
    "文件|目录|读取|写入|创建|删除|file|directory|read|write|create|delete": [
        "file_system_tool"
    ],
    # 网络
    "网络|API|请求|HTTP|调用|network|api|request|http|call": [
        "network_tool"
    ],
    # 创意
    "创意|生成|写作|故事|设计|creative|generate|write|story|design": [
        "creative_tool"
    ]
}

---

四、工具执行（Tool Execution）

4.1 并行执行

# 参考 src/mcp/executor.py
class MCPToolExecutor:
    def __init__(self, mcp_client: MCPClient, max_concurrent: int = 5):
        self.mcp_client = mcp_client
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_parallel_primary(
        self,
        tool_selections: List[MCPToolSelection]
    ) -> Dict[str, Any]:
        """并行执行多个工具"""
        tasks = [
            self._execute_single_tool_with_semaphore(selection)
            for selection in tool_selections
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        primary_results = []
        failed_calls = []

        for i, result in enumerate(results):
            tool_name = tool_selections[i].tool_name
            if isinstance(result, Exception):
                failed_calls.append(tool_name)
                primary_results.append({
                    "tool_name": tool_name,
                    "success": False,
                    "error": str(result)
                })
            else:
                primary_results.append(result)
                if not result.success:
                    failed_calls.append(tool_name)

        return {
            "primary_results": primary_results,
            "failed_calls": failed_calls,
            "success_rate": (len(tool_selections) - len(failed_calls)) / len(tool_selections)
        }

    async def _execute_single_tool_with_semaphore(
        self,
        selection: MCPToolSelection
    ) -> MCPExecutionResult:
        async with self.semaphore:  # 控制并发数
            return await self._execute_single_tool(selection)

    async def _execute_single_tool(
        self,
        selection: MCPToolSelection
    ) -> MCPExecutionResult:
        tool_def = self.registry.get_tool(selection.tool_name)
        if not tool_def:
            return MCPExecutionResult(
                success=False,
                error_message=f"工具 '{selection.tool_name}' 未找到"
            )

        return await self.mcp_client.call_tool(
            server_url=tool_def.server_url,
            tool_name=tool_def.tool_name,
            parameters=selection.parameters or {},
            use_cache=tool_def.cache_enabled
        )

4.2 缓存与重试

# 参考 src/mcp/client.py
class MCPClient:
    def __init__(
        self,
        cache_manager: CacheManager,
        max_retries: int = 3,
        retry_delay: float = 1.0
    ):
        self.cache_manager = cache_manager
        self.max_retries = max_retries
        self.retry_delay = retry_delay

    async def call_tool(
        self,
        server_url: str,
        tool_name: str,
        parameters: Dict[str, Any],
        use_cache: bool = True
    ) -> MCPExecutionResult:
        """调用工具（带缓存和重试）"""
        # 1. 检查缓存
        if use_cache:
            cached = await self.cache_manager.get(server_url, tool_name, parameters)
            if cached is not None:
                return MCPExecutionResult(success=True, result=cached)

        # 2. 带重试的执行
        result = await self._call_tool_with_retry(server_url, tool_name, parameters)

        # 3. 缓存成功结果
        if use_cache and result.success:
            await self.cache_manager.set(server_url, tool_name, parameters, result.result)

        return result

    async def _call_tool_with_retry(
        self,
        server_url: str,
        tool_name: str,
        parameters: Dict[str, Any]
    ) -> MCPExecutionResult:
        for attempt in range(self.max_retries):
            try:
                result = await self._call_tool_once(server_url, tool_name, parameters)
                if result.success:
                    return result
                # 判断是否可重试
                if self._is_retryable_error(result.error_message or ""):
                    if attempt < self.max_retries - 1:
                        await asyncio.sleep(self.retry_delay * (2 ** attempt))  # 指数退避
                        continue
                return result
            except Exception as e:
                if attempt < self.max_retries - 1:
                    await asyncio.sleep(self.retry_delay * (2 ** attempt))
                    continue
                return MCPExecutionResult(success=False, error_message=str(e))

4.3 降级策略

async def execute_with_fallback(
    self,
    primary_selections: List[MCPToolSelection],
    fallback_selections: Optional[List[MCPToolSelection]] = None
) -> Dict[str, Any]:
    """带降级的工具执行"""
    result = await self.execute_parallel_primary(primary_selections)

    # 如果主要工具全部失败，尝试降级方案
    if len(result["failed_calls"]) == len(primary_selections) and fallback_selections:
        print("主要工具全部失败，尝试降级方案...")
        fallback_result = await self.execute_parallel_primary(fallback_selections)
        result["fallback_attempted"] = True
        result["fallback_results"] = fallback_result

    return result

---

五、LangGraph 集成模式

5.1 StateGraph 集成

from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain_mcp_adapters.client import MultiServerMCPClient

async def build_agent_with_mcp():
    async with MultiServerMCPClient(server_configs) as mcp_client:
        tools = mcp_client.get_tools()

        workflow = StateGraph(MessagesState)

        # Agent 节点：LLM 决策
        workflow.add_node("agent", call_model)

        # Tool 节点：执行 MCP 工具
        workflow.add_node("tools", ToolNode(tools))

        # 条件路由
        workflow.add_edge("agent", tools_condition)  # 有工具调用 → tools
        workflow.add_edge("tools", "agent")          # 执行完 → 回到 agent

        return workflow.compile()

5.2 条件路由（MCP 判断）

def _should_use_mcp(self, state: AgentState) -> str:
    """判断是否需要走 MCP 工具"""
    if state.get("tool_selections"):
        return "mcp"      # → mcp_executor
    return "direct"       # → 直接 executor

# 添加条件边
graph.add_conditional_edges(
    "tool_selector",
    self._should_use_mcp,
    {
        "mcp": "mcp_executor",
        "direct": "executor"
    }
)

5.3 多 Agent 编排

from langgraph_supervisor import create_supervisor

async with MultiServerMCPClient(server_configs) as mcp:
    # 按领域拆分工具
    data_tools = [t for t in mcp.get_tools() if t.name.startswith("data_")]
    code_tools = [t for t in mcp.get_tools() if t.name.startswith("code_")]
    search_tools = [t for t in mcp.get_tools() if t.name.startswith("search_")]

    # 创建专业 Agent
    data_agent = create_react_agent(model, data_tools, name="data_expert")
    code_agent = create_react_agent(model, code_tools, name="code_expert")
    search_agent = create_react_agent(model, search_tools, name="search_expert")

    # Supervisor 编排
    supervisor = create_supervisor(
        agents=[data_agent, code_agent, search_agent],
        model=model,
        prompt="根据任务类型分配给对应专家"
    )

---

六、最佳实践

6.1 Server 拆分

✅ 一个 MCP Server 管一个领域（数据库 / 文件 / 搜索 / 代码分析）
❌ 一个巨大的 MCP Server 包含所有工具

6.2 工具命名

# ✅ 带前缀，避免冲突
"data_query_sql"
"data_export_csv"
"code_lint_python"

# ❌ 通用名
"query"
"export"
"lint"

6.3 工具描述

@mcp.tool()
async def query_database(sql: str, database: str = "main") -> str:
    """执行 SQL 查询（只读）。

    Args:
        sql: SQL 查询语句，仅支持 SELECT
        database: 数据库名称，默认 main

    Returns:
        JSON 格式的查询结果

    Note:
        - 仅支持只读查询
        - 结果最多返回 1000 行
        - 超时时间 45 秒
    """
    # ...

6.4 安全原则

• 敏感操作（写文件、发邮件、执行代码）必须 Human-in-the-Loop
• 权限控制：Server 端实现权限检查，Client 端传递用户身份
• 输入验证：Server 端验证参数，防止注入攻击
• 超时控制：设置合理的超时时间，防止长时间阻塞

6.5 性能优化

工具数量 < 20   → 直接全部暴露给 LLM
工具数量 20-100 → 使用 ToolSelector 按任务筛选
工具数量 > 100  → 分层选择：先选类别 → 再选具体工具

---

七、检查清单

□ 理解连接类型：stdio / Streamable HTTP
□ 掌握连接池管理：Per-Server Session、超时、优雅关闭
□ 掌握工具注册：装饰器、动态注册、输出 Schema
□ 理解工具定义：参数、能力、任务类型、优先级
□ 掌握工具发现：list_tools() 动态发现
□ 掌握工具选择：4 策略（任务/关键词/能力/上下文）
□ 掌握并行执行：信号量控制并发、缓存、重试
□ 掌握降级策略：主要工具失败时自动切换备用
□ 掌握 LangGraph 集成：StateGraph + ToolNode + 条件路由
□ 遵循最佳实践：Server 拆分、命名规范、安全原则

---

八、参考资源

资源	链接
MCP 官方规范	https://modelcontextprotocol.io/specification
MCP Python SDK	https://github.com/modelcontextprotocol/python-sdk
langchain-mcp-adapters	https://github.com/langchain-ai/langchain-mcp-adapters
本项目 MCP 实现	src/mcp/ (client/registry/selector/executor)
本项目 MCP 示例	examples/mcp_integration/file_tools.py
LangGraph 多 Agent	https://langchain-ai.github.io/langgraph/concepts/multi_agent/

---

核心思想：MCP 是多 Agent 系统的 USB-C 接口，通过统一的协议实现工具发现、注册、选择、执行。掌握连接管理和工具选择机制，是构建高效多 Agent 系统的关键。