今天來讓 agent-brain 可以外接 MCP servers，讓我們的 AI agent 能夠使用更多外部工具。

為什麼要整合 MCP？

還記得昨天介紹的 MCP 嗎？它就像是 AI 的「USB Hub」，可以讓我們的 agent 連接各種外部服務。

今天要做的事情很簡單：
讓 agent-brain 在初始化時，可以直接掛載 MCP servers

而且使用方式超級簡單，只要傳一個字串就好：

tools: list[BaseTool | str] = [
    GetUserInfoTool(),  # 自定義的 tool
    "http://127.0.0.1:9000/mcp",  # streamable http mcp server
    "./mcp-server/server_stdio.py",  # stdio mcp server
]

async def create_brain():
    return await Brain.create(net="ReAct", memory="Messages", tools=tools)
    
brain = asyncio.run(create_brain())
async for raw in brain.answer(user_msg):
    ...

是不是很直覺？不管是 HTTP endpoint 還是本地的 Python 檔案，都可以直接丟進去。

實作：FastMCPTool

核心思路很簡單：
用 fastmcp 的 Client 把 MCP server 包裝成 BaseTool，這樣 agent-brain 就能無痛使用了。

from fastmcp import Client
from mcp.types import ContentBlock

class FastMCPTool(BaseTool):
    def __init__(self, endpoint: str, name: str, description: str, parameters: dict):
        self.endpoint = endpoint
        super().__init__(name, description, parameters)
    
    @classmethod
    async def from_fastmcp_endpoints(cls, endpoint: str) -> list["FastMCPTool"]:
        tool_maps: dict[str, FastMCPTool] = {}
        try:
            async with Client(endpoint) as client:
                # 測試連線
                await client.ping()
                
                # 取得 server 提供的所有工具
                tools = await client.list_tools()
                
                # 把每個 tool 都包裝成 FastMCPTool
                for tool in tools:
                    tool_maps[tool.name] = FastMCPTool(
                        endpoint=endpoint,
                        name=tool.name,
                        description=tool.description or "",
                        parameters=tool.inputSchema,
                    )
            
            return list(tool_maps.values())
        except Exception as e:
            print(f"Error creating client for {endpoint}: {e}")
            return []
    
    async def execute(self, **kwargs) -> list[ContentBlock]:
        async with Client(self.endpoint) as client:
            result = await client.call_tool(
                name=self.name, 
                arguments=kwargs, 
                timeout=30
            )
            return result.content

運作流程

1. 初始化階段

   • from_fastmcp_endpoints 連上 MCP server
   • 呼叫 list_tools() 取得所有可用工具
   • 每個工具都轉換成一個 FastMCPTool instance

2. 執行階段

   • 當 agent 決定要用某個工具時
   • execute 方法會建立 client 連線
   • 呼叫 call_tool 並傳入參數
   • 回傳執行結果

整合到 Brain

有了 FastMCPTool 之後，只要在 memory.create 時做一點判斷就好：

async def register_tools(self, tools: list[BaseTool | str]) -> None:
    for tool in tools:
        # 如果是字串，就當成 MCP endpoint
        if isinstance(tool, str):
            fastmcp_tools = await FastMCPTool.from_fastmcp_endpoints(tool)
            for fastmcp_tool in fastmcp_tools:
                if fastmcp_tool.name in self.available_tools:
                    raise ValueError(f"Duplicate tool name: {fastmcp_tool.name}")
                self.available_tools[fastmcp_tool.name] = fastmcp_tool
        # 一般的 BaseTool
        elif isinstance(tool, BaseTool):
            if tool.name in self.available_tools:
                raise ValueError(f"Duplicate tool name: {tool.name}")
            self.available_tools[tool.name] = tool
        else:
            raise ValueError(f"Invalid tool type: {type(tool)}")

這樣一來，不管傳入的是：

• "http://127.0.0.1:9000/mcp" → 自動連線並載入所有 tools
• "./server_stdio.py" → 啟動本地 server 並載入 tools
• GetUserInfoTool() → 直接使用

全部都能正常運作！

To-do: 連線優化

目前的實作有個小問題：
每次 execute 都會重新建立 client 連線

async def execute(self, **kwargs) -> list[ContentBlock]:
    async with Client(self.endpoint) as client:  # 每次都重連
        result = await client.call_tool(...)
        return result.content

這樣在頻繁呼叫時會有效能損耗。

改進方向：
用 Singleton pattern 讓 Brain 在初始化時就建好連線，之後重複使用：

class FastMCPTool(BaseTool):
    _clients: dict[str, Client] = {}  # 快取 client instances
    
    @classmethod
    async def get_client(cls, endpoint: str) -> Client:
        if endpoint not in cls._clients:
            cls._clients[endpoint] = Client(endpoint)
            await cls._clients[endpoint].__aenter__()
        return cls._clients[endpoint]
    
    async def execute(self, **kwargs) -> list[ContentBlock]:
        client = await self.get_client(self.endpoint)
        result = await client.call_tool(...)
        return result.content

這樣就能避免重複連線，提升執行效率。

小結

今天完成了 agent-brain 與 MCP 的整合，現在我們可以：

• ✅ 用一個字串就載入整個 MCP server 的所有工具
• ✅ 同時支援 stdio 和 HTTP 兩種傳輸方式
• ✅ 無痛混用自定義 tool 和 MCP tools

下次當我們需要讓 agent 連接 Slack、Gmail、Google Calendar 時，只要：

tools = [
    "http://mcp-slack.example.com/mcp",
    "http://mcp-gmail.example.com/mcp",
    "./mcp-calendar.py",
]

就搞定了！這就是 MCP 帶來的便利性 🚀

明天我們來實際測試一下，讓 agent 真的去呼叫這些外部服務！