SSE MCP And Langchain Client Example

](https://mseep.ai/app/sdcalvo-sse-mcp-and-langchain-client-example)

FastAPI MCP服务器+LangChain客户端示例

示例项目演示如何使用以下方法将FastAPI端点作为模型上下文协议（MCP）工具公开 fastapi-mcp。包括一个基本的LangChain代理(langchain_client.py)使用HTTP/SSE连接到本地FastAPI服务器 langchain-mcp-adapters 发现和使用暴露的工具。

先决条件

在开始之前，请确保已安装以下内容：

• python 建议使用3.10或更高版本。
• uv: 本项目中使用的Python包管理器。 (安装说明)
• Node.js和npm： 要求…… npx （用于运行可选的MCP检查器）。你可以从以下网址下载Node.js（包括npm） .
• Git： 用于克隆存储库。
• OpenAI API密钥： LangChain客户端示例需要。你需要将其设置为 .env 文件。

该项目演示了如何使用以下工具设置基本的FastAPI应用程序，并将其端点作为模型上下文协议（MCP）工具公开 fastapi-mcp 图书馆。它还包括一个连接到并使用这些工具的LangChain代理客户端，并涵盖了配置Cursor以进行连接。

入门/如何跑步

1. 克隆存储库：

   git clone 
   cd 

1. 安装 uv: 如果没有，请安装 uv 包管理器（参见 项目设置 下面是命令）。

1. 设置环境和安装依赖关系：

   uv init # If pyproject.toml doesn't exist
   uv venv # Create virtual environment (.venv)
   # Install all project dependencies
   uv pip install fastapi "uvicorn[standard]" fastapi-mcp langchain-mcp-adapters langgraph langchain-openai python-dotenv

1. 创建 .env 文件： 创建一个名为的文件 .env 在项目根目录中，并添加您的OpenAI API密钥：

   OPENAI_API_KEY=your_openai_api_key_here

1. 运行FastAPI MCP服务器： 打开终端并运行：

   uvicorn main:app --reload --port 8000

保持此终端运行。 _（或者，您可以使用 Python: FastAPI MCP 中定义的调试配置 .vscode/launch.json 在VS Code/Coursor中运行附加调试器的服务器。)_

1. 运行LangChain客户端： 打开A _第二_ 终端和运行：

   uv run python langchain_client.py

客户端将连接到服务器，发现工具，并使用代理运行查询。

1. （可选）带MCP检查器的测试服务器： 在运行LangChain客户端之前，或者为了进行更直接的测试，您可以使用官方的MCP Inspector工具：

- 确保FastAPI服务器正在运行（步骤5）。 - 打开另一个终端并运行： npx @modelcontextprotocol/inspector _(npx 附带Node.js/npm。如果此命令失败，请确保Node.js已安装并可在PATH中访问。)_ - 在检查器UI中，连接到您的服务器URL： http://127.0.0.1:8000/mcp - 导航到“工具”，单击“列出工具”查看 read_root__get 和 greet_user_greet__name__get. - 选择工具、填充参数（例如。， name 对于greet_user），然后单击“运行工具”。

1. （可选）测试 greet_user LangChain客户端： 这 greet_user 端点和相应的测试查询(query2)in langchain_client.py 目前处于活动状态。只需运行LangChain客户端（步骤6），观察其执行的第二部分，在那里它应该尝试问候用户“LangChain”。

- _（如果要禁用此测试，请注释掉 @app.get("/greet/{name}") 端点在 main.py 和那个 query2 部分在 langchain_client.py)_

目标

构建一个具有MCP功能的简单FastAPI服务器，用于学习和测试目的，可在本地运行，并可从MCP客户端（如Cursor编辑器的代理）连接。

项目设置

1. 包管理器： 我们使用 uv，一个用Rust编写的快速Python包安装程序和解析器。

- 安装（Windows PowerShell）： irm https://astral.sh/uv/install.ps1 | iex - 项目初始化： uv init （创建 pyproject.toml) - 虚拟环境： uv venv （创建和管理 .venv)

