为什么要自己开发MCP服务器

现有的MCP服务器库虽然已经很丰富，但还是有很多场景需要自定义：

连接公司内部API（外部工具无法访问）

特定业务逻辑处理

对接私有数据库

封装复杂工作流为简单工具

开发环境准备

bash
安装MCP Python SDK
pip install mcp
或者用uv
uv add mcp

最简单的MCP服务器

python
simple_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import mcp.types as types
创建服务器实例
app = Server("my-tools")
注册工具列表
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="calculate_discount",
            description="计算折扣后的价格",
            inputSchema={
                "type": "object",
                "properties": {
                    "original_price": {"type": "number"},
                    "discount_percent": {"type": "number"}
                },
                "required": ["original_price", "discount_percent"]
            }
        )
    ]
实现工具逻辑
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        # 实际项目中调用真实天气API
        weather_data = {"city": city, "temp": "25°C", "condition": "晴"}
        return [TextContent(type="text", text=str(weather_data))]
    
    elif name == "calculate_discount":
        original = arguments["original_price"]
        discount = arguments["discount_percent"]
        final_price = original * (1 - discount / 100)
        return [TextContent(
            type="text",
            text=f"原价 {original}元，打{100-discount}折后：{final_price:.2f}元"
        )]
    
    raise ValueError(f"Unknown tool: {name}")
启动服务器（stdio模式）
if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(app))

注册到Claude Desktop

json
{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/path/to/simple_server.py"]
    }
  }
}

实战：构建数据库查询工具

python
import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
app = Server("db-tools")
DB_PATH = "./mydata.db"
@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_database",
            description="执行SQL查询（只读）并返回结果",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SELECT SQL语句"}
                },
                "required": ["sql"]
            }
        )
    ]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_database":
        sql = arguments["sql"]
        
        # 安全验证：只允许SELECT
        if not sql.strip().upper().startswith('SELECT'):
            return [TextContent(type="text", text="错误：只允许SELECT查询")]
        
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        conn.close()
        
        result = f"列名：{columns}\n结果（{len(rows)}行）:\n"
        for row in rows[:20]:  # 最多返回20行
            result += str(row) + "\n"
        
        return [TextContent(type="text", text=result)]
if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(app))

调试技巧

bash
测试服务器是否能正常启动
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python simple_server.py
使用MCP Inspector可视化测试
npx @modelcontextprotocol/inspector python simple_server.py

发布到MCP社区

打包为Python包（pyproject.toml）

发布到PyPI：pip publish

在 mcp.so 或 GitHub 提交你的MCP服务器

用户就可以用 uvx your-package-name 直接安装