如何使用FastAPI-MCP将任何FastAPI应用程序转换为MCP服务器?

如何使用FastAPI-MCP将任何FastAPI应用程序转换为MCP服务器?

您是否遇到过这样的情况:您希望聊天机器人使用一个工具,然后回答您的问题?听起来很复杂,对吧!但是现在,MCP(模型上下文协议)为您提供了一种将 LLM 与外部工具轻松集成的方法,LLM 将能够以各种方式使用这些工具。在本教程中,我们将深入探讨如何使用 FastAPI-MCP 转换一个由 MCP 服务器驱动的使用 FastAPI 制作的简单网络应用程序。

使用MCP的FastAPI

FastAPI 是一个非常简单的 Python 工具,可帮助您使用 API 构建网络应用程序。它设计得既简单易用又快速。将 FastAPI 想象成一个聪明的服务员,他接收您的点单(HTTP 请求),前往厨房(数据库/服务器),然后接收您的点单(输出),然后将其显示给您。它是构建网络后端、移动应用程序服务等的绝佳工具。

MCP 是 Anthropic 的一个开放标准协议,它为 LLM 提供了与外部数据源和工具通信的功能。可以将 MCP 视为一个工具包,为特定任务提供合适的工具。我们将使用 MCP 创建服务器。

现在,如果将这些功能赋予您的 LLM 呢?这将使您的生活更加轻松!这就是为什么FastAPI与MCP的集成可以提供很大帮助。FastAPI 负责不同来源的服务,而 MCP 则负责 LLM 的上下文。通过将 FastAPI 与 MCP 服务器结合使用,我们可以访问部署在网络上的所有工具,并将其用作 LLM 工具,使 LLM 更有效地开展工作。

使用MCP的FastAPI

Source: Cloudflare

什么是FastAPI-MCP?

FastAPI-MCP是一种工具,它可以让您将任何FastAPI应用程序转换成像 ChatGPT 或Claude这样的LLM可以理解并轻松使用的工具。通过使用FastAPI-MCP,您可以对FastAPI端点进行包装,使其成为利用LLM的人工智能生态系统中的即插即用工具。

如果您想了解如何使用 MCP,请阅读这篇文章:如何使用 MCP?

使用FastAPI-MCP可以将哪些API转换为MCP?

使用FastAPI-MCP,任何FastAPI端点都可以转换为LLM的MCP工具。这些端点应包括

  • GET 端点:转换为 MCP 资源。
  • POST、PUT、DELETE 端点:转换为 MCP 工具。
  • 自定义实用功能:可添加为额外的 MCP 工具

FastAPI-MCP 是一个非常易于使用的库,能自动发现这些端点并将其转换为 MCP。它还保留了这些 API 的模式和文档。

FastAPI-MCP的实际操作

让我们来看一个如何将 FastAPI 端点转换为 MCP 服务器的简单示例。首先,我们将创建一个 FastAPI 端点,然后使用 fastapi-mcp 将其转换为 MCP 服务器

配置FastAPI

1. 安装依赖项

安装所需的依赖项,使系统兼容。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
pip install fastapi fastapi_mcp uvicorn mcp-proxy
pip install fastapi fastapi_mcp uvicorn mcp-proxy
pip install fastapi fastapi_mcp uvicorn mcp-proxy

2. 导入所需的依赖项

新建一个名为“main.py”的文件,然后在其中导入以下依赖项。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
from fastapi import FastAPI, HTTPException, Query
import httpx
from fastapi_mcp import FastApiMCP
from fastapi import FastAPI, HTTPException, Query import httpx from fastapi_mcp import FastApiMCP
from fastapi import FastAPI, HTTPException, Query
import httpx
from fastapi_mcp import FastApiMCP

3. 定义 FastAPI 应用程序

让我们定义一个名为“Weather Updates API”的 FastAPI 应用程序。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
app = FastAPI(title="Weather Updates API")
app = FastAPI(title="Weather Updates API")
app = FastAPI(title="Weather Updates API")

4. 定义路由和函数

现在,我们将为应用程序定义路由,表示哪个端点将执行哪个函数。在这里,我们使用 weather.gov API(免费)制作一个天气更新应用程序,不需要任何 API 密钥。我们只需在 https://api.weather.gov/points/{lat},{lon} 中输入正确的经纬度值即可。

