MCP 개념

┌─────────────────────────────────────────────────────────┐
│  MCP의 핵심 가치                                          │
├─────────────────────────────────────────────────────────┤
│  1. 표준화: 모든 데이터 소스에 일관된 인터페이스 제공      │
│  2. 확장성: 새로운 도구와 리소스를 쉽게 추가              │
│  3. 재사용성: 한 번 작성한 서버를 여러 클라이언트에서 사용 │
│  4. 분리: 데이터 제공자와 AI 모델의 독립적 개발           │
└─────────────────────────────────────────────────────────┘

// 1. 파일 시스템 접근
"프로젝트 src/ 디렉토리의 모든 TypeScript 파일을 분석해줘"
→ MCP 파일시스템 서버가 파일 목록과 내용 제공

// 2. 데이터베이스 쿼리
"지난 30일간 매출 상위 10개 제품을 보여줘"
→ MCP PostgreSQL 서버가 데이터베이스 쿼리 실행

// 3. API 통합
"GitHub에서 open 상태인 이슈 중 'bug' 레이블이 있는 것들을 찾아줘"
→ MCP GitHub 서버가 GitHub API 호출

// 4. 도구 실행
"이 Python 코드를 실행하고 결과를 보여줘"
→ MCP 실행 환경 서버가 코드 실행

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

// 클라이언트 생성
const client = new Client({
  name: "my-ai-client",
  version: "1.0.0"
}, {
  capabilities: {
    tools: {}
  }
});

// 서버에 연결 (stdio 전송)
const transport = new StdioClientTransport({
  command: "python",
  args: ["server.py"]
});

await client.connect(transport);

// 사용 가능한 도구 목록 조회
const tools = await client.listTools();
console.log(tools);

from mcp.server import Server
from mcp.server.stdio import stdio_server

# 서버 생성
server = Server("my-mcp-server")

# 도구 등록
@server.tool()
async def read_file(path: str) -> str:
    """파일 내용을 읽습니다."""
    with open(path, 'r') as f:
        return f.read()

# 서버 실행
async def main():
    async with stdio_server() as streams:
        await server.run(
            streams[0],
            streams[1],
            server.create_initialization_options()
        )

// Function Calling 방식
const response = await anthropic.messages.create({
  model: "claude-4-5",
  messages: [{ role: "user", content: "파일 읽어줘" }],
  tools: [{
    name: "read_file",
    description: "파일을 읽습니다",
    input_schema: {
      type: "object",
      properties: {
        path: { type: "string" }
      }
    }
  }]
});

// 도구 실행은 클라이언트가 직접 처리
if (response.stop_reason === "tool_use") {
  const result = await readFile(toolUse.input.path);
  // 결과를 다시 모델에게 전달...
}

// ─────────────────────────────────────

// MCP 방식
const client = new MCPClient();
await client.connect(fileSystemServer);

// 도구 목록은 서버에서 자동 제공
const tools = await client.listTools();

// 도구 실행도 서버가 처리
const result = await client.callTool("read_file", {
  path: "/path/to/file"
});

// 직접 API 통합 (각 클라이언트에서 반복)
class GitHubIntegration {
  constructor(token) {
    this.token = token;
  }

  async listIssues(repo) {
    const response = await fetch(`https://api.github.com/repos/${repo}/issues`, {
      headers: { Authorization: `token ${this.token}` }
    });
    return response.json();
  }

  // 수십 개의 메서드...
}

// ─────────────────────────────────────

// MCP 방식 (한 번만 작성, 모든 클라이언트에서 재사용)
const githubServer = await client.connect("github-mcp-server");

// 모든 GitHub 기능을 표준 인터페이스로 사용
const issues = await client.callTool("github_list_issues", {
  repo: "anthropics/anthropic-sdk-python"
});

// 요청 메시지
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/home/user/example.txt"
    }
  }
}

