title: “Agent 工具访问 (MCP)” weight: 600
在本模块中,我们将构建一个包含订单和库存工具的 MCP (Model Context Protocol) server,将其部署在 EKS 上,并将 agent 连接到它。告别硬编码的 mock。

MCP 是一个开放协议,用于定义 agent 如何发现和调用工具。agent 不再将工具硬编码在其内部,而是连接到一个对外提供这些工具的 server。工具逻辑与 agent 逻辑解耦。我们可以在多个 agent 之间复用工具,并且无需重新部署即可更新工具。
cd ~/environment/modules/20-self-managed/600-agent-tools-mcp/mcp-server
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("AnyCompany Tools")
@mcp.tool()
def lookup_order(order_id: str) -> dict:
"""Look up order status, tracking, and details by order ID."""
...
@mcp.tool()
def check_inventory(product_name: str) -> dict:
"""Check stock availability for a product."""
...
@mcp.tool() 装饰器的工作方式与 Strands Agents 中的 @tool 相同。我们编写一个普通的 Python 函数,FastMCP 会自动根据函数签名和 docstring 生成工具 schema。mcp.streamable_http_app() 返回一个 ASGI 应用,由 uvicorn 直接提供服务。mcp-server:v1 镜像已在预置过程中预先构建并推送到 ECR,因此可以直接部署:
cd ~/environment/modules/20-self-managed/600-agent-tools-mcp/mcp-server
envsubst < k8s.yaml | kubectl apply -f -
kubectl rollout status deployment/mcp-server --timeout=60s
cd ~/environment/modules/20-self-managed/600-agent-tools-mcp/customer-agent
tools.py 文件不再需要,因为 mock 数据现在位于 MCP server 中。启动时,agent 从 MCP server 检索可用的工具列表,并直接将其传递给 Agent(tools=...)。
mcp_client = MCPClient(lambda: streamablehttp_client(mcp_server_url))
mcp_client.__enter__()
mcp_tools = mcp_client.list_tools_sync()
print(f"Discovered {len(mcp_tools)} MCP tools: {[t.tool_name for t in mcp_tools]}")
agent = Agent(model=model, system_prompt=SYSTEM_PROMPT,
tools=[search_products, *mcp_tools])
list_tools_sync() 在启动时从 server 拉取工具 schema。在这一行运行之前,agent 并不知道这些工具的存在。customer-agent:mcp 镜像已在预置过程中预先构建并推送到 ECR:
cd ~/environment/modules/20-self-managed/600-agent-tools-mcp/customer-agent
envsubst < k8s.yaml | kubectl apply -f -
kubectl rollout status deployment/customer-agent --timeout=180s
打开聊天 UI,选择 Customer Agent (Self-managed GenAI),并尝试一个在一次对话中使用两个 MCP 工具的流程:
I want to return the headphones from order ORD-11111
agent 应该调用 lookup_order,看到状态为 processing,并拒绝发起退货,所有这些都可以在 Langfuse 中查看。

lookup_order、check_inventory 和 initiate_return一个 agent 完成所有工作。接下来,我们将把它拆分成通过 A2A 协调的多个专家。