我们定义了一个 get_weather 函数,该函数将州名或州代码作为参数,然后在 CITY_COORDINATES 字典中找到相应的坐标,然后用这些坐标点击基础 URL。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# Predefined latitude and longitude for major cities (for simplicity)
# In a production app, you could use a geocoding service like Nominatim or Google Geocoding API
CITY_COORDINATES = {
"Los Angeles": {"lat": 34.0522, "lon": -118.2437},
"San Francisco": {"lat": 37.7749, "lon": -122.4194},
"San Diego": {"lat": 32.7157, "lon": -117.1611},
"New York": {"lat": 40.7128, "lon": -74.0060},
"Chicago": {"lat": 41.8781, "lon": -87.6298},
# Add more cities as needed
}
@app.get("/weather")
async def get_weather(
stateCode: str = Query(..., description="State code (e.g., 'CA' for California)"),
city: str = Query(..., description="City name (e.g., 'Los Angeles')")
):
"""
Retrieve today's weather from the National Weather Service API based on city and state
"""
# Get coordinates (latitude, longitude) for the given city
if city not in CITY_COORDINATES:
raise HTTPException(
status_code=404,
detail=f"City '{city}' not found in predefined list. Please use another city."
)
coordinates = CITY_COORDINATES[city]
lat, lon = coordinates["lat"], coordinates["lon"]
# URL for the NWS API Gridpoints endpoint
base_url = f"https://api.weather.gov/points/{lat},{lon}"
try:
async with httpx.AsyncClient() as client:
# First, get the gridpoint information for the given location
gridpoint_response = await client.get(base_url)
gridpoint_response.raise_for_status()
gridpoint_data = gridpoint_response.json()
# Retrieve the forecast data using the gridpoint information
forecast_url = gridpoint_data["properties"]["forecast"]
forecast_response = await client.get(forecast_url)
forecast_response.raise_for_status()
forecast_data = forecast_response.json()
# Returning today's forecast
today_weather = forecast_data["properties"]["periods"][0]
return {
"city": city,
"state": stateCode,
"date": today_weather["startTime"],
"temperature": today_weather["temperature"],
"temperatureUnit": today_weather["temperatureUnit"],
"forecast": today_weather["detailedForecast"],
}
except httpx.HTTPStatusError as e:
raise HTTPException(
status_code=e.response.status_code,
detail=f"NWS API error: {e.response.text}"
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Internal server error: {str(e)}"
)
# Predefined latitude and longitude for major cities (for simplicity) # In a production app, you could use a geocoding service like Nominatim or Google Geocoding API CITY_COORDINATES = { "Los Angeles": {"lat": 34.0522, "lon": -118.2437}, "San Francisco": {"lat": 37.7749, "lon": -122.4194}, "San Diego": {"lat": 32.7157, "lon": -117.1611}, "New York": {"lat": 40.7128, "lon": -74.0060}, "Chicago": {"lat": 41.8781, "lon": -87.6298}, # Add more cities as needed } @app.get("/weather") async def get_weather( stateCode: str = Query(..., description="State code (e.g., 'CA' for California)"), city: str = Query(..., description="City name (e.g., 'Los Angeles')") ): """ Retrieve today's weather from the National Weather Service API based on city and state """ # Get coordinates (latitude, longitude) for the given city if city not in CITY_COORDINATES: raise HTTPException( status_code=404, detail=f"City '{city}' not found in predefined list. Please use another city." ) coordinates = CITY_COORDINATES[city] lat, lon = coordinates["lat"], coordinates["lon"] # URL for the NWS API Gridpoints endpoint base_url = f"https://api.weather.gov/points/{lat},{lon}" try: async with httpx.AsyncClient() as client: # First, get the gridpoint information for the given location gridpoint_response = await client.get(base_url) gridpoint_response.raise_for_status() gridpoint_data = gridpoint_response.json() # Retrieve the forecast data using the gridpoint information forecast_url = gridpoint_data["properties"]["forecast"] forecast_response = await client.get(forecast_url) forecast_response.raise_for_status() forecast_data = forecast_response.json() # Returning today's forecast today_weather = forecast_data["properties"]["periods"][0] return { "city": city, "state": stateCode, "date": today_weather["startTime"], "temperature": today_weather["temperature"], "temperatureUnit": today_weather["temperatureUnit"], "forecast": today_weather["detailedForecast"], } except httpx.HTTPStatusError as e: raise HTTPException( status_code=e.response.status_code, detail=f"NWS API error: {e.response.text}" ) except Exception as e: raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" )
# Predefined latitude and longitude for major cities (for simplicity)
# In a production app, you could use a geocoding service like Nominatim or Google Geocoding API
CITY_COORDINATES = {
"Los Angeles": {"lat": 34.0522, "lon": -118.2437},
"San Francisco": {"lat": 37.7749, "lon": -122.4194},
"San Diego": {"lat": 32.7157, "lon": -117.1611},
"New York": {"lat": 40.7128, "lon": -74.0060},
"Chicago": {"lat": 41.8781, "lon": -87.6298},
# Add more cities as needed
}
@app.get("/weather")
async def get_weather(
stateCode: str = Query(..., description="State code (e.g., 'CA' for California)"),
city: str = Query(..., description="City name (e.g., 'Los Angeles')")
):
"""
Retrieve today's weather from the National Weather Service API based on city and state
"""
# Get coordinates (latitude, longitude) for the given city
if city not in CITY_COORDINATES:
raise HTTPException(
status_code=404,
detail=f"City '{city}' not found in predefined list. Please use another city."
)
coordinates = CITY_COORDINATES[city]
lat, lon = coordinates["lat"], coordinates["lon"]
# URL for the NWS API Gridpoints endpoint
base_url = f"https://api.weather.gov/points/{lat},{lon}"
try:
async with httpx.AsyncClient() as client:
# First, get the gridpoint information for the given location
gridpoint_response = await client.get(base_url)
gridpoint_response.raise_for_status()
gridpoint_data = gridpoint_response.json()
# Retrieve the forecast data using the gridpoint information
forecast_url = gridpoint_data["properties"]["forecast"]
forecast_response = await client.get(forecast_url)
forecast_response.raise_for_status()
forecast_data = forecast_response.json()
# Returning today's forecast
today_weather = forecast_data["properties"]["periods"][0]
return {
"city": city,
"state": stateCode,
"date": today_weather["startTime"],
"temperature": today_weather["temperature"],
"temperatureUnit": today_weather["temperatureUnit"],
"forecast": today_weather["detailedForecast"],
}
except httpx.HTTPStatusError as e:
raise HTTPException(
status_code=e.response.status_code,
detail=f"NWS API error: {e.response.text}"
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Internal server error: {str(e)}"
)