// 응답 메시지
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Hello, MCP!"
      }
    ]
  }
}

// 에러 응답
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": {
      "details": "File not found"
    }
  }
}

// initialize 요청
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "clientInfo": {
      "name": "Claude Desktop",
      "version": "1.0.0"
    },
    "capabilities": {
      "tools": {}
    }
  }
}

// initialize 응답
{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": {
      "name": "filesystem-server",
      "version": "1.0.0"
    },
    "capabilities": {
      "tools": {},
      "resources": {}
    }
  }
}

// tools/list - 도구 목록 조회
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list"
}

// 응답
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "read_file",
        "description": "파일 내용을 읽습니다",
        "inputSchema": {
          "type": "object",
          "properties": {
            "path": {
              "type": "string",
              "description": "파일 경로"
            }
          },
          "required": ["path"]
        }
      }
    ]
  }
}

// tools/call - 도구 실행
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/etc/hosts"
    }
  }
}

// resources/list - 리소스 목록
{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "resources/list"
}

// resources/read - 리소스 읽기
{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "resources/read",
  "params": {
    "uri": "file:///path/to/document.txt"
  }
}

// prompts/list - 프롬프트 목록
{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "prompts/list"
}

// prompts/get - 프롬프트 가져오기
{
  "jsonrpc": "2.0",
  "id": 6,
  "method": "prompts/get",
  "params": {
    "name": "code-review",
    "arguments": {
      "language": "python"
    }
  }
}

// Request 메시지 스키마
{
  "jsonrpc": "2.0",        // 필수: 항상 "2.0"
  "id": 1,                   // 필수: 정수 또는 문자열 (고유 식별자)
  "method": "tools/list",   // 필수: 호출할 메서드 이름
  "params": {}               // 선택: 메서드에 전달할 매개변수
}

// 도구 목록 조회 요청 예제
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

// 도구 실행 요청 예제 (매개변수 포함)
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/home/user/document.txt"
    }
  }
}

// 성공 응답 스키마
{
  "jsonrpc": "2.0",        // 필수: 항상 "2.0"
  "id": 1,                   // 필수: 요청과 동일한 id
  "result": {               // 필수(성공): 결과 데이터
    "tools": [
      {
        "name": "read_file",
        "description": "파일 내용을 읽습니다",
        "inputSchema": {
          "type": "object",
          "properties": {
            "path": { "type": "string", "description": "파일 경로" }
          },
          "required": ["path"]
        }
      }
    ]
  }
}

// 에러 응답 스키마
{
  "jsonrpc": "2.0",        // 필수: 항상 "2.0"
  "id": 1,                   // 필수: 요청과 동일한 id
  "error": {                // 필수(실패): 에러 정보
    "code": -32601,          // 필수: 에러 코드 (정수)
    "message": "Method not found",  // 필수: 에러 메시지
    "data": {                // 선택: 추가 에러 정보
      "method": "unknown/method"
    }
  }
}

// Notification 스키마 (id 필드 없음)
{
  "jsonrpc": "2.0",        // 필수: 항상 "2.0"
  "method": "notifications/initialized",  // 필수: 알림 메서드
  "params": {}               // 선택: 알림 데이터
}

// 초기화 완료 알림
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

// 진행 상황 알림
{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "task-123",
    "progress": 75,
    "total": 100
  }
}

// 리소스 변경 알림
{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "file:///config.json"
  }
}

// 클라이언트 → 서버: initialize 요청
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "clientInfo": {
      "name": "Claude Desktop",
      "version": "1.5.0"
    },
    "capabilities": {
      "sampling": {},
      "roots": {
        "listChanged": true
      }
    }
  }
}

// 서버 → 클라이언트: initialize 응답
{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": {
      "name": "filesystem-server",
      "version": "2.1.0"
    },
    "capabilities": {
      "tools": {
        "listChanged": true
      },
      "resources": {
        "subscribe": true,
        "listChanged": true
      },
      "prompts": {
        "listChanged": true
      },
      "logging": {}
    }
  }
}

