从零开始：把FastAPI接口变成LLM可调用的工具（FastAPI-MCP）

敲代码运筹帷幄的独夜

1079人浏览 · 2025-12-01 15:50:19

敲代码运筹帷幄的独夜 · 2025-12-01 15:50:19 发布

观前说明：本文全文公开，免费阅读，谢绝商用。无需关注即可阅读全文，也不会设置任何VIP或付费门槛。
相关链接：https://github.com/tadata-org/fastapi_mcp
本文是面向不了解MCP的开发者，目标是快速构建一个使用MCP的简单项目。完整demo代码会在文章末尾给出。

一、整体流程：从FastAPI到「LLM可调用工具」

基础技术简介

MCP（Model Context Protocol）是由Anthropic推出的 “模型与外部工具之间的标准调用协议”。
它解决了过去LLM无法直接“理解”一个API的结构、参数、类型、返回值的问题。
MCP规范包括：
工具发现（Tools Discovery）
模型能够询问MCP Server：“有哪些工具可用？它们分别叫什么？参数结构如何？”
返回：

工具名、输入JSON Schema、输出JSON Schema

工具调用（Tools Call）
通过统一的JSON-RPC（一种轻量级的远程过程调用——RPC协议，使用JSON格式编码请求和响应）请求调用具体工具。
双向流式交互（Streamable HTTP）
工具可在执行中逐步返回数据（如进度、流式结果）。
Schema统一化输入/输出参数使用JSON Schema描述，使得LLM能理解哪些字段必须提供、类型是什么、限制是什么。
MCP的核心目标：
让LLM能够像调用本地函数一样调用外部系统，而不会受到SDK、框架、语言生态的限制，所有工具都被抽象为统一的“可调用单元”。
MCP-Proxy
一个CLI代理，用来让本地LLM客户端（如 5ire）可以支持MCP Server。
职责：

对上层：使用stdin/stdout与5ire通信（JSON-RPC）
对下层：使用Streamable HTTP与FastAPI-MCP通信

在两种协议之间进行完全透明的双向翻译。
Streamable HTTP
MCP引入的新型流式传输规范，特性包括：
标准化：基于 HTTP 1.1/2.0，兼容性强
全双工：支持双方在同一连接内同时发送消息
事件化：工具可随时发送事件（结果、进度、日志等）
比SSE更通用：SSE只允许服务端→客户端单向流，而Streamable HTTP支持双向。
FastAPI-MCP默认使用它，因此MCP客户端必须使用支持它的代理。

第 1 步：用FastAPI写一个普通后端接口

1.定义请求模型、响应模型（比如Pyd）
2.写好路由和业务逻辑
3.用Uvicorn启动服务
在这一层，它就是一个再普通不过的REST API，完全不涉及MCP。
例如：

@app.post("/rag/query", response_model=RagAnswer,
operation_id="rag_query_knowledge_base"  # operation_id会成为LLM能调用的工具名（MCP工具名）
)
async def rag_query(payload: RagQuery):
    ...

第 2 步：用FastApi-MCP把接口「升级」成MCP工具

核心在于：

from fastapi_mcp import FastApiMCP
from fastapi import FastAPI

app = FastAPI(...)
mcp = FastApiMCP(app,...)  # 默认路径为/mcp 
mcp.mount_http()

这一层做了几件事：
1.扫描FastAPI路由、生成工具描述
使用每个路由的operation_id作为MCP工具名，此时整个FastAPI已经成为了一个“工具”（LLM视角下的）。请求模型、响应模型的JSON Schema被完整保留，用来告诉LLM工具需要什么参数、返回什么结构。
2.挂载MCP服务端点
mcp.mount_http()会在FastAPI里挂出/mcp路径：

提供MCP工具发现（tools/list）
提供MCP工具调用（tools/call）
使用Streamable HTTP规范

到这一步，FastAPI应用就既是一个普通REST服务，又是一个MCP Server，可以被任何支持MCP的客户端发现和调用。

第 3 步：准备 LLM 客户端

作者使用的是5ire
1.配置 LLM（需要API Key）
这部分就是普通的“接模型”，跟MCP无关。
2.添加一个本地 MCP工具（比如叫rag-mcp）
在工具 → 本地 → “+”
编辑直到Config Preview变成：

