mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-06-02 09:48:22 +08:00
b2429ec30c
Allows CowAgent to dynamically load tools from any MCP server at startup,
extending the agent from a fixed toolset to an open, extensible tool ecosystem.
## What's added
- `agent/tools/mcp/mcp_client.py`: lightweight JSON-RPC client supporting both
stdio (subprocess) and SSE (HTTP) transports — zero extra dependencies
- `agent/tools/mcp/mcp_tool.py`: `McpTool` wraps a single MCP tool as a
`BaseTool`, with dynamic name/description/params set at instance level
- `agent/tools/tool_manager.py`: new `_load_mcp_tools()` loads MCP servers at
startup via `McpClientRegistry`; falls back gracefully on any error; no-op
when `mcp_servers` is not configured
- `config.py`: registers `mcp_servers` in `available_setting` with inline docs
## Design
- No new dependencies — JSON-RPC implemented from scratch using stdlib only
- MCP clients are long-lived (initialized once, shared across tool calls)
- `McpClientRegistry` holds all subprocess handles and shuts them down cleanly
- Server init failures are non-fatal: logged as warnings, agent continues normally
- Zero overhead when `mcp_servers` is absent from config
## Config example
```json
"mcp_servers": [
{
"name": "filesystem",
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
}
]
```
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
from agent.tools.base_tool import BaseTool, ToolResult
|
|
from common.log import logger
|
|
|
|
|
|
class McpTool(BaseTool):
|
|
"""
|
|
将单个 MCP 工具包装为 BaseTool。
|
|
一个 MCP Server 可以提供多个工具，每个工具对应一个 McpTool 实例。
|
|
"""
|
|
|
|
def __init__(self, client, tool_schema: dict, server_name: str):
|
|
"""
|
|
:param client: 该工具所属的 McpClient 实例
|
|
:param tool_schema: MCP 返回的工具描述，格式：
|
|
{"name": str, "description": str, "inputSchema": dict}
|
|
:param server_name: Server 名称，用于日志
|
|
"""
|
|
self.client = client
|
|
self.server_name = server_name
|
|
self.name = tool_schema["name"]
|
|
self.description = tool_schema.get("description", "")
|
|
self.params = tool_schema.get("inputSchema", {})
|
|
|
|
def execute(self, params: dict) -> ToolResult:
|
|
logger.info(f"[McpTool] server={self.server_name} tool={self.name} params={params}")
|
|
try:
|
|
result = self.client.call_tool(self.name, params)
|
|
return ToolResult.success(result)
|
|
except Exception as e:
|
|
logger.error(f"[McpTool] server={self.server_name} tool={self.name} error: {e}")
|
|
return ToolResult.fail(str(e))
|