返回市场
sse-mcp-and-langchain-客户端示例

sse-mcp-and-langchain-客户端示例

作者:SDCalvo2 星标更新:2025-05-20

项目介绍

MseeP.ai 安全评估徽章

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 Inspector)。你可以从 nodejs.org 下载 Node.js(包括 npm)。
  • Git: 用于克隆仓库。
  • OpenAI API 密钥: 用于 LangChain 客户端示例。你需要在 .env 文件中设置这个密钥。

该项目演示了如何设置一个基本的 FastAPI 应用程序,并使用 fastapi-mcp 库将其端点暴露为模型上下文协议(MCP)工具。它还包括一个 LangChain 客户端代理,该代理连接到并使用这些工具,并涵盖了配置 Cursor 连接的内容。

开始 / 如何运行

  1. 克隆仓库:

    git clone <your-repo-url>
    cd <repo-directory>
    
  2. 安装 uv: 如果没有安装,安装 uv 包管理器(参见下面的“项目设置”部分中的命令)。

  3. 设置环境及安装依赖:

    uv init # 如果 pyproject.toml 不存在
    uv venv # 创建虚拟环境 (.venv)
    # 安装所有项目依赖
    uv pip install fastapi "uvicorn[standard]" fastapi-mcp langchain-mcp-adapters langgraph langchain-openai python-dotenv
    
  4. 创建 .env 文件: 在项目根目录下创建一个名为 .env 的文件,并添加你的 OpenAI API 密钥:

    OPENAI_API_KEY=your_openai_api_key_here
    
  5. 运行 FastAPI MCP 服务器: 打开终端并运行:

    uvicorn main:app --reload --port 8000
    

    让这个终端保持运行状态。 (或者,你可以在 VS Code/Cursor 中使用 .vscode/launch.json 中定义的 Python: FastAPI MCP 调试配置来运行带有调试器的服务器。)

  6. 运行 LangChain 客户端: 打开另一个终端并运行:

    uv run python langchain_client.py
    

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

  7. (可选)使用 MCP Inspector 测试服务器: 在运行 LangChain 客户端之前,或进行更直接的测试时,可以使用官方的 MCP Inspector 工具:

    • 确保 FastAPI 服务器正在运行(步骤 5)。
    • 打开另一个终端并运行:npx @modelcontextprotocol/inspector (npx 随 Node.js/npm 一起提供。如果此命令失败,请确保 Node.js 已安装并且在你的 PATH 中可用。)
    • 在 Inspector UI 中,连接到你的服务器 URL:http://127.0.0.1:8000/mcp
    • 导航到“工具”,点击“列出工具”以查看 read_root__getgreet_user_greet__name__get
    • 选择一个工具,填写参数(例如,name 对于 greet_user),然后点击“运行工具”。
  8. (可选)使用 LangChain 客户端测试 greet_user: 当前 greet_user 端点及其对应的测试查询(query2)在 langchain_client.py 中是活动的。只需运行 LangChain 客户端(步骤 6),并观察其执行的第二部分,它应该尝试问候用户 'LangChain'。

    • (如果你想禁用此测试,请在 main.py 中注释掉 @app.get("/greet/{name}") 端点,并在 langchain_client.py 中注释掉 query2 部分)

目标

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

