Python MCP Server 实战：从工具调用到业务数据查询

MCP（模型上下文协议）让大模型不再只是一个聊天的脑子，还能伸手调用外部工具。用 Python 写一个 MCP Server 并不复杂，而且如果你之前有 FastAPI 的经验，迁移成本几乎为零。

我会先走一遍最精简的例子，再给一个实际落地的化合物数据查询服务，最后说说怎么用 MCP Hub 把多个服务管起来。

一个最简单的 MCP 工具

先装好官方 SDK：

pip install mcp

下面这段代码创建了一个名叫 spider 的服务器，并注册了一个 crawl 工具。客户端只要通过 MCP 协议发来 URL，它就把网页源码吐回去。

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

这里的 FastMCP 是服务器的核心，它负责管理工具、资源和与客户端的通信。@mcp.tool() 装饰器把 crawl 函数标记成一个可调用的工具，函数文档字符串会直接作为工具描述展示给使用者。

传输方式配成了 stdio，也就是标准输入输出，特别适合本地调试。如果想让远程客户端也能调，把 transport 改成 "sse" 就行。

如果你更熟悉 FastAPI

官方还提供了 fastapi-mcp，能把一个现有的 FastAPI 应用直接暴露成 MCP 服务，不用重写工具逻辑。

pip install fastapi-mcp

下面的例子还是那个抓源码的功能，但这次用标准 FastAPI 写法，然后挂载 MCP 端点：

import requests
from fastapi import FastAPI, HTTPException, Query
from fastapi_mcp import add_mcp_server

app = FastAPI(
    title="spider",
    description="A simple example API with integrated MCP server",
    version="0.1.0",
)

@app.get("/items/{url}")
 () -> :
    
    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: 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") # 不存在则尝试用 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", name="Item API 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)