前言

2026年，AI Agent已经成为开发者必备技能，而MCP（Model Context Protocol）则是连接大模型与外部工具的核心协议。截至2026年7月，MCP生态已拥有超过3000个开源Server实现，成为连接LLM与现实世界数据的标准协议。本文带你从零开始，用Python搭建一个完整的MCP Server，让AI模型能调用你的自定义工具。

一、MCP是什么？

MCP（Model Context Protocol）是Anthropic推出的开放标准协议，为AI应用提供统一方式连接外部数据源和工具。你可以把MCP理解为"AI世界的USB接口"——任何工具只要实现MCP协议，就能被任何支持MCP的AI模型调用。

核心架构

AI模型(Claude/GPT) ←→ MCP Client ←→ MCP Server ←→ 你的工具/数据

组件	作用	示例
MCP Host	运行AI模型的宿主	Claude Desktop、Cursor
MCP Client	协议客户端，连接Server	内置于Host中
MCP Server	暴露工具和数据	文件系统、数据库、API
Transport	通信层	stdio、SSE、HTTP

二、环境准备

# 安装MCP Python SDK
pip install mcp
	# 或使用uv（推荐）
uv init mcp-demo && cd mcp-demo
uv add mcp

## 三、实战：搭建天气查询MCP Server
### 3.1 最小化Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types
import httpx
import json

app = Server("weather-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="获取指定城市的天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：上海、北京"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_weather":
        city = arguments.get("city", "未知")
        # 调用免费天气API
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"https://wttr.in/{city}",
                params={"format": "j1"},
                timeout=10
            )
            data = resp.json()
            current = data.get("current_condition", [{}])[0]
            result = {
                "city": city,
                "温度": f"{current.get('temp_C')}°C",
                "天气": current.get('weatherDesc', [{}])[0].get('value'),
                "湿度": f"{current.get('humidity')}%",
                "风速": f"{current.get('windspeedKmph')} km/h"
            }
            return [types.TextContent(
                type="text",
                text=json.dumps(result, ensure_ascii=False, indent=2)
            )]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

### 3.2 配置到Claude Desktop

在Claude Desktop的配置文件中添加：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["path/to/weather_server.py"]
    }
  }
}

配置文件位置：

• Windows: %APPDATA%\Claude\claude_desktop_config.json
• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

重启Claude Desktop后，就可以直接问："上海今天天气怎么样？"AI会自动调用你的MCP Server。
## 四、进阶：数据库查询Server

import sqlite3
from mcp.server import Server
import mcp.types as types

app = Server("db-server")

def get_db():
    conn = sqlite3.connect("products.db")
    conn.row_factory = sqlite3.Row
    return conn

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="query_products",
            description="查询商品数据库，支持按名称、价格范围筛选",
            inputSchema={
                "type": "object",
                "properties": {
                    "keyword": {"type": "string", "description": "商品名称关键词"},
                    "min_price": {"type": "number", "description": "最低价格"},
                    "max_price": {"type": "number", "description": "最高价格"},
                    "limit": {"type": "integer", "description": "返回条数", "default": 10}
                }
            }
        ),
        types.Tool(
            name="get_product_stats",
            description="获取商品统计信息（总数、平均价格等）",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    db = get_db()
    try:
        if name == "query_products":
            sql = "SELECT * FROM products WHERE 1=1"
            params = []
            if arguments.get("keyword"):
                sql += " AND name LIKE ?"
                params.append(f"%{arguments['keyword']}%")
            if arguments.get("min_price"):
                sql += " AND price >= ?"
                params.append(arguments["min_price"])
            if arguments.get("max_price"):
                sql += " AND price <= ?"
                params.append(arguments["max_price"])
            limit = arguments.get("limit", 10)
            sql += f" LIMIT {limit}"
            
            rows = db.execute(sql, params).fetchall()
            results = [dict(row) for row in rows]
            return [types.TextContent(
                type="text",
                text=json.dumps(results, ensure_ascii=False, indent=2)
            )]
            
        elif name == "get_product_stats":
            row = db.execute(
                "SELECT COUNT(*) as total, AVG(price) as avg_price, "
                "MIN(price) as min_price, MAX(price) as max_price FROM products"
            ).fetchone()
            return [types.TextContent(
                type="text",
                text=json.dumps(dict(row), ensure_ascii=False, indent=2)
            )]
    finally:
        db.close()

## 五、MCP的三大资源类型

MCP Server可以暴露三种资源：
### 5.1 Tools（工具）
AI可以调用的函数，如查询数据库、调用API。
### 5.2 Resources（资源）
AI可以读取的数据，如文件内容、数据库表。

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///config.json",
            name="配置文件",
            description="应用配置",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "file:///config.json":
        with open("config.json") as f:
            return f.read()

## 5.3 Prompts（提示词）

预设的对�Y模板，用户可以更接週择使用。

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="code-review",
            description="代码殡查助手",
            arguments=[
                types.PromptArgument(
                    name="language",
                    description="编程语",
                    required=True
                )
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
    if name == "code-review":
        lang = arguments.get("language", "Python")
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"诵审查以下{lang}代码，关注安全性、性能和可读性：\\n\\n``F{lang.lower()}\\n# 粘贴代�8\\n〦```
                    )
                )
            ]
        )

## 公、调试技巧

### 6.1 使用MCP Inspector

npx @modelcontextprotocol/inspector python weather_server.py

这会启动一个Web界面，可以可视化测试所有工具和资源。
### 6.2 日志调试

import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("mcp-server")

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info(f"Tool called: {name} with {arguments}")
    # ...

## 七、部署与发布
### 7.1 打包为命令行工具

# pyproject.toml
[project.scripts]
weather-mcp = "weather_server:main"

pip install -e .
# 现在可以直接运行
weather-mcp

### 7.2 发布到PyPI

python -m build
twine upload dist/*

### 7.3 发布到MCP Hub

将你的Server提交到 mcphub.io，让全球开发者都能使用。
## 八、变现机会

学会MCP开发后，可以通过以下方式变现：

1. 定制MCP Server：为企业开发连接内部系统的MCP Server，单价2000-10000元
2. SaaS化MCP服务：搭建MCP Server云平台，按调用次数收费
3. 技术专栏：在CSDN写MCP系列教程，积累粉丝后开付费专栏
4. 开源赞助：优质MCP Server项目可获GitHub Sponsors赞助
5. 培训课程：录制MCP开发实战课程，在网易云课堂等平台售卖

   总结

MCP是2026年AI开发最热门的方向之一。掌握MCP开发，你就能让任何AI模型使用你构建的工具，打开AI应用开发的全新可能。

专栏后续会更新：MCP+RAG实战、MCP安全防护、MCP多Server编排、企业级MCP部署等内容，欢迎关注！

---

有问题欢迎评论区交流，需要定制MCP Server开发可以私信。