相关文档地址：https://docs.agno.com/tools/mcp/mcp#model-context-protocol-mcp
12306-mcp-server地址：https://github.com/drfccv/mcp-server-12306
学习目标
使用Agno创建一个集成MCP的Agent助手。
MCP Server
MCP支持三种传输方式：
SSE（Server-Sent Events）：使用 HTTP 协议和 Server-Sent Events 进行服务器向客户端的单向推送，适合实时数据更新，但已被标记为弃用- 标准Stdio：通过标准输入输出（stdin/stdout）与本地子进程通信，延迟低、配置简单，但只能本地使用。
- Streamable HTTP：基于 HTTP 的双向流式传输，支持无状态架构和云原生部署，是目前推荐用于远程连接的协议。
本篇采用Streamable HTTP方式调用12306-mcp-server。
部署12306-mcp-server（Docker）
# 直接拉取已构建镜像
docker pull drfccv/mcp-server-12306:latest
# 运行容器（映射8000端口）
docker run -d -p 8000:8000 --name mcp-server-12306 drfccv/mcp-server-12306:latest
创建Agent
安装Agno及相关依赖
# 包管理使用的uv
uv add agno
uv add openai
Agent
Agent是自主运行的AI程序，其核心是模型、工具和指令：
- 模型：控制执行流程。它决定是要推理、行动还是响应。
- 工具：使代理能够采取行动并与外部系统进行交互。
- 指令：用于指导Agent的行为和决策逻辑，教导它如何使用工具和响应。
除此之外，还有推理，知识库，存储，记忆能力：
- 推理：使Agent能够在响应前“思考”并“分析”其行动（即工具调用）的结果，这提高了响应的可靠性和质量。
- 知识库：特定领域的信息，代理可以在运行时搜索这些信息以做出更好的决策并提供准确的响应（RAG）。知识存储在向量数据库中，这种在运行时进行搜索的模式被称为
Agentic RAG，关于Agentic RAG。 - 存储：在数据库中保存会话历史和状态。因为模型API服务是无状态的，而存储历史的对话数据使我们能够从上次中断的地方继续对话。这使得Agent具有多轮、长期对话的能力。
- 记忆：使Agent能够存储和回忆先前与用户之间互动的信息，使他们能够了解用户偏好并个性化其响应。
模型（LLM）
模型是Agent的大脑，帮助它推理、行动并响应用户。模型越好，Agent越智能。
因此，在创建Agent之前，需要先创建Model。
Agno提供了多种连接LLM的方式，文档地址：https://docs.agno.com/models/introduction
包括：OpenAI，DeepSeek，Ollama等等，这里我选择的使用的是OpenAILike，只要你使用的模型提供商支持OpenAI API格式，通过替换base_url，就可以使用OpenAILike来访问。
import os
from agno.models.openai.like import OpenAILike
model = OpenAILike(
id="deepseek-v3", # 服务商提供的模型id
api_key=os.getenv("OPENAI_API_KEY"), # 服务商的apikey
base_url=os.getenv("OPENAI_BASE_URL"), # 服务商的api地址
)
当模型创建完毕，就是把大脑装入躯体中。
import os
from agno.models.openai.like import OpenAILike
from agno.agent import Agent
model = OpenAILike(
id="deepseek-v3", # 服务商提供的模型id
api_key=os.getenv("OPENAI_API_KEY"), # 服务商的apikey
base_url=os.getenv("OPENAI_BASE_URL"), # 服务商的api地址
)
agent = Agent(
model=model,
markdown=True,
)
agent.print_response("Say Hi!", stream=True)
工具（Tools）
工具赋予了Agent调用、执行外部函数或服务的能力，用于扩展其自身能力，完成仅靠“思考”无法直接完成的任务。假设你是一个“AI 厨师”，你的大脑（LLM）知道“番茄炒蛋”的做法，但厨房里没有食材（数据），也没有炉子（执行能力）。此时Tools就是：
- 一个“冰箱工具”（查询数据库）→ 获取番茄、鸡蛋等食材信息
- 一个“炉灶工具”（调用API）→ 启动加热程序
- 一个“计时器工具”（设置提醒）→ 控制炖煮时间
集成MCP服务
-
SSE (Server-Sent Events)
废弃不做演示 -
Streamable HTTP
import asyncio
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools, StreamableHTTPClientParams
async def streamable_http_example():
"""使用StreamableHTTP传输方式连接MCP服务器"""
# 方式1：直接使用URL和传输类型
server_url = "http://localhost:8000/mcp"
async with MCPTools(
url=server_url,
transport="streamable-http",
timeout_seconds=30
) as mcp_tools:
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
tools=[mcp_tools],
markdown=True,
)
await agent.aprint_response(
"查询项目许可证信息",
stream=True
)
# 方式2：使用详细的服务器参数配置
server_params = StreamableHTTPClientParams(
url="http://localhost:8000/mcp",
headers={
"Authorization": "Bearer your-api-token",
"Content-Type": "application/json",
"x-custom-header": "agno-client"
},
timeout=60.0,
sse_read_timeout=30.0,
terminate_on_close=True
)
async with MCPTools(server_params=server_params) as mcp_tools:
agent = Agent(tools=[mcp_tools])
# 代码逻辑
# 自动处理连接关闭
if __name__ == "__main__":
asyncio.run(streamable_http_example())
- Stdio（stdin/stdout）
import asyncio
import os
from pathlib import Path
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools, MultiMCPTools
from mcp import StdioServerParameters
async def stdio_example():
"""使用Stdio传输方式连接MCP服务器"""
# 方式1：简单的命令行连接
async with MCPTools(command="uvx mcp-server-git") as mcp_tools:
agent = Agent(tools=[mcp_tools])
# 代码逻辑
# 自动处理连接关闭
# 方式2：使用详细的服务器参数
github_token = os.getenv("GITHUB_TOKEN")
server_params = StdioServerParameters(
command="npx",
args=["-y", "@modelcontextprotocol/server-github"],
env={
**os.environ, # 继承当前环境变量
"GITHUB_TOKEN": github_token,
"NODE_ENV": "production"
}
)
async with MCPTools(server_params=server_params) as mcp_tools:
agent = Agent(tools=[mcp_tools])
# 代码逻辑
# 自动处理连接关闭
# 方式3：多服务器连接
env = {
**os.environ,
"GOOGLE_MAPS_API_KEY": os.getenv("GOOGLE_MAPS_API_KEY"),
"BRAVE_API_KEY": os.getenv("BRAVE_API_KEY"),
}
async with MultiMCPTools(
commands=[
"npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",
"npx -y @modelcontextprotocol/server-google-maps",
"npx -y @modelcontextprotocol/server-brave-search"
],
env=env,
timeout_seconds=45
) as multi_mcp:
agent = Agent(tools=[mcp_tools])
# 代码逻辑
# 自动处理连接关闭
if __name__ == "__main__":
asyncio.run(stdio_example())
至此支持算是完成Agent中模型和工具的添加，Agent可以根据你的指令自主选择是否使用工具并做出响应。
学习成果
railway_agent.py
import asyncio
import os
from textwrap import dedent
from agno.agent import Agent
from agno.models.openai import OpenAILike
from agno.tools.mcp import MCPTools
async def run_agent(message: str):
openai_model = OpenAILike(id=os.getenv("MODEL_NAME"), api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL"),
request_params={"extra_body": {"enable_thinking": False}})
async with MCPTools(
url='http://127.0.0.1:9000/mcp',
transport="streamable-http",
timeout_seconds=30
) as mcp_tools:
agent = Agent(model=openai_model,
markdown=True,
instructions=[
dedent("""
# 高铁出行助手
您是一个专业的高铁出行助手，专门帮助用户查询高铁车票信息、规划出行路线和提供中转方案。您具备以下核心能力：
## 核心功能
- 查询高铁车票信息（班次、时刻、票价、余票等）
- 提供最优出行路线规划
- 设计中转换乘方案
- 分析不同方案的性价比
- 提供实时列车状态信息
## 交互原则
1. **主动询问关键信息**：始终确认出发城市、目的城市、出行日期、时间偏好等必要信息
2. **多方案对比**：提供多个出行方案供用户选择，包括直达和中转选项
3. **详细说明**：清晰说明每个方案的优缺点（时间、价格、舒适度等）
4. **实时性提醒**：提醒用户车票信息可能实时变化，建议及时购买
## MCP工具调用
您可以调用以下MCP工具来获取准确信息：
- 车票查询服务：获取班次、时刻、票价、余票信息
- 路线规划服务：计算最优路径和中转方案
- 实时状态服务：查询列车正点、延误等状态
## 注意事项
- 始终基于MCP工具返回的实际数据回答
- 如果查询失败，主动说明并建议用户稍后重试
- 提醒用户关注购票时间限制和退改签政策
- 考虑用户的实际需求（时间敏感、价格敏感、舒适度等）
""")
],
tools=[mcp_tools])
await agent.aprint_response(message, stream=True)
if __name__ == '__main__':
os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxx"
os.environ["MODEL_NAME"] = "qwen3-235b-a22b"
os.environ["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"
asyncio.run(run_agent("查询今日南京到北京的高铁车票信息"))
执行代码打印：
┌─ Message ───────────────────────────────────────────────────────────────────┐
│ │
│ 查询今日南京到北京的高铁车票信息 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
┌─ Tool Calls ────────────────────────────────────────────────────────────────┐
│ │
│ • get-current-time(format=YYYY-MM-DD, timezone=Asia/Shanghai) │
│ • query-tickets(from_station=南京, to_station=北京, train_date=2025-08-21) │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
┌─ Response (27.4s) ──────────────────────────────────────────────────────────┐
│ │
│ 🚄 南京 → 北京 (2025-08-21) │
│ │
│ 📊 共找到 32 趟列车。以下是部分推荐车次： │
│ │
│ 🌟 推荐方案（高铁） │
│ │
│ │
│ 车次 出发时间 到达时间 历时 座位类型（余票） │
│ ─────────────────────────────────────────────────────── │
│ G28 20:02 23:18 03:16 商务座:3 │
│ G40 20:06 23:23 03:17 商务座:2 │
│ G26 19:08 22:36 03:28 商务座:12 │
│ G38 19:04 22:30 03:26 商务座:1 │
│ │
│ │
│ 🚆 其他热门车次 │
│ │
│ • G138（14:53 发车，历时 4h38m）：商务座:3 │
│ • G18（15:02 发车，历时 3h33m）：无座 │
│ • G142（15:37 发车，历时 4h33m）：商务座:10 │
│ • G20（16:01 发车，历时 3h35m）：商务座:1 | 一等座:1 │
│ • G146（16:36 发车，历时 4h29m）：商务座:11 | 一等座:3 │
│ │
│ 🛌 普通列车（过夜车） │
│ │
│ • T110（20:07 发车，次日 08:32 到达，历时 12h25m）：暂无票 │
│ • D18（21:38 发车，次日 07:09 到达，历时 9h31m）：暂无票 │
│ • D6（23:56 发车，次日 09:26 到达，历时 9h30m）：仅余无座票 │
│ │
│ ✅ 推荐理由 │
│ │
│ • G28 和 G40 是今晚最后两班高铁，适合下班后出行 │
│ • G26 和 G38 时间适中，座位选择较多 │
│ • 若考虑性价比，可选择 G146（16:36 发车，商务座:11，一等座:3） │
│ │
│ ▌ ⚠️ │
│ ▌ 请注意：车票信息实时变动，请尽快完成购票。若需进一步帮助（如中转方案、 │
│ ▌ 座位类型说明等），请随时告知！ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
进程已结束，退出代码为 0