// 클라이언트 → 서버: initialized 알림
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}
// id 필드 없음 - Notification이므로 응답 불필요

// 에러 응답 예제: Parse error
{
  "jsonrpc": "2.0",
  "id": null,
  "error": {
    "code": -32700,
    "message": "Parse error",
    "data": {
      "details": "Unexpected token at position 42"
    }
  }
}

// 에러 응답 예제: Method not found
{
  "jsonrpc": "2.0",
  "id": 5,
  "error": {
    "code": -32601,
    "message": "Method not found",
    "data": {
      "method": "resources/list",
      "reason": "Server does not support resources capability"
    }
  }
}

// 에러 응답 예제: Tool execution error
{
  "jsonrpc": "2.0",
  "id": 12,
  "error": {
    "code": -32001,
    "message": "Tool execution error",
    "data": {
      "tool": "read_file",
      "reason": "Permission denied: /etc/shadow"
    }
  }
}

import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

// Python 서버 실행
const transport = new StdioClientTransport({
  command: "python",
  args: ["-m", "mcp_server_filesystem"],
  env: {
    ...process.env,
    "PYTHONPATH": "/path/to/server"
  }
});

await client.connect(transport);

import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

// 원격 서버 연결
const transport = new SSEClientTransport(
  new URL("https://mcp-server.example.com/sse")
);

await client.connect(transport);

import sys
import json

# stdio 서버의 메시지 읽기/쓰기 기본 구조
def read_message():
    """stdin에서 한 줄(JSON-RPC 메시지)을 읽습니다."""
    line = sys.stdin.readline()
    if not line:
        return None
    return json.loads(line.strip())

def write_message(message: dict):
    """stdout으로 JSON-RPC 메시지를 한 줄로 출력합니다."""
    line = json.dumps(message, ensure_ascii=False)
    sys.stdout.write(line + "\n")
    sys.stdout.flush()  # 즉시 전송 보장

def log_debug(msg: str):
    """stderr로 디버그 메시지를 출력합니다 (비프로토콜)."""
    sys.stderr.write(f"[DEBUG] {msg}\n")
    sys.stderr.flush()

# 메인 루프
while True:
    message = read_message()
    if message is None:
        break  # stdin이 닫히면 종료

    log_debug(f"수신: {message['method']}")

    # 메시지 처리...
    response = handle_request(message)
    write_message(response)

// SSE 전송 서버 구현 예제 (Express 기반)
import express from "express";

const app = express();
app.use(express.json());

// SSE 엔드포인트: 서버 → 클라이언트 스트림
app.get("/sse", (req, res) => {
  // SSE 헤더 설정
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");

  // 메시지 전송 엔드포인트 URL을 클라이언트에게 전달
  const messageEndpoint = "/messages";
  res.write(`event: endpoint\ndata: ${messageEndpoint}\n\n`);

  // JSON-RPC 메시지를 SSE 이벤트로 전송하는 함수 등록
  const sendMessage = (message) => {
    res.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
  };

  // 클라이언트 연결 등록...
  registerClient(sendMessage);

  req.on("close", () => {
    unregisterClient(sendMessage);
  });
});

// POST 엔드포인트: 클라이언트 → 서버 메시지
app.post("/messages", (req, res) => {
  const jsonRpcMessage = req.body;
  // JSON-RPC 메시지 처리...
  const response = handleMessage(jsonRpcMessage);
  res.json(response);
});

app.listen(3000);

// Streamable HTTP 서버 구현 예제
import express from "express";

const app = express();
app.use(express.json());

