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

Model Context Protocol (MCP) 协议用于扩展大模型工具能力。介绍基于 Python 的两种 MCP 服务器实现方式（官方 SDK 与 FastAPI 扩展），并通过化学数据查询案例演示如何集成本地 API 至大模型生态。内容涵盖基础环境搭建、代码解析、实战应用及本地 MCP Hub 部署方案。

性能调优发布于 2026/2/6更新于 2026/4/186K 浏览

前言

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

如何开发一个 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

案例分析：

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

相关免费在线工具

RSA密钥对生成器

生成新的随机RSA私钥和公钥pem证书。在线工具，RSA密钥对生成器在线工具，online

Mermaid 预览与可视化编辑

基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。在线工具，Mermaid 预览与可视化编辑在线工具，online

curl 转代码

解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。在线工具，curl 转代码在线工具，online

Base64 字符串编码/解码

将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online

Base64 文件转换器

将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online

Markdown转HTML

将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。在线工具，Markdown转HTML在线工具，online

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)