8장: 기존 시스템과 MCP 연동하기

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import pg from "pg";
 
const pool = new pg.Pool({
  host: process.env.DB_HOST || "localhost",
  port: parseInt(process.env.DB_PORT || "5432"),
  database: process.env.DB_NAME || "mydb",
  user: process.env.DB_USER,
  password: process.env.DB_PASSWORD,
});
 
const server = new McpServer({
  name: "postgres-server",
  version: "1.0.0",
});
 
// 리소스: 데이터베이스 스키마 정보
server.resource(
  "db-schema",
  "postgres://schema",
  { description: "데이터베이스 전체 테이블 스키마" },
  async () => {
    const result = await pool.query(
      "SELECT table_name, column_name, data_type, is_nullable " +
      "FROM information_schema.columns " +
      "WHERE table_schema = 'public' " +
      "ORDER BY table_name, ordinal_position"
    );
 
    const schema: Record<string, Array<{
      column: string;
      type: string;
      nullable: string;
    }>> = {};
 
    for (const row of result.rows) {
      if (!schema[row.table_name]) {
        schema[row.table_name] = [];
      }
      schema[row.table_name].push({
        column: row.column_name,
        type: row.data_type,
        nullable: row.is_nullable,
      });
    }
 
    return {
      contents: [{
        uri: "postgres://schema",
        text: JSON.stringify(schema, null, 2),
        mimeType: "application/json",
      }],
    };
  }
);
 
// 리소스: 테이블 목록과 행 수
server.resource(
  "db-tables",
  "postgres://tables",
  { description: "테이블 목록과 각 테이블의 행 수" },
  async () => {
    const result = await pool.query(
      "SELECT schemaname, tablename, n_live_tup as row_count " +
      "FROM pg_stat_user_tables " +
      "ORDER BY n_live_tup DESC"
    );
 
    return {
      contents: [{
        uri: "postgres://tables",
        text: JSON.stringify(result.rows, null, 2),
        mimeType: "application/json",
      }],
    };
  }
);
 
// 도구: SELECT 쿼리 실행 (읽기 전용)
server.tool(
  "query",
  "데이터베이스에 SELECT 쿼리를 실행합니다. SELECT 문만 허용됩니다.",
  {
    sql: z.string().describe("실행할 SQL SELECT 쿼리"),
    limit: z.number().default(100).describe("최대 반환 행 수"),
  },
  async (params) => {
    // SELECT 문만 허용
    const normalized = params.sql.trim().toUpperCase();
    if (!normalized.startsWith("SELECT")) {
      return {
        content: [{ type: "text" as const, text: "SELECT 쿼리만 실행할 수 있습니다." }],
        isError: true,
      };
    }
 
    // LIMIT 강제 적용
    const limitedSql = params.sql.replace(/;?\s*$/, "") +
      " LIMIT " + params.limit;
 
    try {
      const result = await pool.query(limitedSql);
      return {
        content: [{
          type: "text" as const,
          text: JSON.stringify({
            rows: result.rows,
            rowCount: result.rowCount,
            fields: result.fields.map((f) => f.name),
          }, null, 2),
        }],
      };
    } catch (error) {
      return {
        content: [{
          type: "text" as const,
          text: "쿼리 실행 실패: " + (error as Error).message,
        }],
        isError: true,
      };
    }
  }
);
 
// 도구: 데이터 삽입/수정 (쓰기)
server.tool(
  "execute",
  "데이터베이스에 INSERT, UPDATE, DELETE 쿼리를 실행합니다. 주의: 데이터가 변경됩니다.",
  {
    sql: z.string().describe("실행할 SQL 문"),
    params: z.array(z.unknown()).default([]).describe("매개변수화된 쿼리의 값 배열"),
  },
  async (params) => {
    // DDL 차단
    const normalized = params.sql.trim().toUpperCase();
    const blocked = ["DROP", "TRUNCATE", "ALTER", "CREATE"];
    for (const keyword of blocked) {
      if (normalized.startsWith(keyword)) {
        return {
          content: [{
            type: "text" as const,
            text: keyword + " 문은 허용되지 않습니다.",
          }],
          isError: true,
        };
      }
    }
 
    try {
      const result = await pool.query(params.sql, params.params);
      return {
        content: [{
          type: "text" as const,
          text: "실행 완료. 영향받은 행: " + result.rowCount,
        }],
      };
    } catch (error) {
      return {
        content: [{
          type: "text" as const,
          text: "실행 실패: " + (error as Error).message,
        }],
        isError: true,
      };
    }
  }
);
 
const transport = new StdioServerTransport();
await server.connect(transport);

import json
import httpx
from fastmcp import FastMCP
 
mcp = FastMCP("api-wrapper")
 
BASE_URL = "https://api.example.com/v1"
API_KEY = None  # 환경 변수에서 로드
 
import os
API_KEY = os.environ.get("API_KEY", "")
 
 
def get_headers() -> dict:
    return {
        "Authorization": "Bearer " + API_KEY,
        "Content-Type": "application/json",
    }
 
 
