前文展示的是 AI 智能体调用 stdio 的 MCP Server，这里展示调用 Streamable HTTP的 MCP Server。
一、在 test 目录中添加 agent_http.py 文件
并添加如下代码：
import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack
import json
from fastmcp import Client# 加载 .env 文件
load_dotenv()# 使用 config 配置方式
config = {"mcpServers": {"server": {"url": "http://127.0.0.1:3002/mcp_atlas", # 服务器地址，需要根据 MCP Server 配置修改 "transport": "streamable-http"}}
}class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.api_key = os.getenv("API_KEY") # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL") # 读取 BASE URLself.model = os.getenv("MODEL") # 读取 modelif not self.api_key:raise ValueError("未找到 API KEY. 请在 .env 文件中配置 API_KEY")self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)async def process_query(self, query: str) -> str:"""调用大模型处理用户查询并根据返回的 tools 列表调用对应工具"""messages = [{"role": "user", "content": query}]# 获取工具列表available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema}} for tool in self.tools]print(f"debug info === MCP server 返回的工具列表 available_tools：{available_tools}")print(f"debug info === 拼接用户输入后的 messages：{messages}")# 请求 OpenAI 模型处理response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)# 处理返回的内容content = response.choices[0]print(f"debug info === AI 返回的 content： {content}")if content.finish_reason == "tool_calls":# 执行工具调用tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 执行工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")# 将模型返回的原始消息和工具执行的结果都添加到 messages 中messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id,})# 将上面的结果再返回给大模型生产最终的结果response = self.client.chat.completions.create(model=self.model,messages=messages,)return response.choices[0].message.contentreturn content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("MCP 客户端已启动！输入 'exit' 退出")while True:try:query = input("问: ").strip()if query.lower() == 'exit':breakresponse = await self.process_query(query)print(f"AI回复: {response}")except Exception as e:print(f"发生错误: {str(e)}")async def clean(self):"""清理资源"""await self.exit_stack.aclose()async def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器"""self.session = await self.exit_stack.enter_async_context(Client(config))async def list_tools(self):"""列出所有工具"""# 列出 MCP 服务器上的工具response = await self.session.list_tools()self.tools = responseasync def main():# 启动并初始化 MCP 客户端client = MCPClient()try:# 连接到 MCP 服务器await client.connect_to_server('server.py')# 列出 MCP 服务器上的工具await client.list_tools()# 运行交互式聊天循环，处理用户对话await client.chat_loop()finally:# 清理资源await client.clean()if __name__ == "__main__":asyncio.run(main())
注意：同样要修改：
- tool/math_tools.py
- 添加 .env 文件
修改具体内容参见： MCP - AI智能体调用 MCP Serverr - Stdio（六）
二、效果展示
2.1 启动 server
运行如下命令启动 MCP Sever
python server.py
2.2 启动 agent
运行如下命令启动 agent
python agent_http.py
运行后如下图：
源代码： https://pan.baidu.com/s/1DlzeeQuZ4dHhF3CfQKYQng?pwd=gbcr 提取码: gbcr