5. 设置MCP服务器

现在让我们使用 fastapi-mcp 库将 FastAPI 应用程序转换为 MCP。这个过程非常简单,我们只需添加几行代码,fastapi-mcp 就会自动将端点转换为 MCP 工具,并轻松检测其模式和文档。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
mcp = FastApiMCP(
app,
name="Weather Updates API",
description="API for retrieving today's weather from weather.gov",
)
mcp.mount()
mcp = FastApiMCP( app, name="Weather Updates API", description="API for retrieving today's weather from weather.gov", ) mcp.mount()
mcp = FastApiMCP(
app,
name="Weather Updates API",
description="API for retrieving today's weather from weather.gov",
)
mcp.mount()

6. 启动应用程序

现在,在 Python 文件末尾添加以下内容。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

然后转到终端,运行 main.py 文件。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
python main.py
python main.py
python main.py

现在,您的 FastAPI 应用程序应能在 localhost 中成功启动。

配置Cursor

让我们配置用于测试 MCP 服务器的 Cursor IDE。

  1. 从此处下载 Cursor:https://www.cursor.com/downloads
  2. 安装后,注册并进入主屏幕。

Cursor

  1. 现在从顶部工具栏进入 File,点击 Preferences,然后点击 Cursor Settings

Cursor Settings

  1. 在 Cursor 设置中,点击 MCP

MCP

  1. 在 MCP 选项卡上,点击 Add new global MCP Server,然后打开一个 mcp.json 文件。粘贴以下代码并保存文件。
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
{
"mcpServers": {
"National Park Service": {
"command": "mcp-proxy",
"args": ["http://127.0.0.1:8000/mcp"]
}
}
}
{ "mcpServers": { "National Park Service": { "command": "mcp-proxy", "args": ["http://127.0.0.1:8000/mcp"] } } }
{
"mcpServers": {
"National Park Service": {
"command": "mcp-proxy",
"args": ["http://127.0.0.1:8000/mcp"]
}
}
}
  1. 回到 Cursor Settings,您应该看到以下内容:

MCP Server

如果你在屏幕上看到这个,说明你的服务器已成功运行并连接到 Cursor IDE。如果显示错误,请尝试使用右上角的重启按钮。

我们已经在 Cursor IDE 中成功设置了 MCP 服务器。现在,让我们测试一下服务器。

测试MCP服务器

我们的 MCP 服务器可以检索天气更新。我们只需向 Cursor IDE 询问任何地点的天气更新,它就会使用 MCP 服务器为我们获取。

查询:Please tell me what is today’s weather in San Diego

查询圣地亚哥天气

查询:New York weather?

从输出结果可以看出,我们的 MCP 服务器运行良好。我们只需询问天气详情,它就会自行决定是否使用 MCP 服务器。在第二个输出中,我们含糊地问了 “纽约天气如何?”它就能根据我们之前的提示了解到查询的上下文,并使用适当的 MCP 工具进行回答。

小结

MCP 允许 LLM 通过访问外部工具来提高其回答能力,而 FastAPI 则提供了一种简单的方法来实现这一点。在本综合指南中,我们使用 fastapi-mcp 库将这两种技术结合起来。利用这个库,我们可以将任何 API 转换为 MCP 服务器,这将有助于 LLM 和人工智能代理从 API 获取最新信息。无需为每项新任务定义自定义工具。带有 FastAPI 的 MCP 将自动处理一切。MCP的引入带来了LLM的革命,而现在,FastAPI与MCP的搭配正在彻底改变LLM访问这些工具的方式。

评论留言