从 0 开发一个 MCP Server
手把手教你构建 AI 工具调用能力

# Python 路线
python -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]" httpx pydantic

# TypeScript 路线
npm init -y && npm install @modelcontextprotocol/sdk zod
npm install -D typescript tsx @types/node

my-mcp-server/
├── pyproject.toml          # 或 package.json
├── src/
│   ├── server.py           # FastMCP 入口
│   ├── tools/              # 各工具模块
│   ├── resources/          # Resource 提供者
│   └── prompts/            # Prompt 模板
├── tests/
│   └── test_tools.py       # pytest + ClientSession
├── Dockerfile
└── README.md

// Cursor / Claude Desktop MCP 配置示例
{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["-m", "src.server"],
      "env": { "API_KEY": "your-key" }
    }
  }
}

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(..., description="搜索关键词")
    limit: int = Field(10, ge=1, le=100, description="返回条数")

@mcp.tool()
async def search_docs(params: SearchInput) -> str:
    """在文档库中搜索"""
    results = await index.search(params.query, params.limit)
    return json.dumps(results, ensure_ascii=False)

import httpx
from datetime import datetime, timezone

@mcp.tool()
async def fetch_url(url: str) -> str:
    """HTTP GET 获取 URL 内容（域名白名单内）"""
    allowed = ("api.github.com", "nodemini.com")
    if not any(url.startswith(f"https://{d}") for d in allowed):
        raise ValueError(f"Domain not allowed: {url}")
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        return resp.text[:8000]

@mcp.tool()
def get_current_time() -> str:
    """返回当前 UTC 时间"""
    return datetime.now(timezone.utc).isoformat()

from mcp.types import PromptMessage, TextContent

@mcp.prompt()
def code_review_prompt(language: str = "python") -> list[PromptMessage]:
    """标准化 Code Review 多轮模板"""
    return [
        PromptMessage(role="user", content=TextContent(
            type="text",
            text=f"你是资深 {language} 工程师。请按安全、性能、可读性三维审查以下 diff。"
        )),
        PromptMessage(role="assistant", content=TextContent(
            type="text",
            text="请粘贴 diff 或指定 PR 编号，我将按 CHECKLIST 输出结构化 review。"
        )),
    ]

# tests/test_tools.py
import pytest
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

@pytest.mark.asyncio
async def test_say_hello():
    params = StdioServerParameters(command="python", args=["-m", "src.server"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("say_hello", {"name": "MCP"})
            assert "MCP" in result.content[0].text

import chromadb
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

client = chromadb.PersistentClient(path="./data/chroma")
collection = client.get_or_create_collection(
    "wiki", embedding_function=SentenceTransformerEmbeddingFunction()
)

@mcp.tool()
def search_knowledge(query: str, top_k: int = 5) -> str:
    """语义搜索内部知识库"""
    hits = collection.query(query_texts=[query], n_results=top_k)
    return json.dumps(hits["documents"][0], ensure_ascii=False)

@mcp.tool()
def index_document(path: str) -> str:
    """索引单个 Markdown 文件"""
    text = open(path).read()
    collection.upsert(ids=[path], documents=[text], metadatas=[{"path": path}])
    return f"Indexed: {path}"