{
  "key": "rag-mcp",
  "command": "mcp-proxy",
  "args": [
    "http://127.0.0.1:5000/mcp",
    "--transport=streamablehttp"  # 这是HTTP，不是其它
  ],
  "approvalPolicy": "always"  # 审批策略，如果是always，就是LLM每次调用MCP工具都要手动允许
}

意思是：
当5ire需要启用这个工具时，会在本地监听调用一个进程：

mcp-proxy http://127.0.0.1:5000/mcp

对5ire来说，这个进程就是本地MCP Server，用于通信。
此时，5ire将这个MCP Server打包给LLM，模型只需要知道：

有一个叫rag_query_knowledge_base的工具，参数是{query, top_k}

第 4 步：mcp-proxy扮演「协议翻译器」

mcp-proxy在调用链中承担核心作用：

上行方向（5ire → FastAPI）
5ire使用JSON-RPC发送调用请求
mcp-proxy将JSON-RPC转换为Streamable HTTP
请求被发送到FastAPI的/mcp
FastApi-MCP将工具调用映射为FastAPI路由

下行方向（FastAPI → 5ire）
FastAPI返回执行结果（可流式返回）
mcp-proxy将HTTP流转换回JSON-RPC
5ire接收工具的返回并交给模型
模型整合并生成最终回答

调用链示意（来回转译）：
5ire发tools/list→mcp-proxy转发到/mcp→FastAPI返回工具列表→再转回JSON-RPC给5ire；
5ire发tools/call→mcp-proxy转发到/mcp→调用rag_query→返回结果再传给5ire。

第 5 步：LLM 如何决定「要不要用这个工具？」

当在5ire里发出问题，5ire会把两类信息发给LLM：

对话内容（自然语言问题）、当前可用的MCP工具列表

如果模型决定调用工具，会向5ire发送一个工具调用计划，收到后弹出对话框问是否允许AI使用rag-mcp工具，如下图：

点允许后，它才正式把tools/call请求发给本地MCP（也就是 mcp-proxy）。到这一步也就是MCP应用成功了。

（扩展）第 6 步：完整调用链

发送问题：
使用rag-mcp工具查询：“什么是向量数据库”
5ire → LLM：
把问题 + 工具列表发给LLM。
LLM决定调用工具：
生成一个rag_query_knowledge_base的调用 + 参数。
5ire弹窗：
询问是否允许 AI 使用 rag-mcp工具。
点允许后：
5ire向本地MCP（mcp-proxy）发送tools/call请求。
mcp-proxy → FastAPI：
把这个调用转成对/mcp的请求，FastAPI-MCP识别到要调用rag_query这个FastAPI路由。
后端函数执行：
rag_query返回一个RagAnswer。
结果回传：
经FastAPI-MCP→ mcp-proxy → 5ire → 再交给 LLM作为工具返回。
LLM生成最终回答：
会综合工具结果 + 自己的推理生成看到的那段自然语言回答。

从LLM到RAG，再到MCP，技术的演进正一步步打通AI与真实系统的协作边界。

二、5ire配置的常见问题排查

1.API地址错误（Endpoint）

出现下图：

原因：现在配置的baseUrl指向了一个需要浏览器访问、过Cloudflare检查的网页（访问地址被Cloudflare拦住了），而不是一个可以直接被API/客户端调用的纯JSON接口。
解决：检查API地址是否正确。

2.模型不存在（not found, verify your APl base）

原因：请求的目标模型不存在。
解决：在对话窗口更换一个对话的模型。

3.SSE传输和HTTP冲突或后端服务未启动（MCP error-32000:Connection closed）

原因1：MCP的流式规范是Streamable HTTP，而不是SSE，直接用FastAPI会出现协议不兼容。
解决1：使用mcp-proxy或其它代理工具并设置。
原因2：后端服务没有启动。
解决2：启动调用服务。

三、结尾

