agentscope以STUDIO方式调用MCP服务
文章目录
- 解决方法：使用与 STDIO 传输匹配的客户端
- 关键修改说明
- 运行方式
从错误日志和 MCP 服务器启动信息来看，核心问题是 传输方式不匹配：
你的 MCP 服务器启动时使用的是 STDIO 传输方式（日志中显示 📦 Transport: STDIO），但你的客户端代码中却配置了 HttpStatelessClient 并使用 streamable_http 传输方式，两者不兼容，导致连接失败（httpx.ConnectError: All connection attempts failed）。
解决方法：使用与 STDIO 传输匹配的客户端
FastMCP 服务器默认使用 STDIO 传输时，AgentScope 需要对应使用 StdIOStatefulClient 客户端（而非 HTTP 客户端）。修改代码如下：
mcp_stdio_integration.py
import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit
# 关键修改：导入 STDIO 客户端（而非 HTTP 客户端）
from agentscope.mcp import StdIOStatefulClientasync def integrate_local_mcp():# 1. 创建 STDIO 类型的 MCP 客户端（匹配服务器的 STDIO 传输）# 注意：StdIO 客户端需要指定启动 MCP 服务器的命令mcp_client = StdIOStatefulClient(name="local_echo_mcp",# 命令：启动你的 MCP 服务器（替换为实际路径）command="python",args=["/mnt/data/test1/repo/agentscope/server.py"],# 工作目录（你的 server.py 所在目录）cwd="/mnt/data/test1/repo/agentscope/")# 2. 连接到 MCP 服务器（STDIO 客户端需要显式连接）await mcp_client.connect()# 3. 创建工具包并注册 MCP 服务toolkit = Toolkit()await toolkit.register_mcp_client(mcp_client)# 验证已注册的工具print("✅ 从 MCP 服务注册的工具列表：")for tool in toolkit.get_json_schemas():print(f" - 工具名：{tool['function']['name']}")# 4. 实例化智能体（其余配置不变）agent = ReActAgent(name="MCP_Studio_Agent",sys_prompt="你可以是一个提供帮助的机器人",#使用 echo_tool 重复文本，get_weather 查询天气。",model=DashScopeChatModel(model_name="qwen-max",api_key="这里填写API key 去https://bailian.console.aliyun.com/?tab=model#/api-key注册api key", # os.environ.get(stream=True,),formatter=DashScopeChatFormatter(),toolkit=toolkit,memory=InMemoryMemory(),enable_meta_tool=True,)# 5. 测试工具调用test_messages = [Msg("user", "重复：STDIO 连接成功", "user"),Msg("user", "查询南极天气", "user"),Msg("user", "25*25等于几", "user")]for msg in test_messages:print(f"\n📥 用户请求：{msg.content}")print("📤 智能体响应：")await agent(msg)# 6. 关闭客户端（重要：STDIO 客户端需显式关闭）await mcp_client.close()if __name__ == "__main__":asyncio.run(integrate_local_mcp())关键修改说明
-
客户端类型替换：
将HttpStatelessClient替换为StdIOStatefulClient，因为服务器使用STDIO传输。 -
客户端参数调整：
STDIO 客户端需要通过command和args指定启动 MCP 服务器的命令（相当于自动帮你运行python server.py），无需手动在另一个终端启动服务器。 -
连接与关闭：
- 必须调用
await mcp_client.connect()建立连接（STDIO 客户端需要显式连接）。 - 结束后必须调用
await mcp_client.close()释放资源（遵循 LIFO 原则）。
- 必须调用
运行方式
直接运行修改后的代码即可（无需手动启动 MCP 服务器，客户端会自动启动）：
python mcp_stdio_integration.py
此时客户端会自动启动 MCP 服务器并通过 STDIO 传输方式通信，解决连接失败问题。
MCP server.py 简单天气查询功能代码
"""
FastMCP Echo Server
"""from fastmcp import FastMCP# Create server
mcp = FastMCP("Echo Server")@mcp.tool
def echo_tool(text: str) -> str:return text@mcp.resource("echo://static")
def echo_resource() -> str:return "Echo!"@mcp.resource("echo://{text}")
def echo_template(text: str) -> str:"""Echo the input text"""return f"Echo: {text}"@mcp.prompt("echo")
def echo_prompt(text: str) -> str:return text# 新增的天气查询工具
@mcp.tool
def get_weather(city: str) -> str:"""Get the weather for a given city (demo version).Regardless of the city name, returns sunny and 25°C."""return f"城市 {city} 的天气：晴，气温 25 度。"if __name__ == "__main__":mcp.run()
确保在mcp_stdio_integration.py 的mcp_client = StdIOStatefulClient函数当中将自己的MCP基本信息填写好，路径填写正确，然后运行即可