// 단일 엔드포인트에서 모든 MCP 통신 처리
app.post("/mcp", async (req, res) => {
  const message = req.body;
  const accept = req.headers["accept"] || "";

  // 클라이언트가 SSE를 수용하는 경우 스트리밍 응답 가능
  if (accept.includes("text/event-stream")) {
    // SSE 스트리밍 응답
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");

    // 진행 상황을 스트리밍으로 전송
    for await (const progress of processWithProgress(message)) {
      res.write(`data: ${JSON.stringify(progress)}\n\n`);
    }

    // 최종 결과 전송
    const result = await getFinalResult(message);
    res.write(`data: ${JSON.stringify(result)}\n\n`);
    res.end();
  } else {
    // 일반 JSON 응답 (단순 요청/응답)
    const result = await handleMessage(message);
    res.json(result);
  }
});

// 선택적 세션 관리
app.get("/mcp", (req, res) => {
  // 서버→클라이언트 스트림 (알림 수신용)
  res.setHeader("Content-Type", "text/event-stream");
  // 세션 ID를 통한 상태 관리
  const sessionId = req.headers["mcp-session-id"];
  // 알림 스트림 설정...
});

app.listen(3000);

@server.tool()
async def calculate(expression: str) -> str:
    """수학 표현식을 계산합니다.

    Args:
        expression: 계산할 수학 표현식 (예: "2 + 2")

    Returns:
        계산 결과
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"에러: {e}"

@server.resource("file://{path}")
async def read_file_resource(path: str) -> str:
    """파일을 리소스로 제공합니다."""
    with open(path, 'r') as f:
        return f.read()

@server.prompt()
async def code_review_prompt(language: str, code: str) -> str:
    """코드 리뷰 프롬프트를 생성합니다."""
    return f"""다음 {language} 코드를 리뷰해주세요:

```{language}
{code}
```

