参考文献:

• langchain-ai/langchain-mcp-adapters
• modelcontextprotocol/python-sdk: The official Python SDK for Model Context Protocol servers and clients

技术选型

• Python

• Fastapi

• Langchain

• 安装依赖

pip install uvicorn fastapi langchain langchain-mcp-adapters langchain-openai

- 新建 my_mcp_client.py

导入依赖，创建FastAPI

import asyncio
import os
import sys
from typing import Any, Dict
import uvicorn
from fastapi import FastAPI, HTTPException
from langchain import hub
from langchain.agents import (
    AgentExecutor,
    create_openai_tools_agent,
)
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
app = FastAPI()

配置你的MCP服务

• 可以直接把cursor里的配置文件粘贴过来。
  这里我配了一个“howtocook”，解决“今晚吃啥”的世界级难题，感兴趣的朋友们可以去看看，项目地址：worryzyy/HowToCook-mcp: 基于Anduin2017 / HowToCook （程序员在家做饭指南）的mcp server

# MCP客户端配置
mcp_config = {
    "howtocook-mcp": {"command": "npx", "args": ["-y", "howtocook-mcp"]}
    # 可以添加更多MCP服务配置
}

核心代码：创建 Agent|挂载 MCP 工具

• 创建 Agent：agent = create_openai_tools_agent(model, tools, prompt)
• 挂载 MCP工具的库函数：MultiServerMCPClient
• 创建 LLM|记得填写你的apikey:model = ChatOpenAI(model="gpt-4o",api_key="your_api_key")
• Agent执行者：agent_executor = AgentExecutor()

async def get_mcp_agent(user_input: str):
    """
    创建并返回一个 MCP 代理实例 (使用 OpenAI Tools Agent)
    Returns:
        AgentExecutor: 配置好的 OpenAI Tools Agent 执行器实例
    """
    async with MultiServerMCPClient(mcp_config) as client:
        model = ChatOpenAI(model="gpt-4o",api_key="your_api_key")
        tools = client.get_tools() 
        print("--- 可用的工具列表 ---")
        for tool in tools:
            print(f"  名称: {tool.name}")
            print(f"  描述: {tool.description}")
            print(f"  参数 Schema: {tool.args_schema}")
            print("-" * 10)
        # 从 Langchain Hub 拉取适用于 OpenAI Tools Agent 的 Prompt
        prompt = hub.pull("hwchase17/openai-tools-agent")
        # 打印确认 Prompt 输入变量
        print(f"--- 加载的 Prompt 输入变量: {prompt.input_variables} ---")
        # 使用 create_openai_tools_agent 创建 Agent
        agent = create_openai_tools_agent(model, tools, prompt)
        # 创建 Agent Executor
        agent_executor = AgentExecutor(
            agent=agent,
            tools=tools,  # 工具列表
            verbose=True,  # 是否打印详细日志
        )
        print(f"--- 向 Agent Executor 发送的输入 --- \n{user_input}")
        # 调用 invoke 时需要匹配 Prompt 中的输入变量
        # OpenAI Tools Agent Prompt 需要 'input' 和 'chat_history' (作为 MessagesPlaceholder)
        response = await agent_executor.ainvoke(
            {
                "input": user_input,
                "chat_history": [],  # 提供一个空的 chat_history 列表 (或实际历史记录)
            }
        )
        print(f"--- Agent Executor 的最终响应 --- \n{response}")
        return response["output"]

挂载路由|调用

@app.post("/query")
async def query_mcp(
    user_input: str = "宝、我不知道今天中午吃啥，你帮我推荐一下呗，我们两个人，我不吃辣,不吃香菜",
) -> Dict[str, Any]:
    try:
        response_data = await get_mcp_agent(user_input)
        return {"response": response_data, "status": "success"}
    except Exception as e:
        error_message = f"Error processing MCP query with OpenAI Tools Agent: {e}"
        print(error_message)
        raise HTTPException(status_code=500, detail=error_message)

巨坑！ProactorServer --自定义服务器类

• -这里不用通常的 uvicorn

uvicorn.run("mcp_agent:app", host=host, port=port, reload=True)

