【大模型系列篇】大模型基建工程:基于 FastAPI 自动构建 SSE MCP 服务器

【大模型系列篇】大模型基建工程:基于 FastAPI 自动构建 SSE MCP 服务器

今天我们将使用FastAPI来构建 MCP 服务器,Anthropic 推出的这个MCP 协议,目的是让 AI 代理和你的应用程序之间的对话变得更顺畅、更清晰。FastAPI 基于 Starlette 和 Uvicorn,采用异步编程模型,可轻松处理高并发请求,尤其适合 MCP 场景下大模型与外部系统的实时交互需求,其性能接近 Node.js 和 Go,在数据库查询、文件操作等 I/O 密集型任务中表现卓越。

开始今天的正题前,我们来回顾下相关的知识内容:

高性能Python Web服务部署架构解析》、《使用Python开发MCP Server及Inspector工具调试》、《构建智能体MCP客户端:完成大模型与MCP服务端能力集成与最小闭环验证

 

FastAPI基础知识

安装依赖

pip install uvicorn, fastapi

FastAPI服务代码示例 

from fastapi import FastAPI app = FastAPI() @app.get("/") async def root(): return {"data": "Hello MCP!"}

uvicorn启动server 

uvicorn server:app --reload

接下来,我们将基于FastAPI来开发MCP服务器

 

FastAPI开发MCP Server

FastAPI-MCP 一个零配置工具,用于自动将FastAPI端点暴露为模型上下文协议(MCP)工具。其特点在于简洁性和高效性,以下是一些主要特点:

  • 直接集成:不需要复杂的设置,直接集成到FastAPI应用中。
  • 自动转换:无需手动编写代码,自动将FastAPI端点转换为MCP工具。
  • 灵活性:支持自定义MCP工具,与自动生成的工具一同使用。
  • 性能:基于Python 3.10+和FastAPI,保证了高性能的API服务。
  • 文档友好:保持了原有的API文档,方便开发者使用和理解。

安装依赖

pip install fastapi-mcp

MCP服务代码示例

from fastapi import FastAPI from fastapi_mcp import add_mcp_server from typing import Any import httpx # 常量 NWS_API_BASE = "https://api.weather.gov" USER_AGENT = "weather-app/1.0" app = FastAPI() mcp_server = add_mcp_server( app, # FastAPI 应用 mount_path="/mcp", # MCP 服务器挂载的位置 name="Weather MCP Server", # MCP 服务器的名字 describe_all_responses=True, # 默认是 False。就像打开一个百宝箱,把所有可能的响应模式都包含在工具描述里,而不只是成功的响应。 describe_full_response_schema=True # 默认是 False。把完整的 JSON 模式包含在工具描述里,而不只是一个对大语言模型友好的响应示例。 ) async def make_nws_request(url: str) -> dict[str, Any] | None: """向 NWS API 发起请求,并进行错误处理。""" headers = { "User-Agent": USER_AGENT, "Accept": "application/geo+json" } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() except Exception: return None @mcp_server.tool() async def get_forecast(latitude: float, longitude: float) -> str: """获取地点的天气预报。 参数: latitude: 地点的纬度 longitude: 地点的经度 """ points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}" points_data = await make_nws_request(points_url) if not points_data: return "Unable to fetch forecast data for this location." forecast_url = points_data["properties"]["forecast"] forecast_data = await make_nws_request(forecast_url) if not forecast_data: return "Unable to fetch detailed forecast." periods = forecast_data["properties"]["periods"] forecasts = [] for period in periods[:5]: forecast = f""" {period['name']}: Temperature: {period['temperature']}°{period['temperatureUnit']} Wind: {period['windSpeed']} {period['windDirection']} Forecast: {period['detailedForecast']} """ forecasts.append(forecast) return "\n---\n".join(forecasts)

启动 mcp server

uvicorn server:app --host 0.0.0.0 --port 8001 --reload

 启动 mcp inspector 调试

CLIENT_PORT=8081 SERVER_PORT=8082 npx -y @modelcontextprotocol/inspector

当集成了 MCP 的 FastAPI 应用运行起来后,可以用任何支持 SSE 的 MCP 客户端连接它。我们这里还是使用 mcp inspector 进行调试,通过 SSE 连接 Weather MCP 服务器。

SSE是一种单向通信的模式,所以它需要配合HTTP Post来实现客户端与服务端的双向通信。严格的说,这是一种HTTP Post(客户端->服务端) + HTTP SSE(服务端->客户端)的伪双工通信模式,区别于WebSocket双向通信

 

如果MCP客户端不支持SSE,可以使用mcp-proxy连接MCP服务器。本质上是本地通过stdio连接到mcp-proxy,再由mcp-proxy通过SSE连接到MCP Server上。

mcp-proxy 支持两种模式,stdio to SSE SSE to stdio

安装 mcp-proxy

uv tool install mcp-proxy

配置 claude_desktop_config.json 

{ "mcpServers": { "weather-api-mcp-proxy": { "command": "mcp-proxy", "args": ["http://127.0.0.1:8001/mcp"] } } }

