模型上下文协议 (MCP) 使代理能够通过标准化接口与外部系统进行交互。
您可以使用 Agno 的 MCP 集成将您的代理连接到任何 MCP 服务器。
查找要使用的 MCP 服务器
您可以使用任何可正常工作的 MCP 服务器。要查看一些示例,您可以查看由 MCP 维护者自己维护的 此 GitHub 存储库。
初始化 MCP 集成
将 MCPTools
类初始化为上下文管理器。定义 MCP 服务器的推荐方法是使用 command
或 url
参数。使用 command
,您可以传递用于运行所需 MCP 服务器的命令。使用 url
,您可以传递正在运行的所需 MCP 服务器的 URL。
例如,要使用 “mcp-server-git” 服务器,您可以这样做:
from agno.tools.mcp import MCPTools
async with MCPTools(command=f"uvx mcp-server-git") as mcp_tools:
...
将 MCPTools 提供给代理
初始化代理时,请在 tools
参数中传递 MCPTools
类。
现在代理已准备好使用 MCP 服务器:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools
async with MCPTools(command=f"uvx mcp-server-git") as mcp_tools:
# 设置并运行代理
agent = Agent(model=OpenAIChat(id="gpt-4o"), tools=[mcp_tools])
await agent.aprint_response("What is the license for this project?", stream=True)
基本示例:文件系统代理
这是一个文件系统代理,它使用 Filesystem MCP 服务器 来浏览和分析文件:
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 StdioServerParameters
async def run_agent(message: str) -> None:
"""使用给定的消息运行文件系统代理。"""
file_path = str(Path(__file__).parent.parent.parent.parent)
# 用于访问文件系统(通过 `npx`)的 MCP 服务器
async with MCPTools(f"npx -y @modelcontextprotocol/server-filesystem {file_path}") as mcp_tools:
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
tools=[mcp_tools],
instructions=dedent("""\
您是一个文件系统助手。帮助用户浏览文件和目录。
- 导航文件系统以回答问题
- 使用 list_allowed_directories 工具查找您可以访问的目录
- 提供您检查过的文件的清晰上下文
- 使用标题组织您的响应
- 简洁明了,专注于相关信息\
"""),
markdown=True,
show_tool_calls=True,
)
# 运行代理
await agent.aprint_response(message, stream=True)
# 示例用法
if __name__ == "__main__":
# 基本示例 - 探索项目许可证
asyncio.run(run_agent("What is the license for this project?"))
在 Agno Playground 中使用 MCP
您也可以在 Agno Playground 中运行 MCP 服务器,它提供了一个用于与您的代理交互的 Web 界面。以下是在 Playground 中运行的 GitHub 代理的示例:
import asyncio
from os import getenv
from textwrap import dedent
import nest_asyncio
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.playground import Playground
from agno.storage.agent.sqlite import SqliteAgentStorage
from agno.tools.mcp import MCPTools
# 允许嵌套事件循环
nest_asyncio.apply()
agent_storage_file: str = "tmp/agents.db"
async def run_server() -> None:
"""运行 GitHub 代理服务器。"""
github_token = getenv("GITHUB_TOKEN") or getenv("GITHUB_ACCESS_TOKEN")
if not github_token:
raise ValueError("需要 GITHUB_TOKEN 环境变量")
# 创建一个客户端会话以连接到 MCP 服务器
async with MCPTools("npx -y @modelcontextprotocol/server-github") as mcp_tools:
agent = Agent(
name="MCP GitHub Agent",
tools=[mcp_tools],
instructions=dedent("""\
您是一个 GitHub 助手。帮助用户浏览存储库及其活动。
- 使用标题组织您的响应
- 简洁明了,专注于相关信息\
"""),
model=OpenAIChat(id="gpt-4o"),
storage=SqliteAgentStorage(
table_name="basic_agent",
db_file=agent_storage_file,
auto_upgrade_schema=True,
),
add_history_to_messages=True,
num_history_responses=3,
add_datetime_to_instructions=True,
markdown=True,
)
playground = Playground(agents=[agent])
app = playground.get_app()
# 提供应用程序,同时保持 MCPTools 上下文管理器处于活动状态
playground.serve(app)
if __name__ == "__main__":
asyncio.run(run_server())
最佳实践
-
错误处理: 始终包含对 MCP 服务器连接和操作的适当错误处理。
-
资源清理: 将 MCPTools
或 MultiMCPTools
用作异步上下文管理器,以确保正确清理资源:
async with MCPTools(command) as mcp_tools:
# 您的代理代码在这里
- 清晰的说明: 为您的代理提供清晰具体的说明:
instructions = """
您是一个文件系统助手。帮助用户浏览文件和目录。
- 导航文件系统以回答问题
- 使用 list_allowed_directories 工具查找可访问的目录
- 提供您检查过的文件的清晰上下文
- 简洁明了,专注于相关信息
"""
更多信息
Responses are generated using AI and may contain mistakes.