• 而是用 ProactorServer，这里坑惨我半小时，详情见 巨坑！解决 Windows 上 FastAPI/Asyncio 子进程 NotImplementedError 问题File “C:\Python310\lib\asyncio\base_even-CSDN博客

                    
# --- 自定义服务器类 解决Windows上 FastAPI/Asyncio 子进程 `NotImplementedError`
class ProactorServer(uvicorn.Server):
   def run(self, sockets=None):
       # 在服务器运行前设置事件循环策略 (仅 Windows)
       if sys.platform == "win32":
           print("Setting ProactorEventLoopPolicy for Uvicorn server on Windows.")
           asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
       # 使用 asyncio.run 启动服务器的 serve 方法
       asyncio.run(self.serve(sockets=sockets))
                                                               	
if __name__ == "__main__":
   # --- 修改服务器启动方式 ---
   port = 8081
   host = "127.0.0.1"
   print(
       f"Starting MCP Agent server with ProactorServer (OpenAI Tools Agent) at http://{host}:{port}"
   )
   # 1. 创建 Uvicorn 配置，确保 reload=False
   config = uvicorn.Config(app="agent_mcp:app", host=host, port=port, reload=False)
   # 2. 实例化自定义服务器
   server = ProactorServer(config=config)
   # 3. 运行自定义服务器
   server.run()

启动服务

右键用Python运行，或者执行命令

Python my_mcp_client.py

测试接口

可以直接使用fastapi自带的swagger接口文档

• 点击链接 http://127.0.0.1:8080/docs
  

• 输入你的问题

  宝、我不知道今天中午吃啥，你帮我推荐一下呗，我们两个人，我不吃辣,不吃香菜，不吃辣条
  

• 点击Execute

• 我们可以看到控制台打印了信息：

  • 选择了MCP: mcp_howtocook_whatToEat
  • 正确输入了参数，调用MCP :Invoking: mcp_howtocook_whatToEatwith{‘peopleCount’: 2}``
    

• 最后的AI回复
  
  至此我们就完成了一个自己的 MCP 客户端了。
  再把这些代码丢给 AI，让它帮你写个前端页面，就大功告成了！

好了，现在你可以在简历上写上：精通 MCP 了 [doge]

关于作者

• CSDN 大三小白新手菜鸟咸鱼本科生长期更新强烈建议不要关注！

作者的其他文章

RAG调优|AI聊天|知识库问答

• 【面试官：】在你的项目中你是怎么实现混合检索的？-CSDN博客
• 你是一名平平无奇的大三生，你投递了简历和上线的项目链接，结果HR真打开链接看！结果还报错登不进去QAQ！【RAG知识库问答系统】新增模型混用提示和报错排查【用户反馈与优化-2025.04.28-CSDN博客
• 你知不知道像打字机一样的流式输出效果是怎么实现的？AI聊天项目实战经验:流式输出的前后端完整实现！图文解说与源码地址(LangcahinAI,RAG,fastapi,Vue,python,SSE)-CSDN博客
• 【豆包写的标题…】《震惊！重排序为啥是 RAG 调优杀手锏？大学生实战项目，0 基础也能白嫖学起来》（Langchain-CSDN博客
• 【Langchain】RAG 优化：提高语义完整性、向量相关性、召回率–从字符分割到语义分块 (SemanticChunker)-CSDN博客
• • 如何让你的RAG-Langchain项目持久化对话历史\保存到数据库中_rag保存成数据库-CSDN博客
• 【RAG】向量？知识库的底层原理：向量数据库の技术鉴赏 | HNSW(导航小世界)、LSH、K-means-CSDN博客

Agent

• 【MCP】哎只能在cursor中用MCP吗？NONONO!三分钟教你自己造一个MCP客户端！-CSDN博客

docker

• 【docker0基础教学】手把手教你将你的本地项目进行docker部署-CSDN博客
• 怎么把我的前后端项目通过docker部署到云服务器【手把手教学】-CSDN博客

Python

• 【Python】一直搞不懂迭代器是个啥。。-CSDN博客

前端

• 面试官：你会怎么做前端web性能优化？我：.%*&.!-CSDN博客
• CDN是啥？十分钟让Cloudflare免费为你的网站做CDN加速！-CSDN博客

nginx

• Nginx 反向代理，啥是“反向代理“啊，为啥叫“反向“代理？而不叫“正向”代理？-CSDN博客
• 关于nginx，负载均衡是什么？它能给我们的业务带来什么？怎么去配置它？-CSDN博客

好用插件

• 你想不想让你写的博客一键发布多平台？！