使用Fastmcp开发MCP服务

由于时效问题,该文某些代码、技术可能已经过期,请注意!!!本文最后更新于:5 小时前

mcp服务挂载现有的fastapi上

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastmcp import FastMCP
from fastapi import FastAPI

# 创建 MCP 服务器
mcp = FastMCP(stateless_http=True) # 使用http传输务必开启
# 从 MCP 服务器创建 ASGI 应用程序
mcp_app = mcp.http_app(path='/mcp')

@mcp.tool
def add(a: int, b: int) -> int:
"""两数相加"""
return a + b


# 关键:将生命周期传递给 FastAPI
app = FastAPI(title="API", lifespan=mcp_app.lifespan)

# 挂载 MCP 服务器
app.mount("/", mcp_app)

# 现在:API 在 /,MCP 在 /mcp/


合并生命周期

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP

# 您现有的生命周期
@asynccontextmanager
async def app_lifespan(app: FastAPI):
# 启动
print("正在启动应用程序...")
# 初始化数据库、缓存等
yield
# 关闭
print("正在关闭应用程序...")

# 创建 MCP 服务器
mcp = FastMCP("工具")
mcp_app = mcp.http_app(path='/mcp')

# 合并两个生命周期
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
# 运行两个生命周期
async with app_lifespan(app):
async with mcp_app.lifespan(app):
yield

# 使用合并的生命周期
app = FastAPI(lifespan=combined_lifespan)
app.mount("/mcp", mcp_app)

参考:https://fastmcp.wiki/zh/integrations/fastapi#mounting-an-mcp-server


本博客所有文章除特别声明外,均采用 CC BY-SA 4.0 协议 ,转载请注明出处!