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

Python MCP Server 开发实战，涵盖官方 SDK 与 FastAPI 集成方案。通过 FastMCP 快速构建工具服务器，支持 stdio 和 SSE 传输。结合 FastAPI 实现复杂业务逻辑注册为 MCP 服务，并演示本地 MCP Hub 搭建及 1Panel 部署方式。示例包含化合物数据查询接口，展示如何将企业能力接入大模型上下文协议。

技术博主发布于 2026/2/6更新于 2026/10/2190 浏览

前言

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

如何开发一个 MCP？

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

官方 SDK：https://github.com/modelcontextprotocol/python-sdk

mcp 库

pip install mcp

from mcp.server.fastmcp import FastMCP
import requests

mcp = FastMCP("spider")

@mcp.tool()
def crawl(url: str) -> int:
    """获取网页的源代码"""
    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。

FastMCP 是服务器的核心类，它负责管理工具、资源和通信。参数「spider」是服务器的名称，用于标识这个 MCP 服务器。

然后，@mcp.tool() 是一个装饰器（decorator），它告诉 FastMCP 将 crawl 函数注册为一个可供客户端调用的工具。

文档字符串（"""获取网页的源代码"""）：这是函数的描述，客户端（如 MCP Inspector）会显示这个描述，帮助用户理解工具的功能。这个工具可以被外部客户端调用！例如通过 MCP 协议发送请求，传入 url，服务器会返回它的网页源代码。

mcp.run(transport='stdio') 启动 MCP 服务器，进入监听状态，等待客户端的连接和请求。transport='stdio' 指定了通信方式为标准输入输出（Standard I/O）。这意味着服务器通过命令行的标准输入（stdin）和标准输出（stdout）与客户端通信，适合本地开发和测试。

mcp.run(transport='stdio') 启动 MCP 服务器，如果需要远程通信，可以更改为其他传输方式（如 'sse'，Server-Sent Events）。

fastapi-mcp 库

pip install fastapi-mcp

案例分析

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


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


 () -> :
    
    response = requests.get(url)
     {: response.text}

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=)

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

技术博主发布于 2026/2/6更新于 2026/10/2190 浏览

from mcp.server.fastmcp import FastMCP import requests mcp = FastMCP("spider") @mcp.tool() def crawl(url: str) -> int: """获取网页的源代码""" response = requests.get(url) return response.text if __name__ == "__main__": mcp.run(transport='stdio') # mcp.run(transport="sse")

import requests fastapi FastAPI, HTTPException, Query pydantic BaseModel typing fastapi_mcp add_mcp_server app = FastAPI( title=, description=, version=, ) () -> : response = requests.get(url) {: response.text} 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_server = add_mcp_server( app, mount_path="/mcp", # 挂着 mcp 服务器的地址 name="Item API MCP", # mcp 服务的名称 description="MCP server for the Item API", base_url="http://192.168.1.250:9999", describe_all_responses=False, describe_full_response_schema=False, ) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=9999)