MCP Server 开发实战：为 Agent 打造专属工具箱

Agent (MCP Client) ←→ MCP Protocol ←→ MCP Server (工具)
                                          ├── 文件系统
                                          ├── 数据库
                                          ├── API
                                          └── 自定义工具

# 创建项目
mkdir my-mcp-server && cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk

# 或用官方模板
npx @modelcontextprotocol/create-server my-server

pip install mcp

import { Server } from "@modelcontextprotocol/sdk/server";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio";

const server = new Server({
  name: "my-tools",
  version: "1.0.0",
});

// 定义工具
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "get_weather",
      description: "获取指定城市的天气",
      inputSchema: {
        type: "object",
        properties: {
          city: { type: "string", description: "城市名" }
        },
        required: ["city"]
      }
    },
    {
      name: "send_email",
      description: "发送邮件",
      inputSchema: {
        type: "object",
        properties: {
          to: { type: "string" },
          subject: { type: "string" },
          body: { type: "string" }
        },
        required: ["to", "subject", "body"]
      }
    }
  ]
}));

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "get_weather":
      const weather = await fetchWeather(args.city);
      return { content: [{ type: "text", text: JSON.stringify(weather) }] };

    case "send_email":
      await sendEmail(args.to, args.subject, args.body);
      return { content: [{ type: "text", text: "邮件发送成功" }] };

    default:
      throw new Error(`未知工具: ${name}`);
  }
});

// 启动服务
const transport = new StdioServerTransport();
await server.connect(transport);

import { Server } from "@modelcontextprotocol/sdk/server";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

const server = new Server({ name: "db-tools", version: "1.0.0" });

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "query_db",
      description: "执行只读 SQL 查询",
      inputSchema: {
        type: "object",
        properties: {
          sql: { type: "string", description: "SELECT 查询语句" },
          limit: { type: "number", description: "返回行数限制", default: 100 }
        },
        required: ["sql"]
      }
    },
    {
      name: "list_tables",
      description: "列出数据库所有表",
      inputSchema: { type: "object", properties: {} }
    },
    {
      name: "describe_table",
      description: "描述表结构",
      inputSchema: {
        type: "object",
        properties: {
          table: { type: "string" }
        },
        required: ["table"]
      }
    }
  ]
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "query_db": {
      // 安全检查：只允许 SELECT
      if (!args.sql.trim().toUpperCase().startsWith("SELECT")) {
        throw new Error("只允许 SELECT 查询");
      }
      const result = await pool.query(args.sql + ` LIMIT ${args.limit || 100}`);
      return {
        content: [{
          type: "text",
          text: JSON.stringify(result.rows, null, 2)
        }]
      };
    }

    case "list_tables": {
      const result = await pool.query(`
        SELECT table_name FROM information_schema.tables
        WHERE table_schema = 'public'
      `);
      return {
        content: [{
          type: "text",
          text: result.rows.map(r => r.table_name).join("\n")
        }]
      };
    }

    case "describe_table": {
      const result = await pool.query(`
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = $1
      `, [args.table]);
      return {
        content: [{
          type: "text",
          text: JSON.stringify(result.rows, null, 2)
        }]
      };
    }
  }
});

from mcp import Server, Tool
import asyncio

server = Server("python-tools")

@server.tool()
def analyze_csv(file_path: str) -> str:
    """分析 CSV 文件并返回统计信息"""
    import pandas as pd
    df = pd.read_csv(file_path)
    stats = {
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": df.dtypes.to_dict(),
        "null_counts": df.isnull().sum().to_dict(),
        "sample": df.head(5).to_dict("records"),
    }
    return json.dumps(stats, ensure_ascii=False, indent=2, default=str)

@server.tool()
def generate_chart(data: dict, chart_type: str = "bar") -> str:
    """生成图表并保存"""
    import matplotlib.pyplot as plt
    fig, ax = plt.subplots()
    if chart_type == "bar":
        ax.bar(data.keys(), data.values())
    elif chart_type == "line":
        ax.plot(list(data.keys()), list(data.values()))
    plt.savefig("/tmp/chart.png")
    return "图表已保存到 /tmp/chart.png"

@server.tool()
def fetch_url(url: str) -> str:
    """获取网页内容"""
    import requests
    resp = requests.get(url, timeout=10)
    return resp.text[:5000]  # 限制返回长度

asyncio.run(server.run("stdio"))

# 官方调试工具
npx @modelcontextprotocol/inspector node dist/server.js

// MCP Server 的日志输出到 stderr（stdout 留给协议通信）
console.error(`[MCP] 工具调用: ${name} 参数: ${JSON.stringify(args)}`);

// package.json
{
  "name": "@my-org/mcp-server-db",
  "version": "1.0.0",
  "bin": {
    "mcp-server-db": "./dist/index.js"
  }
}

// MCP Client 配置
{
  "mcpServers": {
    "my-database": {
      "command": "npx",
      "args": ["-y", "@my-org/mcp-server-db"],
      "env": {
        "DATABASE_URL": "postgresql://..."
      }
    }
  }
}