Python 开发 MongoDB 数据库 MCP Server 实战指南

前言

MCP（Model Context Protocol）协议相当于给大模型装上了工具链的翅膀。它不仅让模型拥有强大的推理和文本生成能力，还能赋予其执行具体任务的工具调用能力。

如何开发一个 MCP 服务？

MCP 是一种协议，全称为 Model Context Protocol。官方 SDK 仓库位于：https://github.com/modelcontextprotocol/python-sdk

方案一：使用官方 FastMCP 库

这是最轻量级的实现方式，适合快速验证想法。

pip install mcp

from mcp.server.fastmcp import FastMCP
import requests

# 创建服务器实例，名称用于标识
mcp = FastMCP("spider")

@mcp.tool()
def crawl(url: str) -> str:
    """获取网页的源代码"""
    response = requests.get(url)
    return response.text

if __name__ == "__main__":
    # stdio 模式适合本地开发和测试
    mcp.run(transport='stdio') 
    # 如需远程通信，可更改为 sse 模式
    # mcp.run(transport="sse")

代码解析： 这里我们导入了 FastMCP 类并实例化了一个名为 spider 的服务器。核心在于 @mcp.tool() 装饰器，它将 crawl 函数注册为一个可供客户端调用的工具。函数的文档字符串会被客户端（如 MCP Inspector）展示，帮助用户理解功能。

启动时指定 transport='stdio'，意味着服务器通过标准输入输出与客户端通信，非常适合本地调试。如果需要对外提供服务，可以切换到 sse（Server-Sent Events）传输方式。

方案二：基于 FastAPI 集成

如果你已经在使用 FastAPI 构建后端，可以直接利用 fastapi-mcp 库将现有服务注册为 MCP 服务，无需重复造轮子。

pip install fastapi-mcp

import requests
from fastapi import FastAPI, HTTPException, Query
 pydantic  BaseModel
 typing  
 fastapi_mcp  add_mcp_server


app = FastAPI(
    title=,
    description=,
    version=,
)


 () -> :
    
    :
        response = requests.get(url)
         {: response.text}
     Exception  e:
         HTTPException(status_code=, detail=(e))


mcp_server = add_mcp_server(
    app, 
    mount_path=,      
    name=,    
    description=,
    base_url=, 
    describe_all_responses=,
    describe_full_response_schema=,
)

 __name__ == :
     uvicorn
    uvicorn.run(app, host=, port=)

import sys from datetime import datetime import traceback import uvicorn from fastapi_mcp import add_mcp_server from fastapi import FastAPI, status, HTTPException, Response, Query import motor.motor_asyncio app = FastAPI() # ... 省略部分数据库连接及辅助类定义 ... # 根据结构式获取化合物计算属性的值 @app.get("/api/computes/", status_code=status.HTTP_200_OK, summary="根据结构式获取化合物计算属性的值") async def computed(smiles: Optional[str] = Query(None, description="Search smiles string")): result = ChemRdkit(smiles=smiles).computed if result and result.get("code") == 200: del result["code"] return ResponseModel(data=result, message="success") return ErrorResponseModel(error=result.get("error"), code=status.HTTP_404_NOT_FOUND, message=result.get("message")) # 获取单个化合物 @app.get("/api/chemicals/{casno}/", status_code=status.HTTP_200_OK, summary="根据 casno 获取单个化合物") async def chemicals(casno: str): """根据 casno 获取化合物""" result = await chemical_get_data(casno=casno) if result: return ResponseModel(data=result, message="success") return ErrorResponseModel(error="数据不存在", code=status.HTTP_404_NOT_FOUND, message="数据不存在") # 访问结构式图片 @app.get("/image/{filename}/", status_code=status.HTTP_200_OK, summary="访问结构式图片") async def get_image(filename: str): print(filename) image_content = struct_obj.get_struct(filename) if image_content: return Response(content=image_content, media_type="image/jpeg") # 如果文件不存在，尝试根据 filename 获取 casno 号的基本信息，使用 rdkit 库根据 smiles 生成结构式图片 obj = await get_smiles(casno=filename.replace("_", "-")) if obj: ChemRdkit(smiles=obj.get("smiles", "")).struct(filename=obj.get("casno")) image_content = struct_obj.get_struct(filename) if image_content: return Response(content=image_content, media_type="image/jpeg") return None # 注册 MCP 服务 mcp_server = add_mcp_server( app, mount_path="/mcp", name="Item API MCP", description="MCP server for the Item API", base_url="http://localhost:9999", describe_all_responses=False, describe_full_response_schema=False, ) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=9999)

Python 开发 MongoDB 数据库 MCP Server 实战指南