注意
Go to the end 下载完整示例代码。
模型上下文协议¶
本教程涵盖以下支持MCP(Model Context Protocol)的AgentScope功能:
支持 HTTP (可流式传输的 HTTP 和 SSE) 和 StdIO MCP 服务器
提供有状态和无状态的MCP客户端
提供服务器级别和函数级别的MCP工具管理
此处有状态/无状态的区别指的是客户端是否与MCP服务器保持持久会话。 下表总结了支持的MCP客户端类型和协议:
客户端类型 |
HTTP (可流式HTTP和SSE) |
标准输入输出 |
|---|---|---|
有状态客户端 |
|
|
无状态客户端 |
|
import asyncio
import json
import os
from agentscope.mcp import HttpStatefulClient, HttpStatelessClient
from agentscope.tool import Toolkit
MCP 客户端¶
在AgentScope中,MCP客户端负责
连接到 MCP 服务器,
从服务器获取工具函数,以及
调用 MCP 服务器中的工具函数。
AgentScope中有两种类型的MCP客户端:有状态(Stateful)和无状态(Stateless)。 它们的区别仅在于管理与MCP服务器的会话方式。
有状态客户端: 状态化MCP客户端在其生存期内保持与MCP服务器的持久会话。开发者应显式调用
connect()和close()方法来管理连接生命周期。无状态客户端:无状态 MCP 客户端在调用工具函数时创建新会话,并在工具函数调用后立即销毁会话,这种方式更加轻量级。
注意
StdIO MCP 服务器仅支持有状态客户端,当调用
connect()时,它会在本地启动 MCP 服务器并连接到该服务器。对于有状态的客户端,开发者必须确保客户端在调用工具函数时处于连接状态。
当多个 HttpStatefulClients 或 StdIOStatefulClients 连接时, 应当采用后进先出(LIFO)的顺序关闭,以防止出现错误。
以高德地图MCP服务器为例,有状态和无状态客户端的创建方式非常相似:
stateful_client = HttpStatefulClient(
# The name to identify the MCP
name="mcp_services_stateful",
transport="streamable_http",
url=f"https://mcp.amap.com/mcp?key={os.environ['GAODE_API_KEY']}",
)
stateless_client = HttpStatelessClient(
# The name to identify the MCP
name="mcp_services_stateless",
transport="streamable_http",
url=f"https://mcp.amap.com/mcp?key={os.environ['GAODE_API_KEY']}",
)
有状态和无状态客户端均提供以下方法:
方法 |
描述 |
|---|---|
|
列出MCP服务器中所有可用的工具。 |
|
通过名称从MCP服务器获取可调用函数对象。 |
MCP 作为工具¶
AgentScope 提供了对 MCP(Managed Code Platform)工具的细粒度管理,包括服务器级别和函数级别的管理。
服务器级管理¶
你可以将所有工具从一个MCP服务器注册到 Toolkit 中,具体方式如下所示。
提示
可选地,您可以指定一个分组名称来组织这些工具。请参阅工具章节以了解分组工具管理的内容。
toolkit = Toolkit()
async def example_register_stateless_mcp() -> None:
"""Example of registering MCP tools from a stateless client."""
# Register all tools from the MCP server
await toolkit.register_mcp_client(
stateless_client,
# group_name="map_services", # Optional group name
)
print(
"Total number of MCP tools registered:",
len(toolkit.get_json_schemas()),
)
maps_geo = next(
tool
for tool in toolkit.get_json_schemas()
if tool["function"]["name"] == "maps_geo"
)
print("\nThe example ``maps_geo`` function:")
print(
json.dumps(
maps_geo,
indent=4,
ensure_ascii=False,
),
)
asyncio.run(example_register_stateless_mcp())
Total number of MCP tools registered: 15
The example ``maps_geo`` function:
{
"type": "function",
"function": {
"name": "maps_geo",
"description": "将详细的结构化地址转换为经纬度坐标。支持对地标性名胜景区、建筑物名称解析为经纬度坐标",
"parameters": {
"type": "object",
"properties": {
"address": {
"type": "string",
"description": "待解析的结构化地址信息"
},
"city": {
"type": "string",
"description": "指定查询的城市"
}
},
"required": [
"address"
]
}
}
}
要移除注册的工具,您可以使用remove_tool_function来移除特定工具函数,或者使用remove_mcp_clients来移除来自特定 MCP 的所有工具。
async def example_remove_mcp_tools() -> None:
"""Example of removing MCP tools."""
print(
"Total number of tools before removal: ",
len(toolkit.get_json_schemas()),
)
# Remove a specific tool function by its name
toolkit.remove_tool_function("maps_geo")
print("Number of tools: ", len(toolkit.get_json_schemas()))
# Remove all tools from the MCP client by its name
await toolkit.remove_mcp_clients(client_names=["mcp_services_stateless"])
print("Number of tools: ", len(toolkit.get_json_schemas()))
asyncio.run(example_remove_mcp_tools())
Total number of tools before removal: 15
Number of tools: 14
Number of tools: 0
函数级管理¶
我们注意到对于更细粒度地控制MCP工具的需求,例如对工具结果进行后处理,或者使用它们来创建更复杂的工具功能。
因此,AgentScope支持通过名称从MCP获取可调用函数对象,这样您就可以
直接调用它,
将其封装到您自己的函数中,或按您喜欢的任何方式处理。
此外,你可以指定是否将工具结果包装到AgentScope中的ToolResponse对象,以便能够无缝地使用Toolkit。
如果你设置wrap_tool_result=False,将会返回原始结果类型mcp.types.CallToolResult。
以maps_geo函数为例,你可以如下方式获取它作为一个可调用函数对象:
async def example_function_level_usage() -> None:
"""Example of using function-level MCP tool."""
func_obj = await stateless_client.get_callable_function(
func_name="maps_geo",
# Whether to wrap the tool result into ToolResponse in AgentScope
wrap_tool_result=True,
)
# You can obtain its name, description and json schema
print("Function name:", func_obj.name)
print("Function description:", func_obj.description)
print(
"Function JSON schema:",
json.dumps(func_obj.json_schema, indent=4, ensure_ascii=False),
)
# Call the function object directly
res = await func_obj(
address="Tiananmen Square",
city="Beijing",
)
print("\nFunction call result:")
print(res)
asyncio.run(example_function_level_usage())
Function name: maps_geo
Function description: 将详细的结构化地址转换为经纬度坐标。支持对地标性名胜景区、建筑物名称解析为经纬度坐标
Function JSON schema: {
"type": "function",
"function": {
"name": "maps_geo",
"description": "将详细的结构化地址转换为经纬度坐标。支持对地标性名胜景区、建筑物名称解析为经纬度坐标",
"parameters": {
"type": "object",
"properties": {
"address": {
"type": "string",
"description": "待解析的结构化地址信息"
},
"city": {
"type": "string",
"description": "指定查询的城市"
}
},
"required": [
"address"
]
}
}
}
Function call result:
ToolResponse(content=[{'type': 'text', 'text': '{"results":[{"country":"中国","province":"北京市","city":"北京市","citycode":"010","district":"东城区","street":[],"number":[],"adcode":"110101","location":"116.397455,39.909187","level":"兴趣点"}]}'}], metadata=None, stream=False, is_last=True, is_interrupted=False, id='2025-09-08 07:54:48.907_a83cb4')
扩展阅读¶
更多详细信息,请查看:
脚本总运行时间: (0 分 7.910 秒)