@mcp.tool()
async def list_users(page: int = 1, per_page: int = 20) -> str:
    """사용자 목록을 조회합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            BASE_URL + "/users",
            headers=get_headers(),
            params={"page": page, "per_page": per_page},
        )
        response.raise_for_status()
        return json.dumps(response.json(), ensure_ascii=False, indent=2)
 
 
@mcp.tool()
async def get_user(user_id: str) -> str:
    """특정 사용자의 상세 정보를 조회합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            BASE_URL + "/users/" + user_id,
            headers=get_headers(),
        )
        response.raise_for_status()
        return json.dumps(response.json(), ensure_ascii=False, indent=2)
 
 
@mcp.tool()
async def create_user(name: str, email: str, role: str = "member") -> str:
    """새 사용자를 생성합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            BASE_URL + "/users",
            headers=get_headers(),
            json={"name": name, "email": email, "role": role},
        )
        response.raise_for_status()
        return json.dumps(response.json(), ensure_ascii=False, indent=2)
 
 
# API 문서를 리소스로 제공
@mcp.resource("api://docs")
def get_api_docs() -> str:
    """API 엔드포인트 목록과 사용 가이드"""
    docs = {
        "base_url": BASE_URL,
        "endpoints": [
            {"method": "GET", "path": "/users", "description": "사용자 목록 조회"},
            {"method": "GET", "path": "/users/:id", "description": "사용자 상세 조회"},
            {"method": "POST", "path": "/users", "description": "사용자 생성"},
            {"method": "PUT", "path": "/users/:id", "description": "사용자 수정"},
            {"method": "DELETE", "path": "/users/:id", "description": "사용자 삭제"},
        ],
        "rate_limit": "60 requests/minute",
    }
    return json.dumps(docs, ensure_ascii=False, indent=2)
 
 
if __name__ == "__main__":
    mcp.run()

@mcp.tool()
async def get_order_summary(order_id: str) -> str:
    """주문 요약 정보를 조회합니다."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            BASE_URL + "/orders/" + order_id,
            headers=get_headers(),
        )
        response.raise_for_status()
        data = response.json()
 
        # 전체 API 응답 대신 핵심 정보만 추출
        summary = {
            "order_id": data["id"],
            "status": data["status"],
            "total": data["total_amount"],
            "items_count": len(data["line_items"]),
            "created_at": data["created_at"],
            "customer": data["customer"]["name"],
        }
        return json.dumps(summary, ensure_ascii=False, indent=2)

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { exec } from "child_process";
import { promisify } from "util";
 
const execAsync = promisify(exec);
 
const server = new McpServer({
  name: "cli-tools-server",
  version: "1.0.0",
});
 
// Docker 컨테이너 관리
server.tool(
  "docker_ps",
  "실행 중인 Docker 컨테이너 목록을 조회합니다",
  {
    all: z.boolean().default(false).describe("중지된 컨테이너도 포함할지 여부"),
  },
  async (params) => {
    const flag = params.all ? " -a" : "";
    const { stdout } = await execAsync(
      "docker ps" + flag + " --format '{{json .}}'"
    );
 
    const containers = stdout
      .trim()
      .split("\n")
      .filter(Boolean)
      .map((line) => JSON.parse(line));
 
    return {
      content: [{
        type: "text" as const,
        text: JSON.stringify(containers, null, 2),
      }],
    };
  }
);
 
// Git 저장소 정보
server.tool(
  "git_log",
  "Git 커밋 히스토리를 조회합니다",
  {
    repo_path: z.string().describe("Git 저장소 경로"),
    count: z.number().default(10).describe("조회할 커밋 수"),
  },
  async (params) => {
    const { stdout } = await execAsync(
      "git -C " + params.repo_path +
      " log --oneline -" + params.count +
      " --format='%H|%an|%ad|%s' --date=short"
    );
 
    const commits = stdout
      .trim()
      .split("\n")
      .filter(Boolean)
      .map((line) => {
        const [hash, author, date, ...msgParts] = line.split("|");
        return { hash, author, date, message: msgParts.join("|") };
      });
 
    return {
      content: [{
        type: "text" as const,
        text: JSON.stringify(commits, null, 2),
      }],
    };
  }
);
 
const transport = new StdioServerTransport();
await server.connect(transport);

// 기존 시스템의 인터페이스
interface LegacyOrderSystem {
  getOrder(id: string): Promise<LegacyOrder>;
  createOrder(data: LegacyOrderInput): Promise<string>;
  cancelOrder(id: string, reason: string): Promise<boolean>;
}
 
// MCP 어댑터
function registerOrderTools(
  server: McpServer,
  orderSystem: LegacyOrderSystem
) {
  server.tool(
    "get_order",
    "주문 정보를 조회합니다",
    { order_id: z.string() },
    async (params) => {
      const order = await orderSystem.getOrder(params.order_id);
      return {
        content: [{
          type: "text" as const,
          text: JSON.stringify(adaptOrderForAI(order), null, 2),
        }],
      };
    }
  );
 
  server.tool(
    "cancel_order",
    "주문을 취소합니다. 취소 사유를 반드시 포함해야 합니다.",
    {
      order_id: z.string(),
      reason: z.string().describe("취소 사유"),
    },
    async (params) => {
      const success = await orderSystem.cancelOrder(
        params.order_id,
        params.reason
      );
      return {
        content: [{
          type: "text" as const,
          text: success
            ? "주문이 취소되었습니다."
            : "주문 취소에 실패했습니다.",
        }],
      };
    }
  );
}