다음 관점에서 검토해주세요:
1. 코드 품질
2. 성능
3. 보안
4. 베스트 프랙티스
"""

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function main() {
  // 1. 클라이언트 생성
  const client = new Client({
    name: "example-client",
    version: "1.0.0"
  }, {
    capabilities: {
      tools: {}
    }
  });

  // 2. 전송 계층 설정
  const transport = new StdioClientTransport({
    command: "python",
    args: ["server.py"]
  });

  try {
    // 3. 연결 및 초기화
    await client.connect(transport);
    console.log("연결 성공!");

    // 4. 작업 수행
    const tools = await client.listTools();
    console.log("사용 가능한 도구:", tools);

    const result = await client.callTool("read_file", {
      path: "/etc/hosts"
    });
    console.log("결과:", result);

  } finally {
    // 5. 종료
    await client.close();
    console.log("연결 종료");
  }
}

main().catch(console.error);

{
  "capabilities": {
    "tools": {},                    // 도구 호출 지원
    "resources": {
      "subscribe": true          // 리소스 변경 구독
    },
    "prompts": {},                  // 프롬프트 사용 지원
    "sampling": {}                  // 샘플링 요청 수신 (서버→클라이언트)
  }
}

{
  "capabilities": {
    "tools": {
      "listChanged": true        // 도구 목록 변경 알림
    },
    "resources": {
      "subscribe": true,         // 리소스 구독 지원
      "listChanged": true        // 리소스 목록 변경 알림
    },
    "prompts": {
      "listChanged": true        // 프롬프트 목록 변경 알림
    },
    "logging": {}                   // 로그 메시지 전송
  }
}

from mcp.server import McpError

@server.tool()
async def read_file(path: str) -> str:
    if not os.path.exists(path):
        raise McpError(
            code=-32001,
            message="File not found",
            data={"path": path}
        )

    with open(path, 'r') as f:
        return f.read()

try {
  const result = await client.callTool("read_file", {
    path: "/nonexistent.txt"
  });
} catch (error) {
  if (error.code === -32001) {
    console.error("파일을 찾을 수 없습니다:", error.data.path);
  } else {
    console.error("예상치 못한 오류:", error.message);
  }
}

// Capability 확인 후 기능 사용 예제
const serverCaps = client.getServerCapabilities();

// 서버가 resources를 지원하는지 확인
if (serverCaps.resources) {
  const resources = await client.listResources();
  console.log("리소스 목록:", resources);

  // 구독 기능도 지원하는지 추가 확인
  if (serverCaps.resources.subscribe) {
    await client.subscribeResource("file:///config.json");
  } else {
    console.log("서버가 리소스 구독을 지원하지 않습니다.");
  }
} else {
  console.log("서버가 리소스 기능을 지원하지 않습니다.");
}

// 서버가 도구를 지원하는지 확인
if (serverCaps.tools) {
  const tools = await client.listTools();

  // 도구 목록 변경 알림 수신 설정
  if (serverCaps.tools.listChanged) {
    client.on("tools/list_changed", async () => {
      const updatedTools = await client.listTools();
      console.log("도구 목록이 업데이트되었습니다:", updatedTools);
    });
  }
}

// 시나리오: 클라이언트가 sampling을 지원하고, 서버가 tools만 제공하는 경우

// 1. 클라이언트 → 서버: initialize 요청
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "clientInfo": { "name": "my-client", "version": "1.0.0" },
    "capabilities": {
      "sampling": {},
      "roots": { "listChanged": true }
    }
  }
}

// 2. 서버 → 클라이언트: initialize 응답
{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": { "name": "tools-server", "version": "2.0.0" },
    "capabilities": {
      "tools": { "listChanged": true }
    }
  }
}

// 결과:
// - tools 사용 가능 (서버가 제공)
// - sampling 사용 가능 (클라이언트가 지원, 서버가 요청 가능)
// - resources 사용 불가 (서버가 미제공)
// - prompts 사용 불가 (서버가 미제공)
// - roots 사용 가능 (클라이언트가 제공, 서버가 조회 가능)

// 프로토콜 버전 예시
{
  "protocolVersion": "2024-11-05"   // 2024년 11월 5일 확정 사양
}

// 향후 새로운 버전 예시
{
  "protocolVersion": "2025-03-15"   // 가상의 차기 버전
}

// 시나리오 1: 버전 일치 (정상 케이스)

// 클라이언트 → 서버
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "clientInfo": { "name": "claude-desktop", "version": "1.5.0" },
    "capabilities": {}
  }
}

// 서버 → 클라이언트 (동일 버전 수용)
{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": { "name": "my-server", "version": "1.0.0" },
    "capabilities": {}
  }
}

// 시나리오 2: 버전 불일치 시 다운그레이드

// 클라이언트가 최신 버전을 요청
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-15",
    "clientInfo": { "name": "new-client", "version": "2.0.0" },
    "capabilities": {}
  }
}

// 서버가 구 버전만 지원하는 경우
{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2024-11-05",
    "serverInfo": { "name": "legacy-server", "version": "0.9.0" },
    "capabilities": {}
  }
}

// 클라이언트는 서버 버전을 수용하고 해당 버전의 기능만 사용
// 또는 호환 불가 시 연결을 거부할 수 있음

// 서버 측 버전 관리 구현 예제
const SUPPORTED_VERSIONS = ["2024-11-05"];
const LATEST_VERSION = "2024-11-05";

function handleInitialize(request: InitializeRequest): InitializeResult {
  const clientVersion = request.params.protocolVersion;

  // 클라이언트 요청 버전을 지원하는지 확인
  if (SUPPORTED_VERSIONS.includes(clientVersion)) {
    return {
      protocolVersion: clientVersion,
      serverInfo: { name: "my-server", version: "1.0.0" },
      capabilities: { tools: {} }
    };
  }

  // 지원하지 않으면 최신 지원 버전으로 응답
  return {
    protocolVersion: LATEST_VERSION,
    serverInfo: { name: "my-server", version: "1.0.0" },
    capabilities: { tools: {} }
  };
}