MCP(Model Context Protocol) 가이드 - AI 도구 연동의 새로운 표준

from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
 
# MCP 서버 생성
server = Server("data-dynamics-tools")
 
# 도구 정의
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_database",
            description="사내 데이터베이스에 SQL 쿼리를 실행합니다. SELECT만 허용.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL SELECT 쿼리"},
                    "database": {
                        "type": "string",
                        "enum": ["analytics", "production"],
                        "description": "대상 데이터베이스"
                    }
                },
                "required": ["sql", "database"]
            }
        ),
        Tool(
            name="search_docs",
            description="사내 기술 문서를 검색합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색 질의"}
                },
                "required": ["query"]
            }
        )
    ]
 
# 도구 실행
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_database":
        result = execute_sql(arguments["sql"], arguments["database"])
        return [TextContent(type="text", text=str(result))]
    
    elif name == "search_docs":
        results = search_vector_db(arguments["query"])
        return [TextContent(type="text", text=format_results(results))]
    
    raise ValueError(f"Unknown tool: {name}")
 
# 서버 실행
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())
 
import asyncio
asyncio.run(main())

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
 
const server = new McpServer({
  name: "data-dynamics-tools",
  version: "1.0.0"
});
 
// 도구 등록
server.tool(
  "query_database",
  "사내 데이터베이스에 SQL 쿼리를 실행합니다.",
  {
    sql: z.string().describe("SQL SELECT 쿼리"),
    database: z.enum(["analytics", "production"])
  },
  async ({ sql, database }) => {
    const result = await executeSQL(sql, database);
    return { content: [{ type: "text", text: JSON.stringify(result) }] };
  }
);
 
// 리소스 등록
server.resource(
  "schema://analytics",
  "analytics 데이터베이스의 스키마 정보",
  async () => ({
    contents: [{
      uri: "schema://analytics",
      text: await getSchemaInfo("analytics"),
      mimeType: "text/plain"
    }]
  })
);
 
// 서버 시작
const transport = new StdioServerTransport();
await server.connect(transport);

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "data-dynamics-db": {
      "command": "python",
      "args": ["./mcp_servers/db_server.py"],
      "env": {
        "DB_HOST": "localhost",
        "DB_PORT": "5432"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxx"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    }
  }
}

// .claude/settings.json 또는 프로젝트 루트
{
  "mcpServers": {
    "database": {
      "command": "python",
      "args": ["./tools/db_mcp_server.py"]
    }
  }
}

# Spark 작업 관리 MCP 서버
@server.list_tools()
async def list_tools():
    return [
        Tool(name="spark_submit", description="Spark 작업을 제출합니다.", ...),
        Tool(name="spark_status", description="실행 중인 Spark 작업 상태를 조회합니다.", ...),
        Tool(name="spark_logs", description="Spark 작업 로그를 가져옵니다.", ...),
        Tool(name="hdfs_ls", description="HDFS 디렉토리를 조회합니다.", ...),
        Tool(name="hive_query", description="Hive 쿼리를 실행합니다.", ...),
    ]

# 보안이 강화된 MCP 서버
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    # 1. 입력 검증
    if name == "query_database":
        sql = arguments["sql"].strip().upper()
        if not sql.startswith("SELECT"):
            raise ValueError("SELECT 쿼리만 허용됩니다.")
        
        dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
        for keyword in dangerous_keywords:
            if keyword in sql:
                raise ValueError(f"위험한 키워드 감지: {keyword}")
    
    # 2. 실행
    result = await execute_tool(name, arguments)
    
    # 3. 감사 로그
    audit_log.info(f"Tool: {name}, Args: {arguments}, Result: {len(str(result))} chars")
    
    return result