项目设置

  1. 包管理器: 我们使用了 uv,这是一个用 Rust 编写的快速 Python 包安装器和解析器。
    • 安装(Windows PowerShell):irm https://astral.sh/uv/install.ps1 | iex
    • 项目初始化:uv init(创建 pyproject.toml
    • 虚拟环境:uv venv(创建和管理 .venv
  2. 依赖项: 使用 uv 安装:
    # 开发期间使用的特定命令(由“开始”部分中的综合安装覆盖):
    # uv pip install fastapi "uvicorn[standard]" fastapi-mcp
    # uv pip install langchain-mcp-adapters langgraph langchain-openai python-dotenv
    
    这将 FastAPI、Uvicorn ASGI 服务器、fastapi-mcp、LangChain 组件和 python-dotenv 安装到 .venv 虚拟环境中。

应用程序 (main.py)

创建了一个简单的 FastAPI 应用程序,具有以下端点:

  • /: 返回欢迎消息。
  • /greet/{name}: 返回个性化的问候语(当前被注释掉了)。

关键的是,fastapi-mcp 集成发生在 FastAPI 路由定义之后:

from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI(...)

# --- 定义 FastAPI 路由 (@app.get, @app.post 等) ---
@app.get("/")
async def read_root():
    # ... 端点逻辑 ...

# --- 在路由之后初始化并挂载 fastapi-mcp ---
mcp = FastApiMCP(app)
mcp.mount() # 默认情况下,在 /mcp 下暴露工具

运行服务器

  • 直接运行: uvicorn main:app --reload --port 8000
  • 带调试器运行 (Cursor/VS Code):.vscode 中创建了一个 launch.json 文件,以便附带调试器运行应用程序。

LangChain 客户端 (langchain_client.py)

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

Cursor 集成 (MCP 客户端)

  1. 配置: 在项目根目录下创建了一个 .cursor/mcp.json 文件:
    {
      "mcpServers": {
        "local-fastapi-mcp": {
          "url": "http://127.0.0.1:8000/mcp"
        }
      }
    }
    
  2. 激活: 需要在 Cursor 设置中启用服务器(Ctrl+, 搜索“Cursor 设置”并转到“MCP”标签页)。
  3. 使用: Cursor 代理(聊天面板中)可以被要求使用发现的工具。

关键学习

  • uv 基础: uv init, uv venvuv pip install 提供了一种快速设置 Python 项目环境的方法。
  • fastapi-mcp 初始化顺序: FastApiMCP(app) 实例必须在定义要暴露为工具的 FastAPI 路由(如 @app.get 等)之后创建,并调用 mcp.mount()。否则,工具将不会被发现。
  • 端口冲突: 后台进程(如 uvicorn)可能会占用端口。在 Windows 上,netstat -ano | findstr "<PORT>" 可以找到进程 ID (PID),而 taskkill /F /PID <PID> 可以终止它。有时需要短暂等待或重启 IDE 以便操作系统完全释放端口。
  • Cursor MCP 配置: 使用 .cursor/mcp.json 文件中的 url 键将 Cursor 连接到基于 HTTP 的 MCP 服务器(如 fastapi-mcp 提供的)。
  • 代理工具调用: 虽然一般的 Cursor 代理环境使用配置的 MCP 服务器,但特定的配对编程 AI 助手使用自动生成的内部名称(如 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 <your-token>"
        }
      }
    }
    
  • 若要在 MCP 服务器端强制认证,向 AuthConfig 添加依赖项:

    from fastapi import Depends
    from fastapi_mcp import FastApiMCP, AuthConfig
    
    # 假设 verify_auth 是检查令牌的 FastAPI 依赖项
    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, # 创建兼容性端点
        ),
    )
    
  • 使用 mcp-remote 通常需要指定固定端口(mcp-remote ... 8080),以便回调 URL(如 http://127.0.0.1:8080/oauth/callback)可以在 OAuth 提供商中配置。

工具命名

  • MCP 工具名称源自 FastAPI 路由的 operation_id
  • 如果未指定,FastAPI 会自动生成一个(例如 read_user_users__user_id__get)。
  • 建议在 FastAPI 路由上显式设置 operation_id 以获得更清晰的 MCP 工具名称:
    @app.get("/users/{user_id}", operation_id="get_user_info")
    async def read_user(user_id: int):
        # ...
    

刷新工具

  • 如果在调用 mcp.mount() 之后添加了 FastAPI 路由,它们将不会自动包含。

  • 解决方案:在定义新路由后再次调用 mcp.setup_server()

    app = FastAPI()
    mcp = FastApiMCP(app)
    mcp.mount()
    
    @app.get("/new/endpoint", operation_id="new_tool")
    async def new_endpoint(): ...
    
    # 刷新工具
    mcp.setup_server()
    

测试

  • 可以使用 @modelcontextprotocol/inspector 工具测试 MCP 服务器:
    # 运行 inspector
    npx @modelcontextprotocol/inspector
    # 连接到你的服务器 URL(例如 http://127.0.0.1:8000/mcp)
    # 使用 UI 列出并运行工具。
    

部署

  • 可以将从一个 FastAPI 应用程序 (api_app) 创建的 MCP 服务器挂载到另一个 FastAPI 应用程序 (mcp_app) 上进行独立部署:

    api_app = FastAPI()
    # ... 定义 API 端点 ...
    
    mcp_app = FastAPI()
    mcp = FastApiMCP(api_app) # 从 api_app 创建
    mcp.mount(mcp_app) # 挂载到 mcp_app
    
    # 分别运行:
    # uvicorn main:api_app --port 8001
    # uvicorn main:mcp_app --port 8000
    

自定义

  • 服务器名称/描述可以在 FastApiMCP 初始化时设置:
    mcp = FastApiMCP(
        app,
        name="我的自定义 MCP 名称",
        description="服务器的描述。"
    )
    
  • 工具/模式描述可以自定义(例如,包括所有可能的响应):
    mcp = FastApiMCP(
        app,
        describe_all_responses=True,
        describe_full_response_schema=True
    )
    
  • 可以使用 include_operations, exclude_operations, include_tags, exclude_tagsFastApiMCP 初始化时过滤暴露的端点。

致谢

本项目在开发过程中得到了 AI 配对编程工具的帮助,包括集成到 Cursor 中的 Google Gemini 模型和用于研究和文档检索的 OpenAI ChatGPT。他们的贡献在整个过程中都是不可或缺的。