1. 依赖关系： 使用安装 uv:

   # Specific commands used during development (covered by the combined install in Getting Started):
   # uv pip install fastapi "uvicorn[standard]" fastapi-mcp
   # uv pip install langchain-mcp-adapters langgraph langchain-openai python-dotenv

这将安装FastAPI、Uvicorn ASGI服务器， fastapi-mcpLangChain组件，以及 python-dotenv 进入 .venv 虚拟环境。

应用程序(main.py)

使用端点创建了一个简单的FastAPI应用程序：

• /：返回欢迎消息。
• /greet/{name}：返回个性化问候语（当前已被注释掉）。

至关重要的是 fastapi-mcp 集成发生 之后 FastAPI路由定义：

from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI(...)

# --- Define FastAPI routes (@app.get, @app.post, etc.) ---
@app.get("/")
async def read_root():
    # ... endpoint logic ...

# --- Initialize and mount fastapi-mcp AFTER routes ---
mcp = FastApiMCP(app)
mcp.mount() # Exposes tools under /mcp by default

运行服务器

• 直接： uvicorn main:app --reload --port 8000
• 使用调试器（游标/VS代码）： A. launch.json 文件创建于 .vscode 运行带有调试器的应用程序。

朗链客户端(langchain_client.py)

• 阅读 .env 文件为 OPENAI_API_KEY.
• 用途 langchain-mcp-adapters (MultiServerMCPClient)连接到正在运行的FastAPI服务器 http://127.0.0.1:8000/mcp 通过SSE。
• 自动发现MCP服务器暴露的工具。
• 使用发现的工具和OpenAI LLM创建LangGraph ReAct代理。
• 通过代理运行示例查询。

光标集成（MCP客户端）

1. 配置： A. .cursor/mcp.json 文件是在项目根目录中创建的：

   {
     "mcpServers": {
       "local-fastapi-mcp": {
         "url": "http://127.0.0.1:8000/mcp"
       }
     }
   }

1. 激活： 需要在Cursor的设置中启用服务器（Ctrl+，搜索“Cursor settings”并转到“MCP”选项卡）。
2. 用途： 然后，可以要求Cursor代理（在聊天面板中）使用发现的工具。

主要学习内容

• uv 基础知识： uv init, uv venv，以及 uv pip install 提供了一种快速设置Python项目环境的方法。
• fastapi-mcp 初始化顺序： 这 FastApiMCP(app) 必须创建实例 mcp.mount() 必须调用 _之后_ FastAPI路由(@app.get等），您希望将其作为工具公开。否则，工具就不会被发现。
• 端口冲突： 后台流程（如 uvicorn)可以持有港口。在Windows上， `netstat -ano | findstr "

" 可以找到进程ID（PID），以及 taskkill /F /PID ` 可以终止它。有时需要短暂等待或重新启动IDE，操作系统才能完全释放端口。

• 光标MCP配置： 使用 url 输入 .cursor/mcp.json 将Cursor连接到基于HTTP的MCP服务器（如 fastapi-mcp).
• 代理工具调用： 当通用Cursor代理环境使用配置的MCP服务器时。， mcp_local-fastapi-mcp_read_root__get)而不是通用的工具调用函数。一旦建立连接并发现工具，明确要求聊天中的代理“使用该工具…”就可以了。

进一步探索和功能（来自fastapi-mcp文档）

本节总结了 fastapi-mcp 我们尚未实现但了解这些文档很有用。

身份验证和授权

fastapi-mcp 支持使用FastAPI依赖进行身份验证，还包括OAuth 2支持。

基本令牌传递：

• 最初不需要特殊的服务器配置。

• MCP客户端需要发送 Authorization 头球这通常可以通过一座桥来实现，比如 mcp-remote:

  {
    "mcpServers": {
      "remote-example": {
        "command": "npx",
        "args": [
          "mcp-remote",
          "http://localhost:8000/mcp",
          "--header",
          "Authorization:${AUTH_HEADER}"
        ]
      },
      "env": {
        "AUTH_HEADER": "Bearer "
      }
    }
  }

