文档中心

AgentKit

MCP 接入

在 Agent 中集成 MCP

在 Agent 中集成 MCP

本文以查询天气为例介绍如何为 VeADK 构建的 Agent 集成 MCP 服务或 MCP 工具集。

注意

本文仅适用于入站身份认证为 API Key 类型的 MCP 服务或工具集。

前提条件

已使用 VeADK 构建 Agent。

已获取 MCP 服务的地址（URL），支持在 MCP 服务或工具集调用示例中获取，操作详情参见获取 MCP 服务调用示例或获取 MCP 工具集调用示例。

已获取 MCP 服务的入站 API Key，支持在 MCP 服务或工具集的详情页的“基本信息>入站身份认证”中获取。

使用 VeADK 对接 MCP

入站类型为 API Key 的 MCP 服务或工具集接入的代码示例如下。

集成 MCP 服务

配置环境变量。

# MCP服务的地址（URL），支持在 MCP服务示例中获取，操作详情参见获取 MCP 服务调用示例。
export MCP_SERVICE_URL=<mcp_service_url>
# MCP服务的入站API Key凭证
export MCP_SERVICE_API_KEY=<mcp_service_api_key>
# 平台的 AK和 SK。
export VOLCENGINE_ACCESS_KEY=<your_access_key>
export VOLCENGINE_SECRET_KEY=<your_secret_key>

安装 Python 依赖。

pip install veadk-python

集成示例。

from veadk import Agent, Runner
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
import asyncio
import os

mcp_service_url = os.getenv("MCP_SERVICE_URL")
mcp_service_api_key = os.getenv("MCP_SERVICE_API_KEY")

mcp_service_tool = MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=mcp_service_url, headers={"Authorization": f"Bearer {mcp_service_api_key}"}
),
)

agent = Agent(
# instruction是Agent的 Prompt。这里只是一个示例，具体场景下需要根据实际情况调整。
instruction="You are an intelligent assistant, skilled at using the mcp_service_tool to solve problems and replying to users' questions politely.",
tools=[mcp_service_tool],
)

runner = Runner(agent=agent)
user_prompt = "Hello, how many steps does it take to put an elephant into a refrigerator?"
session_id = "mcp_service_session_id_123"
response = asyncio.run(
runner.run(messages=user_prompt, session_id=session_id)
)

print(response)

集成 MCP 工具集

配置环境变量。

# 平台的 AK 和 SK。
export VOLCENGINE_ACCESS_KEY=<your_access_key>
export VOLCENGINE_SECRET_KEY=<your_secret_key>
# MCP工具集的地址（URL），支持在工具集调用示例中获取，操作详情参见获取 MCP 工具集调用示例。
export TOOL_MCP_ROUTER_URL=<tool_mcp_router_url>
# MCP工具集的入站 API Key 凭证
export TOOL_MCP_ROUTER_API_KEY=<tool_mcp_router_api_key>

安装 Python 依赖。

pip install veadk-python

集成示例。

from veadk import Agent, Runner
from veadk.tools.builtin_tools.mcp_router import mcp_router
import asyncio

agent = Agent(
# instruction是Agent的 Prompt。这里只是一个示例，具体场景下需要根据实际情况调整。
instruction="You are an intelligent assistant, skilled at using the mcp_service_tool to solve problems and replying to users' questions politely.",
tools=[mcp_router],
)

runner = Runner(agent=agent)
user_prompt = "Hello, how many steps does it take to put an elephant into a refrigerator?"
session_id = "mcp_toolset_session_id_123"
response = asyncio.run(
runner.run(messages=user_prompt, session_id=session_id)
)

print(response)

最近更新时间：2026.09.20 19:18:56

这个页面对您有帮助吗？

有用

有用

无用

无用