Agent 工具发现机制：从静态注册到动态发现

第一代：静态注册 (2023-2024)
┌─────────────────────────────┐
│         Agent 代码           │
│  ┌──────┐ ┌──────┐ ┌──────┐│
│  │search│ │calc  │ │write ││  ← 硬编码在代码中
│  └──────┘ └──────┘ └──────┘│
└─────────────────────────────┘
问题：添加工具需要改代码、重新部署

第二代：配置化注册 (2024-2025)
┌─────────────────────────────┐
│         Agent 运行时         │
│  ┌─────────────────────┐   │
│  │   Tool Registry     │   │  ← 从配置文件/YAML加载
│  │  ┌──┐┌──┐┌──┐┌──┐ │   │
│  │  │S ││C ││W ││E │ │   │
│  │  └──┘└──┘└──┘└──┘ │   │
│  └─────────────────────┘   │
└─────────────────────────────┘
问题：工具集固定，无法按需扩展

第三代：动态发现 (2025-2026)
┌─────────────────────────────┐
│         Agent 运行时         │
│  ┌─────────────────────┐   │
│  │  Tool Discovery     │   │
│  │  ┌────────────────┐ │   │
│  │  │  MCP Registry  │ │   │  ← 标准化协议
│  │  │  ┌─┐┌─┐┌─┐┌─┐ │ │   │
│  │  │  │S││C││W││?│ │ │   │  ← 动态发现
│  │  │  └─┘└─┘└─┘└─┘ │ │   │
│  │  └────────────────┘ │   │
│  └─────────────────────┘   │
└─────────────────────────────┘
优势：即插即用、运行时扩展、生态共享

# 硬编码工具——最简单但最不灵活
class SimpleAgent:
    def __init__(self, llm):
        self.llm = llm
        self.tools = {
            "search": self._search,
            "calculate": self._calculate,
            "write_file": self._write_file,
        }
    
    async def run(self, query: str) -> str:
        # 将工具定义注入 Prompt
        tools_desc = "\n".join(
            f"- {name}: {func.__doc__}" 
            for name, func in self.tools.items()
        )
        
        prompt = f"""Tools available:
{tools_desc}

To use a tool, respond with JSON: {{"tool": "name", "args": {{...}}}}

User: {query}"""
        
        response = await self.llm.invoke(prompt)
        # 解析并执行工具
        ...

from pydantic import BaseModel
import yaml

class ToolDefinition(BaseModel):
    name: str
    description: str
    input_schema: dict       # JSON Schema
    endpoint: str            # HTTP endpoint
    method: str = "POST"
    auth: dict | None = None
    timeout_ms: int = 10000
    tags: list[str] = []

