13.4.1 为什么不直接调用 API
在 13.3 节中，我们设计了四个 Agent 来协作完成旅行规划任务。其中 AttractionSearchAgent、WeatherQueryAgent 和 HotelAgent 都需要调用高德地图的 API 来获取数据。一个自然的问题是：为什么不直接在 Agent 中调用高德地图的 HTTP API？
让我们先看看直接调用 API 会是什么样子。高德地图提供了 POI 搜索 API，我们需要构造 HTTP 请求，传递参数，解析响应：
import requests
def search_poi(keywords: str,city: str,api_key: str):
"""直接调用高德地图POI搜索API"""
url = "https://restapi.amap.com/v3/place/text"
params = {
"keywords": keywords,
"city": city,
"key": api_key,
"output": "json"
}
response = requests.get(url,params=params)
data = response.json()
return data
这种方式看起来很简单，但在实际使用中会遇到几个问题。首先是Agent 无法自主调用。在我们的 HelloAgents 框架中，Agent 通过识别提示词中的工具调用标记(比如[TOOL_CALL:tool_name:arg1=value1])来调用工具。如果我们直接在代码中调用 API，Agent 就失去了自主决策的能力，变成了一个简单的函数调用。
其次是参数传递复杂。高德地图的 API 有很多参数，比如 POI 搜索有keywords、city、types、offset、page等十几个参数。如果我们要让 Agent 能够灵活使用这些参数，就需要在提示词中详细说明每个参数的含义和格式，这会让提示词变得非常复杂。
第三是响应解析困难。高德地图 API 返回的是 JSON 格式的数据，结构比较复杂。我们需要编写代码来解析这些数据，提取我们需要的字段。如果 API 的响应格式发生变化，我们就需要修改解析代码。
最后是工具管理混乱。高德地图提供了十几个不同的 API(POI 搜索、天气查询、路线规划等)，如果我们为每个 API 都编写一个函数，然后手动注册到 Agent 的工具列表中，代码会变得很冗长。而且当我们想添加新的 API 时，需要修改多个地方。
13.4.2 高德地图 MCP 集成
MCP(Model Context Protocol)是 Anthropic 提出的标准化协议，用于连接 LLM 和外部工具。本节将介绍如何在项目中集成高德地图 MCP 服务器。我们的项目用的是amap-mcp-server，这是一个用 Node.js 实现的 MCP 服务器：
图 13.7 amap-mcp-server 工具
高德地图 MCP 服务器提供了多种工具，主要分为以下类别，如表 13.1 所示:
表 13.1 高德地图 MCP 工具分类
通过 MCP 协议，我们可以很方便地在 HelloAgents 中集成:
from hello_agents.tools import MCPTool
from app.config import get_settings
settings = get_settings()
# 创建MCP工具
mcp_tool = MCPTool(
name="amap_mcp",
command="npx",
args=["-y", "@sugarforever/amap-mcp-server"],
env={"AMAP_API_KEY": settings.amap_api_key},
auto_expand=True
)
这段代码做了什么呢？首先，command和args指定了如何启动 MCP 服务器。npx -y @sugarforever/amap-mcp-server会从 npm 仓库下载并运行amap-mcp-server这个包。env参数传递了环境变量，这里我们传递了高德地图的 API 密钥。
**注意：**本文档中部分示例使用 npx 启动 MCP（Model Context Protocol）服务。而在本节代码仓中，我们实际采用的是 uvx 方式。需要说明的是，npx 和 uvx 在设计理念上高度一致，区别仅在于所处的生态系统，npx 面向 JavaScript/Node.js（包来自 npm），而uvx 面向 Python（包来自 PyPI）。两种方式并无优劣之分，请大家在使用时按需进行选择。
当我们创建MCPTool对象时，它会在后台启动 MCP 服务器进程，并通过标准输入输出(stdin/stdout)与服务器通信。这是 MCP 协议的一个特点：使用进程间通信而不是 HTTP，这样更高效，也更容易管理。
最关键的是auto_expand=True这个参数。当设置为 True 时，MCPTool会自动查询 MCP 服务器提供了哪些工具，然后为每个工具创建一个独立的 Tool 对象。这就是为什么我们只创建了一个MCPTool，但 Agent 却获得了 16 个工具。让我们看看这个过程：
# 创建一个MCPTool
mcp_tool = MCPTool(..., auto_expand=True)
agent.add_tool(mcp_tool)
# Agent实际上获得了16个工具！
print(list(agent.tools.keys()))
# ['amap_maps_text_search', 'amap_maps_weather', ...]
如图 13.8 所示，假设用户想搜索北京的景点，AttractionSearchAgent 接收到查询"请搜索北京的历史文化景点"。Agent 分析这个查询，决定调用amap_maps_text_search工具，参数是keywords=景点，city=北京。
图 13.8 MCP 工具调用流程
Agent 生成工具调用标记：[TOOL_CALL:amap_maps_text_search:keywords=景点，city=北京]。HelloAgents 框架解析这个标记，提取工具名称和参数，然后调用对应的 Tool 对象。
Tool 对象是MCPTool自动创建的，它会把调用请求发送给 MCP 服务器。具体来说，它会构造一个 JSON-RPC 格式的消息，通过 stdin 发送给服务器进程：
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "amap_maps_text_search",
"arguments": {
"keywords": "景点",
"city": "北京"
}
}
}
MCP 服务器接收到这个消息，解析参数，然后调用高德地图的 HTTP API。它会构造 HTTP 请求，添加 API 密钥，发送请求，接收响应。
高德地图 API 返回 JSON 格式的数据，包含景点列表、地址、坐标等信息。MCP 服务器解析这些数据，提取关键字段，然后构造响应消息，通过 stdout 返回给MCPTool：
{
"jsonrpc": "2.0",
"result": {
"content": [
{
"type": "text",
"text": "找到以下景点：\n1. 故宫博物院 - 地址：东城区景山前街4号\n2. 天坛公园 - 地址：东城区天坛路\n..."
}
]
}
}
MCPTool接收到响应，提取文本内容，返回给 Agent。Agent 把这个结果作为工具调用的输出，继续生成最终的回复。
这个流程看起来很复杂，但对于 Agent 来说，它只需要知道有一个叫amap_maps_text_search的工具，可以搜索景点。所有的底层细节都被 MCP 协议和MCPTool封装起来了。
13.4.3 共享 MCP 实例
在我们的多 Agent 系统中，有三个 Agent 都需要使用高德地图的工具。那么每个 Agent 应该创建自己的MCPTool实例，还是共享同一个实例？
如果每个 Agent 都创建一个MCPTool实例，这意味着会有三个服务器进程同时运行。每个进程都会独立地调用高德地图 API，这可能会超过 API 的速率限制。而且多个进程会占用更多的内存和 CPU 资源。
更好的做法是让所有 Agent 共享同一个MCPTool实例。这样只需要启动一个 MCP 服务器进程，所有的 API 调用都通过这个进程进行。这不仅节省资源，还可以更好地控制 API 调用频率。
在代码中，我们在TripPlannerAgent的构造函数中创建一个MCPTool实例，然后把它添加到每个子 Agent 的工具列表中：
class TripPlannerAgent:
def __init__(self):
settings = get_settings()
self.llm = HelloAgentsLLM()
# 创建共享的MCP工具实例(只创建一次)
self.mcp_tool = MCPTool(
name="amap_mcp",
command="npx",
args=["-y", "@sugarforever/amap-mcp-server"],
env={"AMAP_API_KEY": settings.amap_api_key},
auto_expand=True
)
# 创建多个Agent,共享同一个MCP工具
self.attraction_agent = SimpleAgent(
name="AttractionSearchAgent",
llm=self.llm,
system_prompt=ATTRACTION_AGENT_PROMPT
)
self.attraction_agent.add_tool(self.mcp_tool) # 共享
self.weather_agent = SimpleAgent(
name="WeatherQueryAgent",
llm=self.llm,
system_prompt=WEATHER_AGENT_PROMPT
)
self.weather_agent.add_tool(self.mcp_tool) # 共享
self.hotel_agent = SimpleAgent(
name="HotelAgent",
llm=self.llm,
system_prompt=HOTEL_AGENT_PROMPT
)
self.hotel_agent.add_tool(self.mcp_tool) # 共享
这样，三个 Agent 都可以使用高德地图的 16 个工具，但底层只有一个 MCP 服务器进程在运行。当我们调用TripPlannerAgent的plan_trip方法时，三个 Agent 会依次调用工具，所有的请求都通过同一个 MCP 服务器发送到高德地图 API。
13.4.4 Unsplash 图片 API 集成
除了高德地图，我们还需要为景点获取图片，让旅行计划更加生动直观。我们使用 Unsplash API 来搜索景点图片。需要注意的是，Unsplash 是国外的服务，而且是为数不多可以免费使用的图片 API，所以搜索结果可能不够准确。在实际项目中，可以考虑使用必应、百度或高德的 POI 图片 API，但这些服务通常需要付费。
Unsplash API 的集成比较简单，我们创建一个UnsplashService类来封装 API 调用：
# backend/app/services/unsplash_service.py
import requests
from typing import Optional, List, Dict
import logging
logger = logging.getLogger(__name__)
class UnsplashService:
"""Unsplash图片服务"""
def __init__(self, access_key: str):
self.access_key = access_key
self.base_url = "https://api.unsplash.com"
def search_photos(self, query: str, per_page: int = 10) -> List[Dict]:
"""搜索图片"""
try:
url = f"{self.base_url}/search/photos"
params = {
"query": query,
"per_page": per_page,
"client_id": self.access_key
}
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
results = data.get("results", [])
# 提取图片URL
photos = []
for result in results:
photos.append({
"url": result["urls"]["regular"],
"description": result.get("description", ""),
"photographer": result["user"]["name"]
})
return photos
except Exception as e:
logger.error(f"搜索图片失败: {e}")
return []
def get_photo_url(self, query: str) -> Optional[str]:
"""获取单张图片URL"""
photos = self.search_photos(query, per_page=1)
return photos[0].get("url") if photos else None
这个服务类提供了两个方法：search_photos搜索多张图片，get_photo_url获取单张图片的 URL。我们在 API 路由中使用这个服务，为每个景点获取图片：
# backend/app/api/routes/trip.py
from app.services.unsplash_service import UnsplashService
unsplash_service = UnsplashService(settings.unsplash_access_key)
@router.post("/plan", response_model=TripPlan)
async def create_trip_plan(request: TripPlanRequest) -> TripPlan:
# 生成旅行计划
trip_plan = trip_planner_agent.plan_trip(request)
# 为每个景点获取图片
for day in trip_plan.days:
for attraction in day.attractions:
if not attraction.image_url:
image_url = unsplash_service.get_photo_url(
f"{attraction.name} {trip_plan.city}"
)
attraction.image_url = image_url
return trip_plan
注意我们没有把 Unsplash 封装成 Tool 或 MCP 工具，而是直接在 API 路由中调用。这是因为图片搜索不需要 Agent 的智能决策，只是一个简单的数据增强步骤。如果你想让 Agent 能够自主决定是否需要图片，或者选择不同的图片来源，可以考虑把它封装成 Tool。