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

前言

目前 MCP（Model Context Protocol）协议是给大模型插上工具链的翅膀，让大模型不仅拥有超高的推理和文本生成能力，还能具备执行大脑意识的工具能力。

如何开发一个 MCP Server

MCP 是一种协议，指的是模型上下文协议 (Model Context Protocol)。

官方 SDK 方式

安装依赖：

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__":
    mcp.run(transport='stdio')
    # mcp.run(transport="sse")

以上代码实现了 FastMCP 库来创建一个简单的 MCP 服务器，它提供一个基本的工具（工具名：crawl），用于实现源代码的获取。这是一个非常简单的爬虫 MCP 入门示例，适合理解 MCP 服务器的工作原理和 FastMCP 的基本用法。

代码分析

从 mcp.server.fastmcp 模块中导入了 FastMCP 类。
创建了一个 FastMCP 类的实例，命名为 spider。
@mcp.tool() 是一个装饰器，它将 crawl 函数注册为一个可供客户端调用的工具。
文档字符串是函数的描述，客户端会显示这个描述，帮助用户理解工具的功能。
mcp.run(transport='stdio') 启动 MCP 服务器，进入监听状态。transport='stdio' 指定了通信方式为标准输入输出，适合本地开发和测试。
如果需要远程通信，可以更改为其他传输方式（如 'sse'，Server-Sent Events）。

FastAPI-MCP 集成方式

安装依赖：

pip install fastapi-mcp

案例分析：

import requests
from fastapi import FastAPI, HTTPException, Query
from pydantic import 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 from typing import Optional app = FastAPI() # TODO: 在此处初始化 MongoDB 连接 # client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017/") # db = client["your_database"] # 定义响应模型占位符 class ResponseModel(BaseModel): data: dict message: str class ErrorResponseModel(BaseModel): error: str code: int message: str # 根据结构式获取化合物计算属性的值 @app.get("/api/computes/", status_code=status.HTTP_200_OK, summary="根据结构式获取化合物计算属性的值") async def computed(smiles: Optional[str] = Query(None, description="Search smiles string")): # result = await chemical_get_data(casno=smiles) # 模拟返回结果 result = {"code": 200, "data": {"property": "value"}} 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) # 模拟返回结果 result = {"casno": casno, "name": "Example Compound"} 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) # 模拟返回空内容 image_content = None 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) return None 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 实战指南