MCP Server开发教程:从零写一个自己的AI工具

📘 AI教程 💬 🔥 Trending 发布者: leakey
MCP Server开发教程:从零写一个自己的AI工具

🩺 摘要

MCP协议知道了,但怎么开发一个自己的MCP Server?

📝 详情

MCP Server从零搭建:把你的系统变成AI可调用的API

MCP是什么

MCP(Model Context Protocol)是由Anthropic提出的开放协议,类似于AI模型的"USB接口"——写一个MCP Server,任何支持MCP的AI(Claude、GPT、DeepSeek等)都能自动发现并调用你的工具。

快速开始:完整代码

第一步:安装MCP SDK

pip install mcp  # Python MCP SDK,支持stdio和HTTP传输

第二步:写一个MCP Server

# weather_mcp_server.py
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import httpx
from typing import Any

# 创建MCP Server实例
server = Server("weather-server")

# 用装饰器注册工具——这就是AI能调用的接口
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="查询指定城市的天气",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名,如北京、上海"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        ),
        types.Tool(
            name="get_forecast",
            description="获取未来7天天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "days": {"type": "integer", "maximum": 7}
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
    if name == "get_weather":
        city = arguments["city"]
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"https://api.weather.com/v1/{city}")
            data = resp.json()
        return [types.TextContent(type="text", text=json.dumps(data, ensure_ascii=False))]
    elif name == "get_forecast":
        city = arguments["city"]
        days = arguments.get("days", 3)
        # ... 预报逻辑
        return [types.TextContent(type="text", text=f"{city}未来{days}天预报...")]
    raise ValueError(f"未知工具: {name}")

# 启动Server(stdio模式,与AI客户端通过标准输入输出通信)
async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationOptions(server_name="weather-server")
        )

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

第三步:配置AI客户端

// Claude Desktop 配置(claude_desktop_config.json)
{
  "mcpServers": {
    "weather-server": {
      "command": "python",
      "args": ["weather_mcp_server.py"]
    }
  }
}

// 支持HTTP模式的配置
{
  "mcpServers": {
    "weather-server": {
      "url": "http://localhost:8080/mcp"
    }
  }
}

进阶:MCP Server的最佳实践

# mcp_best_practice.py — 带认证和错误处理
from mcp.server import Server
import mcp.types as types

server = Server("enterprise-server")
TOKEN = "your-service-token"

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        # 参数校验
        if "auth_token" not in arguments or arguments["auth_token"] != TOKEN:
            return [types.TextContent(type="text", text="{\"error\": \"未授权\"}")]

        if name == "query_database":
            sql = arguments.get("sql", "")
            # SQL注入防护
            if any(kw in sql.upper() for kw in ["DROP", "DELETE", "UPDATE"]):
                return [types.TextContent(type="text", text="{\"error\": \"只允许查询操作\"}")]
            result = execute_readonly_query(sql)
            return [types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]

        elif name == "search_knowledge_base":
            query = arguments.get("query", "")
            results = vector_db.search(query, top_k=5)
            return [types.TextContent(type="text", text=json.dumps(results, ensure_ascii=False))]

    except Exception as e:
        return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))]

    raise ValueError(f"未知工具: {name}")

已上线的MCP Server案例

类型 示例 被集成的AI数
数据库查询 MySQL/PostgreSQL查询接口 50+
搜索 Elasticsearch/Meilisearch 100+
文件操作 Google Drive/Dropbox 30+
业务系统 Salesforce/SAP 20+
自定义 企业内部系统API 5000+(GitHub)

避坑指南

  1. 工具描述要详细:AI根据description决定调用哪个工具,描述越具体AI用得越准
  2. 参数要有默认值:尽量给每个参数设定合理的默认值,降低AI调用的门槛
  3. 返回值要结构化:用JSON而不是纯文本,方便AI解析
  4. 部署首选stdio:本地开发用stdio最简单;线上用HTTP模式支持多客户端