• 向 _需要_ 在MCP服务器端进行身份验证，向添加依赖项 AuthConfig:

  from fastapi import Depends
  from fastapi_mcp import FastApiMCP, AuthConfig

  # Assuming verify_auth is your FastAPI dependency that checks the token
  mcp = FastApiMCP(
      app,
      auth_config=AuthConfig(
          dependencies=[Depends(verify_auth)],
      ),
  )

OAuth 2流程：

• 完全支持OAuth 2（MCP规范2025-03-26）。
• 需要配置 AuthConfig 包含提供者详细信息（发行者、URL、客户端ID/秘密）。
• setup_proxies=True 通常需要处理标准OAuth提供者和MCP客户端期望之间的不兼容性（如缺少动态注册、范围处理）。

  mcp = FastApiMCP(
      app,
      auth_config=AuthConfig(
          issuer=f"https://auth.example.com/",
          authorize_url=f"https://auth.example.com/authorize",
          oauth_metadata_url=f"https://auth.example.com/.well-known/oauth-authorization-server",
          audience="my-audience",
          client_id="my-client-id",
          client_secret="my-client-secret",
          dependencies=[Depends(verify_auth)],
          setup_proxies=True, # Creates compatibility endpoints
      ),
  )

• 使用 mcp-remote 通常需要指定一个固定端口(mcp-remote ... 8080)因此回调URL(http://127.0.0.1:8080/oauth/callback)可以在OAuth提供程序中配置。

工具命名

• MCP工具名称来源于FastAPI路由 operation_id.
• 如果没有指定。， read_user_users__user_id__get).
• 它是 推荐 设置显式 operation_idFastAPI路由上的s，以获得更清晰的MCP工具名称：

  @app.get("/users/{user_id}", operation_id="get_user_info")
  async def read_user(user_id: int):
      # ...

刷新工具

• 如果添加了FastAPI路由 _之后_ mcp.mount() 如果被调用，则不会自动包含它们。

• 解决方案：致电 mcp.setup_server() 在定义新路线之后。

  app = FastAPI()
  mcp = FastApiMCP(app)
  mcp.mount()

  @app.get("/new/endpoint", operation_id="new_tool")
  async def new_endpoint(): ...

  # Refresh the tools
  mcp.setup_server()

测试

• 这 @modelcontextprotocol/inspector 该工具可用于测试MCP服务器：

  # Run the inspector
  npx @modelcontextprotocol/inspector
  # Connect to your server URL (e.g., http://127.0.0.1:8000/mcp)
  # Use the UI to list and run tools.

部署

• 您可以挂载从一个FastAPI应用程序创建的MCP服务器(api_app)到a _不同的_ FastAPI应用程序(mcp_app)对于单独部署：

  api_app = FastAPI()
  # ... define API endpoints ...

  mcp_app = FastAPI()
  mcp = FastApiMCP(api_app) # Create from api_app
  mcp.mount(mcp_app) # Mount onto mcp_app

  # Run separately:
  # uvicorn main:api_app --port 8001
  # uvicorn main:mcp_app --port 8000

定制

• 服务器名称/描述可以在 FastApiMCP 初始化：

  mcp = FastApiMCP(
      app,
      name="My Custom MCP Name",
      description="Description for the server."
  )

• 工具/模式描述可以定制（例如，包括所有可能的响应）：

  mcp = FastApiMCP(
      app,
      describe_all_responses=True,
      describe_full_response_schema=True
  )

• 可以使用以下方式过滤暴露的端点 include_operations, exclude_operations, include_tags, exclude_tags 在...期间 FastApiMCP 初始化。

致谢

该项目是在人工智能配对编程工具的帮助下开发的，包括谷歌的Gemini模型集成到Cursor和OpenAI的ChatGPT中，用于研究和文档检索。他们的贡献在整个过程中是无价的。