mcp-builder

MCP 构建器

根据所需工具的描述，构建可运行的 MCP 服务。生成一个可部署的 Python 服务，使用 FastMCP。

工作流

步骤 1：定义要公开的内容

询问服务需要提供什么：

• 工具 -- Claude 可以调用的函数（API 封装、计算、文件操作）
• 资源 -- Claude 可以读取的数据（数据库记录、配置、文档）
• 提示词 -- 带参数的可复用提示词模板

像“MCP 服务，用于查询我们的客户数据库”这样的简短描述就足够了。

步骤 2：搭建服务框架

pip install fastmcp

创建服务文件。服务实例必须位于模块级别：

from fastmcp import FastMCP

# MUST be at module level for FastMCP Cloud
mcp = FastMCP("My Server")

@mcp.tool()
async def search_customers(query: str) -> str:
    """Search customers by name or email."""
    # Implementation here
    return f"Found customers matching: {query}"

@mcp.resource("customers://{customer_id}")
async def get_customer(customer_id: str) -> str:
    """Get customer details by ID."""
    return f"Customer {customer_id} details"

if __name__ == "__main__":
    mcp.run()

步骤 3：添加配套 CLI 脚本（可选）

如需在 Claude Code 终端中使用，可在 MCP 服务旁添加脚本：

my-mcp-server/
├── src/index.ts          # MCP server (for Claude.ai)
├── scripts/
│   ├── search.ts         # CLI version of search tool
│   └── _shared.ts        # Shared auth/config
├── SCRIPTS.md            # Documents available scripts
└── package.json

CLI 脚本可提供 MCP 无法实现的文件 I/O、批处理和更丰富的输出。 TypeScript 模板请参阅 assets/SCRIPTS-TEMPLATE.md 和 assets/script-template.ts。

步骤 4：在本地测试

快速测试 -- 直接运行：

python server.py

通过检查器 UI 使用开发模式（推荐）：

fastmcp dev server.py
# Opens inspector at http://localhost:5173
# Hot reload, detailed logging, tool/resource inspection

面向远程客户端的 HTTP 模式：

python server.py --transport http --port 8000

使用 FastMCP Client 的自动化测试脚本：

import asyncio
from fastmcp import Client

async def test_server(server_path):
    async with Client(server_path) as client:
        # List everything
        tools = await client.list_tools()
        resources = await client.list_resources()
        prompts = await client.list_prompts()

        print(f"Tools: {[t.name for t in tools]}")
        print(f"Resources: {[r.uri for r in resources]}")
        print(f"Prompts: {[p.name for p in prompts]}")

        # Call first tool
        if tools:
            result = await client.call_tool(tools[0].name, {})
            print(f"Tool result: {result}")

        # Read first resource
        if resources:
            data = await client.read_resource(resources[0].uri)
            print(f"Resource data: {data}")

asyncio.run(test_server("server.py"))

步骤 5：部署前检查清单

部署前运行这些检查。所有必需检查都必须通过。

必需（失败将导致部署失败）：

1. 服务文件存在
2. Python 语法有效：python3 -m py_compile server.py
3. 服务对象位于模块级别（不在函数内）：

   grep -q "^mcp = FastMCP\|^server = FastMCP\|^app = FastMCP" server.py
   

4. 存在 requirements.txt，且仅包含 PyPI 软件包（不含 git+、-e、.whl、.tar.gz）
5. 不含硬编码密钥（检查 api_key = "..." 之类的模式，排除 os.getenv/os.environ）

建议（警告）：

6. requirements.txt 中列出 fastmcp
7. .gitignore 包含 .env
8. 不存在循环导入
9. 已初始化带有远程仓库的 Git 仓库
10. 服务可以加载：timeout 5 fastmcp inspect server.py

步骤 6：部署

FastMCP Cloud（最简单）：

git add . && git commit -m "Ready for deployment"
git push -u origin main
# Visit https://fastmcp.cloud, connect repo, add env vars, deploy
# URL: https://your-project.fastmcp.app/mcp

云端要求：

• 名为 mcp、server 或 app 的模块级服务对象
• requirements.txt 中仅包含 PyPI 依赖项
• 公开的 GitHub 仓库
• 使用环境变量存储密钥（不得使用硬编码值）
• 推送到 main 时自动部署，并为 PR 创建预览部署

Docker（可自托管）：

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["python", "server.py", "--transport", "http", "--port", "8000"]

Cloudflare Workers（边缘）： 使用 Cloudflare 的 agents SDK（McpAgent）以 TypeScript 构建；Cloudflare 的远程 MCP 服务文档提供当前模板。

---

关键模式

模块级服务实例

FastMCP Cloud 要求服务实例位于模块级别：

# CORRECT
mcp = FastMCP("My Server")

@mcp.tool()
def my_tool(): ...

# WRONG -- Cloud can't find the server
def create_server():
    mcp = FastMCP("My Server")
    return mcp

# FIX for factory pattern -- export at module level
def create_server() -> FastMCP:
    mcp = FastMCP("server")
    return mcp
mcp = create_server()

必须使用类型注解

FastMCP 使用类型注解生成工具架构：

@mcp.tool()
async def search(
    query: str,           # Required parameter
    limit: int = 10,      # Optional with default
    tags: list[str] = []  # Complex types supported
) -> str:
    """Docstring becomes the tool description."""
    ...

错误处理

将错误作为字符串返回，不要引发异常：

@mcp.tool()
async def get_data(id: str) -> str:
    try:
        result = await fetch_data(id)
        return json.dumps(result)
    except NotFoundError:
        return f"Error: No data found for ID {id}"

