高级 MCP 用法

Agno 的 MCP 集成还支持处理与多个服务器的连接、指定服务器参数以及使用您自己的 MCP 服务器：

连接到多个 MCP 服务器

您可以通过使用 MultiMCPTools 类在单个代理中同时使用多个 MCP 服务器。

multiple_mcp_servers.py

import asyncio
import os

from agno.agent import Agent
from agno.tools.mcp import MultiMCPTools


async def run_agent(message: str) -> None:
    """使用给定的消息运行 Airbnb 和 Google Maps 代理。"""

    env = {
        **os.environ,
        "GOOGLE_MAPS_API_KEY": os.getenv("GOOGLE_MAPS_API_KEY"),
    }

    async with MultiMCPTools(
        [
            "npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",
            "npx -y @modelcontextprotocol/server-google-maps",
        ],
        env=env,
    ) as mcp_tools:
        agent = Agent(
            tools=[mcp_tools],
            markdown=True,
            show_tool_calls=True,
        )

        await agent.aprint_response(message, stream=True)


# 示例用法
if __name__ == "__main__":
    # Pull request 示例
    asyncio.run(
        run_agent(
            "2025年8月1日至4日，开普敦有哪些适合2人入住3晚的房源？"
        )
    )

理解服务器参数

配置 MCPTools 或 MultiMCPTools 的推荐方式是使用 command 或 url 参数。

或者，您可以使用 MCPTools 的 server_params 参数来更详细地配置到 MCP 服务器的连接。

当使用 stdio 传输时，server_params 参数应为 StdioServerParameters 的实例。它包含以下键：

• command: 运行 MCP 服务器的命令。
  • 对于可以通过 npm 安装的 mcp 服务器，请使用 npx（或在 Windows 上运行则使用 node）。
  • 对于可以通过 uvx 安装的 mcp 服务器，请使用 uvx。
• args: 传递给 MCP 服务器的参数。
• env: 可选的要传递给 MCP 服务器的环境变量。请记住将所有当前环境变量包含在 env 字典中。如果未提供 env，则将使用当前环境变量。 例如：

{
    **os.environ,
    "GOOGLE_MAPS_API_KEY": os.getenv("GOOGLE_MAPS_API_KEY"),
}

当使用 SSE 传输时，server_params 参数应为 SSEClientParams 的实例。它包含以下字段：

• url: MCP 服务器的 URL。
• headers: 传递给 MCP 服务器的标头（可选）。
• timeout: 到 MCP 服务器连接的超时时间（可选）。
• sse_read_timeout: SSE 连接本身超时的等待时间（可选）。

当使用 Streamable HTTP 传输时，server_params 参数应为 StreamableHTTPClientParams 的实例。它包含以下字段：

• url: MCP 服务器的 URL。
• headers: 传递给 MCP 服务器的标头（可选）。
• timeout: 到 MCP 服务器连接的超时时间（可选）。
• sse_read_timeout: 客户端等待新事件断开连接的最长时间（以秒为单位）。所有其他 HTTP 操作由 timeout 控制（可选）。
• terminate_on_close: 客户端关闭时是否终止连接（可选）。

更大的灵活性

您也可以自己创建 MCP 服务器并将其传递给 MCPTools 构造函数。

filesystem_agent.py

import asyncio
from pathlib import Path
from textwrap import dedent

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def create_filesystem_agent(session):
    """创建并配置一个带有 MCP 工具的文件系统代理。"""
    # 初始化 MCP 工具包
    mcp_tools = MCPTools(session=session)
    await mcp_tools.initialize()

    # 创建一个带有 MCP 工具包的代理
    return Agent(
        model=OpenAIChat(id="gpt-4o"),
        tools=[mcp_tools],
        instructions=dedent("""\
            你是一个文件系统助手。帮助用户浏览文件和目录。

            - 浏览文件系统以回答问题
            - 使用 list_allowed_directories 工具查找您可以访问的目录
            - 提供您检查的文件的清晰上下文
            - 使用标题组织您的回应
            - 保持简洁并专注于相关信息\
        """),
        markdown=True,
        show_tool_calls=True,
    )


async def run_agent(message: str) -> None:
    """使用给定的消息运行文件系统代理。"""

    # 初始化 MCP 服务器
    server_params = StdioServerParameters(
        command="npx",
        args=[
            "-y",
            "@modelcontextprotocol/server-filesystem",
            str(Path(__file__).parent.parent.parent.parent),  # 将此设置为您要探索的项目根目录
        ],
    )

    # 创建一个客户端会话以连接到 MCP 服务器
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            agent = await create_filesystem_agent(session)

            # 运行代理
            await agent.aprint_response(message, stream=True)


# 示例用法
if __name__ == "__main__":
    # 基本示例 - 探索项目许可证
    asyncio.run(run_agent("这个项目的许可证是什么？"))