FastAPI-MCP 目前还有很多功能不完善,我们将持续关注进展。在《大模型基建工程:基于 FastAPI 自动构建 SSE MCP 服务器 —— 进阶篇》中我们手搓了一个自动挂载的功能,并基于现有fastapi base_url 将 api 挂载至 mcp_server。

大模型基建工程总结

FastAPI 构建 MCP 服务器的核心价值在于:通过类型安全的异步接口,将企业现有能力快速转化为大模型可调用的标准化服务。这种架构既保留了 FastAPI 的高效开发体验,又通过 MCP 协议实现了与前沿 AI 技术的无缝对接,同时结合 Docker 和 Kubernetes 实现弹性伸缩部署,可以快速应对大模型调用量的突发增长,是构建下一代智能系统的理想选择。

Read more

【算法一周目】滑动窗口(2)

【算法一周目】滑动窗口(2)

目录 水果成篮 解题思路 代码实现 找到字符串中所有字母异位词 解题思路 代码实现 串联所有单词的子串 解题思路 代码实现 最小覆盖子串 解题思路 代码实现 水果成篮 题目链接:904. 水果成篮 题目描述: 你正在探访一家农场,农场从左到右种植了一排果树。这些树用一个整数数组 fruits 表示,其中 fruits[i] 是第 i 棵树上的水果种类。你想要尽可能多地收集水果,但是有一些规则: * 你有两个篮子,每个篮子只能装一种类型的水果,篮子的容量无限制。 * 你可以选择任意一棵树开始采摘,但必须从这棵树开始依次向右采摘每棵树上的水果。 * 一旦遇到某棵树上的水果不符合篮子中的水果种类,你必须停止采摘。 返回你能采摘的最多的水果数量。 解题思路 解法:滑动窗口+哈希 根据题目要求,所求问题其实就是找一段最多只含两个不同元素的最长子区间,我们使用滑动窗口+哈希解决。 有一点值得注意,fruits[i]

By Ne0inhk
数据结构:队列

数据结构:队列

前言  本篇文章将讲解队列的概念和结构,队列的实现等知识的相关内容,本章代码实现的知识,与单向链表相关,所以如果还没看过单向链表文章,可以看看: https://blog.ZEEKLOG.net/2401_86982201/article/details/154615762?fromshare=blogdetail&sharetype=blogdetail&sharerId=154615762&sharerefer=PC&sharesource=2401_86982201&sharefrom=from_link 一、队列概念与结构 概念 与栈的数据结构类似,队列:只允许在⼀端进⾏插⼊数据操作,在另⼀端进⾏删除数据操作的特殊线性表,队列具有先进先出FIFO(First In

By Ne0inhk
基础算法篇(5)(蓝桥杯常考点)—动态规划(C/C++)

基础算法篇(5)(蓝桥杯常考点)—动态规划(C/C++)

文章目录 * 动态规划 * 前言 * 线性dp * 路径类dp * 经典线性dp * 背包问题分类 * 01背包问题 * 完全背包问题 * 多重背包 * 分组背包问题 * 混合背包问题 * 多维费用的背包问题 * 区间dp 动态规划 前言 在竞赛中,如果遇到动态规划的题目,只要不是经典题型,那么大概率就是以压轴题的形式出现 用动态规划解决问题的步骤:(递推形式) 1.定义状态表示: 根据经验+需要的意义,赋予dp数组相应的含义 (主要还是看需要记什么) 2.推导状态转移方程: 在dp表中分析,当前格子如何通过其余格子推导出来的 3.初始化: 将显而易见的以及边界情况下的位置填上值,来让后续填表的结果是正确的 4.确定填表顺序: 根据状态转移方程,确定按照什么顺序来填表 5.找出最终结果: 在表中找出需要的最终结果 洛谷 P10250 [GESP样题 六级] 下楼梯 洛谷 P1216

By Ne0inhk
【强化学习】演员评论家Actor-Critic算法(万字长文、附代码)

【强化学习】演员评论家Actor-Critic算法(万字长文、附代码)

📢本篇文章是博主强化学习(RL)领域学习时,用于个人学习、研究或者欣赏使用,并基于博主对相关等领域的一些理解而记录的学习摘录和笔记,若有不当和侵权之处,指出后将会立即改正,还望谅解。文章分类在👉强化学习专栏:        【强化学习】- 【单智能体强化学习】(7)---《演员评论家Actor-Critic算法》 演员评论家Actor-Critic算法 目录 Actor-Critic算法理解 1. 角色设定 2. 两者如何协作 3. 学习的核心 4. 为什么叫Actor-Critic? 生活中例子: Actor-Critic算法的背景与来源 1. 强化学习的起源 2. 策略梯度方法的局限性 3. Actor-Critic的提出 4. 历史发展与应用 Actor-Critic算法流程的推导 1. 强化学习的优化目标 2. 策略梯度定理 3. Critic:值函数估计 4. Actor:策略优化 5.

By Ne0inhk