云就绪服务模式

import os
from fastmcp import FastMCP

mcp = FastMCP("production-server")
API_KEY = os.getenv("API_KEY")

@mcp.tool()
async def production_tool(data: str) -> dict:
    if not API_KEY:
        return {"error": "API_KEY not configured"}
    return {"status": "success", "data": data}

if __name__ == "__main__":
    mcp.run()

---

常见错误及修复方法

以下是你会遇到的错误。部署前请先修复。

错误	原因	修复方法
RuntimeError: No server object found at module level	服务位于函数内部	在模块级别导出 mcp = FastMCP(...)
RuntimeError: no running event loop	缺少 async/await	对异步操作使用 async def
TypeError: missing required argument 'context'	未为 Context 添加类型提示	添加带类型提示的 context: Context
ValueError: Invalid resource URI	缺少 URI scheme	使用 data://、file://、info://、api://
Resource template parameter mismatch	名称不匹配	user://{user_id} 需要 def get_user(user_id: str)
Pydantic validation error	类型提示错误	确保类型提示与实际数据类型一致
Transport mismatch	客户端和服务端协议不同	两端都使用 stdio 或都使用 http
Import errors with editable package	未安装软件包	使用 pip install -e . 或将其添加到 PYTHONPATH
DeprecationWarning: mcp.settings	API 过时	改用 os.getenv()
端口已被占用	残留进程	lsof -ti:8000 | xargs kill -9
Schema generation failure	使用了非 JSON 类型	使用兼容 JSON 的类型（不要使用 NumPy 数组）
JSON serialization error	响应中包含 datetime/bytes	转换为 .isoformat() 或字符串
Circular import	工厂函数位于 __init__.py 中	使用直接导入，避免工厂模式
Python 3.12+ datetime warning	datetime.utcnow() 已弃用	使用 datetime.now(timezone.utc)
Import-time execution	模块级别存在异步资源	使用惰性初始化模式

---

生产环境模式

自包含服务

将所有实用工具放在一个文件中，以避免循环导入：

from fastmcp import FastMCP
import os

mcp = FastMCP("my-server")

# Config
class Config:
    API_KEY = os.getenv("API_KEY", "")
    BASE_URL = os.getenv("BASE_URL", "https://api.example.com")

# Helpers
def format_success(data): return {"status": "success", "data": data}
def format_error(msg): return {"status": "error", "message": msg}

@mcp.tool()
async def my_tool(query: str) -> dict:
    if not Config.API_KEY:
        return format_error("API_KEY not configured")
    return format_success({"query": query})

惰性初始化

不要在模块级别创建异步资源。请在首次使用时初始化：

_db = None

async def get_db():
    global _db
    if _db is None:
        _db = await create_connection(Config.DB_URL)
    return _db

健康检查资源

@mcp.resource("health://status")
async def health_check() -> dict:
    return {
        "status": "healthy",
        "version": "1.0.0",
        "checks": {
            "api": "connected",
            "database": "connected"
        }
    }

连接池

import httpx

_client = None

def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None:
        _client = httpx.AsyncClient(
            base_url=Config.BASE_URL,
            headers={"Authorization": f"Bearer {Config.API_KEY}"},
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=5),
            timeout=30.0
        )
    return _client

带退避机制的重试

async def retry_with_backoff(func, max_retries=3, initial_delay=1.0):
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = initial_delay * (2 ** attempt)
            await asyncio.sleep(delay)

---

Context 功能（高级）

Context 注入

from fastmcp import Context

@mcp.tool()
async def tool_with_context(param: str, context: Context) -> dict:
    # Context parameter MUST have type hint
    pass

进度跟踪

@mcp.tool()
async def long_task(items: list[str], context: Context) -> str:
    for i, item in enumerate(items):
        await context.report_progress(i + 1, len(items), f"Processing {item}")
        await process(item)
    return "Done"

采样（在工具内调用大模型）

@mcp.tool()
async def summarise(text: str, context: Context) -> str:
    result = await context.request_sampling(
        messages=[{"role": "user", "content": f"Summarise: {text}"}],
        max_tokens=200
    )
    return result

---

CLI 快速参考

fastmcp dev server.py              # Dev mode with inspector UI
fastmcp run server.py              # Run (stdio)
fastmcp run server.py --transport http --port 8000  # Run (HTTP)
fastmcp inspect server.py          # Inspect without running
fastmcp install server.py          # Install to Claude Desktop
fastmcp deploy server.py --name my-server  # Deploy to Cloud

环境变量：FASTMCP_LOG_LEVEL（DEBUG/INFO/WARNING/ERROR）、FASTMCP_ENV（development/staging/production）。

---

集成模式（可选）

有关特定集成方法，请参阅 references/integration-patterns.md：

• 手动调用 API -- 使用可复用客户端的 httpx.AsyncClient
• 自动生成 OpenAPI -- FastMCP.from_openapi(spec, client, route_maps=[...])
• 转换 FastAPI -- FastMCP.from_fastapi(app)

---

资源文件

• assets/basic-server.py -- 最简 FastMCP 服务模板
• assets/self-contained-server.py -- 带存储和中间件的服务
• assets/tools-examples.py -- 工具模式和类型注解
• assets/resources-examples.py -- 资源 URI 模式
• assets/prompts-examples.py -- 提示词模板模式
• assets/client-example.py -- MCP 客户端用法
• assets/SCRIPTS-TEMPLATE.md -- CLI 配套文档模板
• assets/script-template.ts -- TypeScript CLI 脚本模板