Claude MCP 서버를 직접 구현하면서 배운 것들 — 도구 정의, 핸들러 라우팅, 스키마 검증까지

npm install @modelcontextprotocol/sdk zod pg

// index.ts (package.json에 "type": "module" 필요, tsconfig의 module은 ES2022 이상)
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { Pool } from "pg";
 
const db = new Pool({ connectionString: process.env.DATABASE_URL });
 
const server = new McpServer({
  name: "my-db-server",
  version: "1.0.0",
});
 
server.tool(
  "query_database",
  "PostgreSQL 데이터베이스에서 사전 정의된 형태의 조회를 수행합니다",
  {
    table: z.enum(["users", "orders", "products"]).describe("조회 대상 테이블"),
    limit: z.number().int().min(1).max(100).default(10),
  },
  async ({ table, limit }) => {
    const result = await db.query(
      `SELECT * FROM ${table} LIMIT $1`,
      [limit]
    );
    return {
      content: [{ type: "text", text: JSON.stringify(result.rows, null, 2) }],
    };
  }
);
 
const transport = new StdioServerTransport();
await server.connect(transport);

from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("my-db-server")
 
@mcp.tool()
async def query_table(table: str, limit: int = 10) -> str:
    """사전 정의된 테이블에서 데이터를 조회합니다.
 
    Args:
        table: 조회 대상 테이블 (users, orders, products 중 하나)
        limit: 최대 결과 수 (1-100)
    """
    allowed = {"users", "orders", "products"}
    if table not in allowed:
        raise ValueError(f"허용되지 않은 테이블: {table}")
    result = await run_query(f"SELECT * FROM {table} LIMIT $1", [limit])
    return str(result)

// tools/database.ts
import { z } from "zod";
import type { Pool } from "pg";
 
export function makeDatabaseTools(db: Pool) {
  return {
    query_table: {
      description: "허용된 테이블의 데이터 조회",
      schema: {
        table: z.enum(["users", "orders", "products"]),
        limit: z.number().int().min(1).max(100).default(10),
      },
      handler: async ({ table, limit }: { table: "users" | "orders" | "products"; limit: number }) => {
        const result = await db.query(`SELECT * FROM ${table} LIMIT $1`, [limit]);
        return {
          content: [{ type: "text" as const, text: JSON.stringify(result.rows) }],
        };
      },
    },
    list_tables: {
      description: "public 스키마의 테이블 목록 반환",
      schema: {},
      handler: async () => {
        const result = await db.query(
          "SELECT tablename FROM pg_tables WHERE schemaname = $1",
          ["public"]
        );
        return {
          content: [{ type: "text" as const, text: JSON.stringify(result.rows) }],
        };
      },
    },
  };
}

// index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { Pool } from "pg";
import { makeDatabaseTools } from "./tools/database.js";
 
const db = new Pool({ connectionString: process.env.DATABASE_URL });
const server = new McpServer({ name: "my-db-server", version: "1.0.0" });
 
const tools = makeDatabaseTools(db);
for (const [name, tool] of Object.entries(tools)) {
  server.tool(name, tool.description, tool.schema, tool.handler);
}
 
const transport = new StdioServerTransport();
await server.connect(transport);

const inputShape = {
  sql: z
    .string()
    .min(1)
    .refine(
      (s) => s.trim().toUpperCase().startsWith("SELECT"),
      "SELECT로 시작하는 쿼리만 허용됩니다 (기초 필터)"
    ),
  limit: z.number().int().min(1).max(100).default(10),
} as const;
 
server.tool(
  "run_select",
  "임시 SELECT 쿼리 실행 (개념 예시, 프로덕션 부적합)",
  inputShape,
  async ({ sql, limit }) => {
    // 여기 도달했다면 sql, limit은 이미 SDK가 검증한 값
    // ...
  }
);

from pydantic import BaseModel, field_validator
from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("my-db-server")
 
class QueryInput(BaseModel):
    sql: str
    limit: int = 10
 
    @field_validator("sql")
    @classmethod
    def basic_select_filter(cls, v: str) -> str:
        if not v.strip().upper().startswith("SELECT"):
            raise ValueError("SELECT로 시작해야 합니다 (기초 필터)")
        return v
 
    @field_validator("limit")
    @classmethod
    def limit_range(cls, v: int) -> int:
        if not 1 <= v <= 100:
            raise ValueError("limit는 1-100 사이여야 합니다")
        return v
 
@mcp.tool()
async def query_database(input: QueryInput) -> str:
    """개념 예시. 실제로는 DB 롤 권한 + 화이트리스트로 좁히세요."""
    result = await run_query(input.sql, input.limit)
    return str(result)

{
  "mcpServers": {
    "my-db-server": {
      "command": "node",
      "args": ["/absolute/path/to/server/index.js"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    }
  }
}

{
  "mcpServers": {
    "my-db-server": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    }
  }
}

npx @modelcontextprotocol/inspector node /path/to/server/index.js

import { execFile } from "node:child_process";
import { promisify } from "node:util";
const execFileAsync = promisify(execFile);
 
server.tool(
  "git_log",
  "지정 디렉터리의 최근 커밋 조회",
  {
    repoPath: z.string().refine((p) => /^\/allowed\/repos\/[\w.-]+$/.test(p)),
    count: z.number().int().min(1).max(50).default(10),
  },
  async ({ repoPath, count }) => {
    // shell을 거치지 않고 인자를 배열로 전달 → 인젝션 원천 차단
    const { stdout } = await execFileAsync(
      "git",
      ["-C", repoPath, "log", `-n${count}`, "--oneline"],
      { timeout: 5000 }
    );
    return { content: [{ type: "text", text: stdout }] };
  }
);