MCP的意义远不止“让模型调用接口”，而是为AI生态提供了一层统一标准：
模型可自动理解标准化工具的输入输出；
客户端无需适配即可接入不同服务；
开发者无需重构即可让LLM直接使用现有功能；
工具间能通过流式能力实现更丰富交互。
更重要的是，MCP基于开放协议，任何语言、框架或部署方式都可实现自己的Server或Client。随着生态成熟，MCP有望成为“AI调用外部能力”的通用标准。

附代码：

from fastapi import FastAPI
from pydantic import BaseModel
from fastapi_mcp import FastApiMCP, AuthConfig
import httpx
import logging
logging.basicConfig(level=logging.INFO)


# 1. 定义请求&响应模型（模拟RAG）
# 这两个模型一方面用于FastAPI的入参/出参校验
# 另一方面也会被FastApiMCP用来生成MCP工具的JSON Schema
class RagQuery(BaseModel):
	"""
	RAG 查询的请求结构：
	- query: 用户查询的问题
	- top_k: 希望返回的候选文档数量（默认3）
	"""
	query: str
	top_k: int = 3

class RagAnswer(BaseModel):
	"""
	RAG 查询的响应结构：
	- answer: 给用户的回答
	- sources: 本次回答涉及到的“来源文档”列表
	"""
	answer: str
	sources: list[str]

# 2. 创建FastAPI应用
app = FastAPI(
	title="MCP RAG Demo",
	description="A demo project: expose a fake RAG endpoint as MCP tools.",
	version="0.1.0",
)

# 3. 普通REST接口：模拟一个RAG查询
# - operation_id:会作为MCP工具的name暴露出去
# - tags:配合FastApiMCP的include_tags/exclude_tags做筛选
@app.post(
	"/rag/query",
	response_model=RagAnswer,
	operation_id="rag_query_knowledge_base",  # MCP工具名
	tags=["rag"],  # 打上"rag"标签，后面只暴露这类接口
)
async def rag_query(payload: RagQuery) -> RagAnswer:
	"""
	模拟RAG查询：在真实业务中，这里可以接入向量库/知识库检索。
	"""
	fake_answer = (
		f"你问的是：{payload.query}。这是一个假 RAG 答案（top_k={payload.top_k}）。"
	)
	fake_sources = [
		"doc_1.md",
		"doc_2.pdf",
		"wiki_internal_page.html",
	]
	return RagAnswer(
		answer=fake_answer,
		sources=fake_sources[: payload.top_k],  # 截断到top_k条
	)

# 4. MCP挂载：让上述接口变成“LLM 可调用工具”
# FastApiMCP会：
# 1）扫描app中的路由；
# 2）根据include_tags/include_operations等参数筛选要暴露的端点；
# 3）基于请求/响应模型生成MCP工具的schema；
# 4）通过mount_http()在/mcp路径挂载一个MCP Server。
mcp = FastApiMCP(
	app,  # 当前FastAPI应用
	name="RAG MCP Server",  # MCP服务显示名称
	include_tags=["rag"],  # 只暴露打了"rag"标签的端点为工具
	description="A demo MCP server for RAG",  # 供客户端展示的服务描述
	describe_full_response_schema=True,  # 在工具描述中包含完整响应JSON Schema（便于调试/文档）
	describe_all_responses=True,  # 若接口有多种响应模式，也一起描述（如 200/400/500）
)

# 对于fastapi-mcp 0.4.0+，官方推荐使用 HTTP 传输：
# - mount_http(): 使用最新的Streamable HTTP规范，默认挂载路径为/mcp，可通过mount_path指定
# mcp.mount()
mcp.mount_http()


# 5. 根路由（健康检查）
@app.get("/", operation_id="health_check")
async def root():
	return {"status": "ok", "msg": "MCP RAG Demo is running."}

# 6. 启动方式：使用uvicorn作为ASGI服务器
# 1）先在终端运行本文件，启动FastAPI + MCP Server；
# 2）再到5ire里启用rag-mcp工具（内部会启动mcp-proxy连接到/mcp）。
# 使用Python 3.10，我认为最好的版本
if __name__ == "__main__":
	import uvicorn

	logging.info("服务已启动")
	uvicorn.run(
		"main:app",
		host="0.0.0.0",
		port=5000,
		reload=True,
	)