class ConfigurableToolRegistry:
    """从配置文件加载工具"""
    
    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}
    
    def load_from_yaml(self, path: str):
        with open(path) as f:
            config = yaml.safe_load(f)
        
        for tool_config in config["tools"]:
            tool = ToolDefinition(**tool_config)
            self._tools[tool.name] = tool
    
    def load_from_directory(self, dir_path: str):
        """从目录中加载所有工具配置"""
        import os
        for filename in os.listdir(dir_path):
            if filename.endswith((".yaml", ".yml", ".json")):
                self.load_from_file(os.path.join(dir_path, filename))
    
    async def execute(self, tool_name: str, args: dict) -> dict:
        tool = self._tools.get(tool_name)
        if not tool:
            raise ToolNotFoundError(tool_name)
        
        # 参数校验
        self._validate_args(args, tool.input_schema)
        
        # HTTP 调用
        async with httpx.AsyncClient(timeout=tool.timeout_ms/1000) as client:
            response = await client.request(
                tool.method,
                tool.endpoint,
                json=args,
                headers=self._build_auth_headers(tool.auth)
            )
            return response.json()
    
    def get_openai_tools_schema(self) -> list[dict]:
        """生成 OpenAI Function Calling 格式"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.input_schema,
                }
            }
            for tool in self._tools.values()
        ]

# tools/research.yaml
tools:
  - name: web_search
    description: "Search the web for current information"
    input_schema:
      type: object
      properties:
        query:
          type: string
          description: "Search query"
        max_results:
          type: integer
          default: 5
      required: [query]
    endpoint: "https://api.search.example.com/v1/search"
    method: POST
    auth:
      type: api_key
      header: "X-API-Key"
      secret_ref: "search-api-key"
    timeout_ms: 5000
    tags: [research, read-only]

  - name: send_email
    description: "Send an email to specified recipients"
    input_schema:
      type: object
      properties:
        to:
          type: array
          items: {type: string}
        subject: {type: string}
        body: {type: string}
      required: [to, subject, body]
    endpoint: "https://internal-mail.example.com/send"
    method: POST
    auth:
      type: bearer
      secret_ref: "mail-token"
    timeout_ms: 10000
    tags: [communication, write]

┌───────────────────────────────────────────┐
│              MCP 协议原语                  │
├───────────┬───────────┬───────────────────┤
│  Tools    │ Resources │    Prompts        │
│ (可执行)  │ (可读取)  │   (可使用)        │
├───────────┼───────────┼───────────────────┤
│ 搜索网页  │ 读取文件  │ 代码审查模板      │
│ 执行代码  │ 查询数据库│ 分析报告模板      │
│ 发送邮件  │ 访问API   │ 调试指南          │
└───────────┴───────────┴───────────────────┘

from mcp import Server, Tool
from mcp.types import Tool as ToolType, TextContent

class FileSystemMCPServer(Server):
    """文件系统 MCP 服务器"""
    
    def __init__(self, root_path: str):
        super().__init__("filesystem-server", "1.0.0")
        self.root = Path(root_path)
    
    @self.list_tools()
    async def list_tools(self) -> list[ToolType]:
        """动态返回可用工具列表"""
        tools = [
            ToolType(
                name="read_file",
                description="Read the contents of a file",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {
                            "type": "string",
                            "description": "Relative path to the file"
                        }
                    },
                    "required": ["path"]
                }
            ),
            ToolType(
                name="list_directory",
                description="List files in a directory",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {
                            "type": "string",
                            "default": "."
                        }
                    }
                }
            ),
            ToolType(
                name="search_files",
                description="Search for files matching a pattern",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "pattern": {"type": "string"},
                        "path": {"type": "string", "default": "."}
                    },
                    "required": ["pattern"]
                }
            ),
        ]
        
        # 根据权限动态添加写入工具
        if self._has_write_permission():
            tools.append(ToolType(
                name="write_file",
                description="Write content to a file",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string"},
                        "content": {"type": "string"}
                    },
                    "required": ["path", "content"]
                }
            ))
        
        return tools
    
    @self.call_tool()
    async def call_tool(
        self,
        name: str,
        arguments: dict
    ) -> list[TextContent]:
        """执行工具调用"""
        if name == "read_file":
            path = self.root / arguments["path"]
            if not path.is_relative_to(self.root):
                raise PermissionError("Path traversal detected")
            content = path.read_text()
            return [TextContent(type="text", text=content)]
        
        elif name == "list_directory":
            path = self.root / arguments.get("path", ".")
            entries = [str(p.relative_to(self.root)) for p in path.iterdir()]
            return [TextContent(type="text", text="\n".join(entries))]
        
        elif name == "search_files":
            import fnmatch
            pattern = arguments["pattern"]
            search_path = self.root / arguments.get("path", ".")
            matches = [
                str(p.relative_to(self.root))
                for p in search_path.rglob(pattern)
            ]
            return [TextContent(type="text", text="\n".join(matches))]

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

class MCPToolDiscovery:
    """MCP 动态工具发现客户端"""
    
    def __init__(self):
        self.servers: dict[str, ClientSession] = {}
        self.discovered_tools: dict[str, DiscoveredTool] = {}
    
    async def connect_server(
        self,
        name: str,
        command: str,
        args: list[str] = None,
        env: dict = None
    ):
        """连接 MCP 服务器"""
        server_params = StdioServerParameters(
            command=command,
            args=args or [],
            env=env
        )
        
        # 建立连接
        stdio_transport = await stdio_client(server_params)
        session = ClientSession(*stdio_transport)
        await session.initialize()
        
        self.servers[name] = session
        
        # 发现工具
        await self._refresh_tools(name)
    
    async def _refresh_tools(self, server_name: str):
        """刷新服务器的工具列表"""
        session = self.servers[server_name]
        result = await session.list_tools()
        
        for tool in result.tools:
            discovered = DiscoveredTool(
                name=tool.name,
                description=tool.description,
                input_schema=tool.inputSchema,
                server=server_name
            )
            self.discovered_tools[f"{server_name}.{tool.name}"] = discovered
    
    async def execute_tool(
        self,
        full_name: str,
        args: dict
    ) -> str:
        """执行动态发现的工具"""
        tool = self.discovered_tools.get(full_name)
        if not tool:
            raise ToolNotFoundError(f"Unknown tool: {full_name}")
        
        session = self.servers[tool.server]
        result = await session.call_tool(tool.name, args)
        
        # 合并返回内容
        return "\n".join(
            content.text for content in result.content
            if hasattr(content, 'text')
        )
    
    def get_all_tools_schema(self) -> list[dict]:
        """获取所有已发现工具的 Schema"""
        return [
            {
                "type": "function",
                "function": {
                    "name": full_name,
                    "description": tool.description,
                    "parameters": tool.input_schema,
                }
            }
            for full_name, tool in self.discovered_tools.items()
        ]
    
    async def auto_discover(self):
        """自动发现网络中的 MCP 服务器"""
        # 通过 mDNS / 服务发现找到 MCP 服务器
        discovered = await self.service_discovery.find_services("_mcp._tcp")
        
        for service in discovered:
            try:
                await self.connect_server(
                    name=service.name,
                    command="npx",
                    args=["-y", "@mcp/remote", f"--url={service.url}"]
                )
                logger.info(f"Discovered and connected: {service.name}")
            except Exception as e:
                logger.warning(f"Failed to connect {service.name}: {e}")

class DynamicAgent:
    """支持动态工具发现的 Agent"""
    
    def __init__(self, llm, mcp_discovery: MCPToolDiscovery):
        self.llm = llm
        self.mcp = mcp_discovery
    
    async def run(self, query: str, max_iterations: int = 10) -> str:
        messages = [{"role": "user", "content": query}]
        
        for i in range(max_iterations):
            # 每次迭代刷新可用工具
            tools = self.mcp.get_all_tools_schema()
            
            # LLM 决策
            response = await self.llm.invoke(
                messages=messages,
                tools=tools,
                tool_choice="auto"
            )
            
            if not response.tool_calls:
                return response.content
            
            messages.append(response)
            
            # 执行工具
            for tool_call in response.tool_calls:
                try:
                    result = await self.mcp.execute_tool(
                        tool_call.function.name,
                        json.loads(tool_call.function.arguments)
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
                except Exception as e:
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": f"Error: {e}"
                    })
        
        return "Max iterations reached"

class ToolSecurityValidator:
    """动态工具安全验证"""
    
    async def validate_discovered_tool(
        self,
        tool: DiscoveredTool
    ) -> SecurityAssessment:
        """验证动态发现的工具安全性"""
        
        risks = []
        
        # 1. 检查工具来源可信度
        if not await self._is_trusted_server(tool.server):
            risks.append(Risk(
                level="high",
                reason=f"Untrusted MCP server: {tool.server}"
            ))
        
        # 2. 分析工具能力
        dangerous_patterns = [
            ("delete", "remove", "drop"),  # 破坏性操作
            ("execute", "eval", "system"),  # 代码执行
            ("send", "upload", "post"),     # 数据外传
        ]
        
        tool_desc_lower = tool.description.lower()
        tool_name_lower = tool.name.lower()
        
        for patterns in dangerous_patterns:
            if any(p in tool_name_lower or p in tool_desc_lower for p in patterns):
                risks.append(Risk(
                    level="medium",
                    reason=f"Tool may perform sensitive operation: {tool.name}"
                ))
        
        # 3. Schema 分析
        if self._has_unbounded_input(tool.input_schema):
            risks.append(Risk(
                level="medium",
                reason="Tool accepts unbounded input"
            ))
        
        return SecurityAssessment(
            tool=tool.name,
            risks=risks,
            requires_approval=any(r.level == "high" for r in risks),
            safe_to_use=all(r.level != "high" for r in risks)
        )

# 避免命名冲突：使用 server.tool 命名空间
"filesystem.read_file"      # 文件系统服务器的读文件
"github.read_file"          # GitHub 服务器的读文件
"database.read_file"        # 数据库服务器的读文件

# 工具别名：为常用工具提供短名
TOOL_ALIASES = {
    "search": "filesystem.search_files",
    "read": "filesystem.read_file",
}

class ToolCache:
    """工具描述缓存，减少每次请求的工具列表长度"""
    
    async def get_relevant_tools(
        self,
        query: str,
        all_tools: list[DiscoveredTool]
    ) -> list[DiscoveredTool]:
        """根据查询语义筛选相关工具"""
        
        # 1. 关键词匹配
        keyword_matches = [
            t for t in all_tools
            if any(kw in query.lower() for kw in t.keywords)
        ]
        
        # 2. 语义匹配
        query_embedding = await self.embedder.embed(query)
        tool_embeddings = await self._get_tool_embeddings(all_tools)
        
        similarities = cosine_similarity(query_embedding, tool_embeddings)
        semantic_matches = [
            all_tools[i] for i in np.argsort(similarities)[-10:]
            if similarities[i] > 0.5
        ]
        
        # 3. 合并去重
        relevant = list(set(keyword_matches + semantic_matches))
        
        # 4. 总是包含核心工具
        core_tools = [t for t in all_tools if "core" in t.tags]
        
        return list(set(relevant + core_tools))

□ 工具配置与代码分离（YAML/JSON/数据库）
□ 支持 MCP 协议的标准化工具接口
□ 动态发现的工具经过安全验证
□ 工具命名空间避免冲突
□ 工具列表按相关性筛选（避免 Context 膨胀）
□ 工具可用性健康检查
□ 工具版本兼容性管理
□ 工具权限分级（只读/写入/危险）
□ 工具调用审计日志
□ 支持工具热加载（不重启 Agent）