MCP 서버 개발

# pip 사용
pip install mcp

# uv 사용 (권장)
uv pip install mcp

# 또는 프로젝트 의존성에 추가
uv add mcp

#!/usr/bin/env python3
"""
간단한 MCP 서버 예제
"""

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

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

# 도구 등록
@server.tool()
async def hello(name: str = "World") -> str:
    """
    인사 메시지를 반환합니다.

    Args:
        name: 인사할 이름

    Returns:
        인사 메시지
    """
    return f"Hello, {name}!"

# 메인 함수
async def main():
    # stdio 전송으로 서버 실행
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

@server.tool()
async def add(a: float, b: float) -> float:
    """두 숫자를 더합니다."""
    return a + b

@server.tool()
async def multiply(a: float, b: float) -> float:
    """두 숫자를 곱합니다."""
    return a * b

from typing import List, Dict, Any
from mcp.types import TextContent, ImageContent, EmbeddedResource

@server.tool()
async def search_files(
    directory: str,
    pattern: str,
    max_results: int = 10
) -> List[Dict[str, Any]]:
    """
    디렉토리에서 파일을 검색합니다.

    Args:
        directory: 검색할 디렉토리
        pattern: 파일명 패턴 (glob)
        max_results: 최대 결과 수

    Returns:
        파일 정보 리스트
    """
    import os
    import glob

    search_path = os.path.join(directory, pattern)
    files = glob.glob(search_path, recursive=True)

    results = []
    for filepath in files[:max_results]:
        stat = os.stat(filepath)
        results.append({
            "path": filepath,
            "size": stat.st_size,
            "modified": stat.st_mtime
        })

    return results

import os
from pathlib import Path

@server.tool()
async def read_file(path: str) -> str:
    """파일 내용을 읽습니다."""
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

@server.tool()
async def write_file(path: str, content: str) -> str:
    """파일에 내용을 씁니다."""
    # 디렉토리가 없으면 생성
    os.makedirs(os.path.dirname(path), exist_ok=True)

    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)

    return f"파일 작성 완료: {path}"

@server.tool()
async def list_directory(path: str) -> List[Dict[str, Any]]:
    """디렉토리 내용을 나열합니다."""
    items = []
    for entry in os.listdir(path):
        full_path = os.path.join(path, entry)
        is_dir = os.path.isdir(full_path)

        items.append({
            "name": entry,
            "path": full_path,
            "type": "directory" if is_dir else "file"
        })

    return items

from mcp.server import McpError

@server.tool()
async def safe_read_file(path: str) -> str:
    """안전하게 파일을 읽습니다."""
    # 경로 검증
    if ".." in path:
        raise McpError(
            code=-32602,
            message="Invalid path: path traversal not allowed"
        )

    # 파일 존재 확인
    if not os.path.exists(path):
        raise McpError(
            code=-32001,
            message="File not found",
            data={"path": path}
        )

    # 읽기 권한 확인
    if not os.access(path, os.R_OK):
        raise McpError(
            code=-32002,
            message="Permission denied",
            data={"path": path}
        )

    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        raise McpError(
            code=-32603,
            message=f"Failed to read file: {e}"
        )

from mcp.types import Resource, TextContent

# 1. 리소스 목록 핸들러
@server.list_resources()
async def list_resources() -> List[Resource]:
    """사용 가능한 리소스 목록을 반환합니다."""
    return [
        Resource(
            uri="file:///etc/hosts",
            name="System hosts file",
            description="시스템 호스트 파일",
            mimeType="text/plain"
        ),
        Resource(
            uri="config://app",
            name="Application config",
            description="애플리케이션 설정",
            mimeType="application/json"
        )
    ]

# 2. 리소스 읽기 핸들러
@server.read_resource()
async def read_resource(uri: str) -> str:
    """리소스 내용을 읽습니다."""
    if uri.startswith("file://"):
        path = uri[7:]  # "file://" 제거
        with open(path, 'r') as f:
            return f.read()

    elif uri == "config://app":
        import json
        config = {
            "version": "1.0.0",
            "debug": False
        }
        return json.dumps(config, indent=2)

    else:
        raise McpError(
            code=-32602,
            message=f"Unknown resource URI: {uri}"
        )

@server.list_resources()
async def list_project_files() -> List[Resource]:
    """프로젝트 파일을 동적으로 리소스로 제공합니다."""
    resources = []
    project_dir = "/path/to/project"

    for root, dirs, files in os.walk(project_dir):
        for filename in files:
            if filename.endswith(('.py', '.js', '.ts')):
                filepath = os.path.join(root, filename)
                uri = f"file://{filepath}"

                resources.append(Resource(
                    uri=uri,
                    name=filename,
                    description=f"Source file: {filepath}",
                    mimeType="text/plain"
                ))

    return resources

from mcp.types import Prompt, PromptMessage, TextContent

# 1. 프롬프트 목록
@server.list_prompts()
async def list_prompts() -> List[Prompt]:
    """사용 가능한 프롬프트 목록을 반환합니다."""
    return [
        Prompt(
            name="code-review",
            description="코드 리뷰 프롬프트",
            arguments=[
                {
                    "name": "language",
                    "description": "프로그래밍 언어",
                    "required": True
                },
                {
                    "name": "code",
                    "description": "리뷰할 코드",
                    "required": True
                }
            ]
        ),
        Prompt(
            name="translate",
            description="번역 프롬프트",
            arguments=[
                {
                    "name": "text",
                    "description": "번역할 텍스트",
                    "required": True
                },
                {
                    "name": "target_lang",
                    "description": "목표 언어",
                    "required": True
                }
            ]
        )
    ]

# 2. 프롬프트 가져오기
@server.get_prompt()
async def get_prompt(name: str, arguments: Dict[str, str]) -> List[PromptMessage]:
    """프롬프트를 생성하여 반환합니다."""
    if name == "code-review":
        language = arguments.get("language")
        code = arguments.get("code")

        prompt_text = f"""다음 {language} 코드를 리뷰해주세요:

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

다음 관점에서 검토해주세요:
1. 코드 품질 및 가독성
2. 성능 최적화 가능성
3. 보안 취약점
4. 베스트 프랙티스 준수 여부
5. 개선 제안

각 항목에 대해 구체적인 피드백을 제공해주세요."""

        return [
            PromptMessage(
                role="user",
                content=TextContent(type="text", text=prompt_text)
            )
        ]

    elif name == "translate":
        text = arguments.get("text")
        target_lang = arguments.get("target_lang")

        prompt_text = f"""다음 텍스트를 {target_lang}로 번역해주세요:

{text}

번역 시 다음을 고려해주세요:
- 자연스러운 표현 사용
- 문화적 맥락 반영
- 전문 용어의 정확한 번역"""

        return [
            PromptMessage(
                role="user",
                content=TextContent(type="text", text=prompt_text)
            )
        ]

    else:
        raise McpError(
            code=-32602,
            message=f"Unknown prompt: {name}"
        )

#!/usr/bin/env python3
"""
완전한 MCP 파일시스템 서버
"""

import os
import json
from typing import List, Dict, Any
from pathlib import Path

from mcp.server import Server, McpError
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Prompt, PromptMessage, TextContent

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

# 허용된 디렉토리 (보안)
ALLOWED_DIRS = [
    str(Path.home() / "Documents"),
    str(Path.home() / "Projects")
]

def is_path_allowed(path: str) -> bool:
    """경로가 허용된 디렉토리 내에 있는지 확인합니다."""
    abs_path = os.path.abspath(path)
    return any(abs_path.startswith(allowed) for allowed in ALLOWED_DIRS)

# ===== 도구 =====

@server.tool()
async def read_file(path: str) -> str:
    """파일 내용을 읽습니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        raise McpError(code=-32001, message="File not found")
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def write_file(path: str, content: str) -> str:
    """파일에 내용을 씁니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Successfully wrote {len(content)} bytes to {path}"
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def list_directory(path: str) -> List[Dict[str, Any]]:
    """디렉토리 내용을 나열합니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        items = []
        for entry in os.listdir(path):
            full_path = os.path.join(path, entry)
            stat = os.stat(full_path)

            items.append({
                "name": entry,
                "path": full_path,
                "type": "directory" if os.path.isdir(full_path) else "file",
                "size": stat.st_size,
                "modified": stat.st_mtime
            })

        return items
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def search_files(
    directory: str,
    pattern: str,
    max_results: int = 50
) -> List[str]:
    """파일을 검색합니다."""
    if not is_path_allowed(directory):
        raise McpError(code=-32602, message="Path not allowed")

    import glob

    search_path = os.path.join(directory, "**", pattern)
    results = glob.glob(search_path, recursive=True)
    return results[:max_results]

# ===== 리소스 =====

@server.list_resources()
async def list_resources() -> List[Resource]:
    """허용된 디렉토리를 리소스로 나열합니다."""
    resources = []
    for dir_path in ALLOWED_DIRS:
        if os.path.exists(dir_path):
            resources.append(Resource(
                uri=f"file://{dir_path}",
                name=os.path.basename(dir_path),
                description=f"Directory: {dir_path}",
                mimeType="text/directory"
            ))
    return resources

@server.read_resource()
async def read_resource(uri: str) -> str:
    """리소스를 읽습니다."""
    if not uri.startswith("file://"):
        raise McpError(code=-32602, message="Only file:// URIs supported")

    path = uri[7:]
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    if os.path.isdir(path):
        # 디렉토리 내용을 JSON으로 반환
        items = await list_directory(path)
        return json.dumps(items, indent=2)
    else:
        # 파일 내용 반환
        return await read_file(path)

# ===== 메인 =====

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# 프로젝트 초기화
mkdir my-mcp-server
cd my-mcp-server
npm init -y

# 의존성 설치
npm install @modelcontextprotocol/sdk
npm install -D typescript @types/node

# TypeScript 설정
npx tsc --init

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules"]
}

#!/usr/bin/env node
// src/index.ts

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  Tool
} from "@modelcontextprotocol/sdk/types.js";

// 서버 생성
const server = new Server(
  {
    name: "my-mcp-server",
    version: "1.0.0"
  },
  {
    capabilities: {
      tools: {}
    }
  }
);

// 도구 목록
const TOOLS: Tool[] = [
  {
    name: "hello",
    description: "인사 메시지를 반환합니다",
    inputSchema: {
      type: "object",
      properties: {
        name: {
          type: "string",
          description: "인사할 이름"
        }
      },
      required: []
    }
  }
];

// 도구 목록 핸들러
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: TOOLS
}));

// 도구 호출 핸들러
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "hello") {
    const userName = (args as { name?: string }).name || "World";
    return {
      content: [
        {
          type: "text",
          text: `Hello, ${userName}!`
        }
      ]
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

// 서버 실행
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

import * as fs from "fs/promises";
import * as path from "path";

// 도구 정의
const TOOLS: Tool[] = [
  {
    name: "read_file",
    description: "파일 내용을 읽습니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "파일 경로"
        }
      },
      required: ["path"]
    }
  },
  {
    name: "write_file",
    description: "파일에 내용을 씁니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "파일 경로"
        },
        content: {
          type: "string",
          description: "파일 내용"
        }
      },
      required: ["path", "content"]
    }
  },
  {
    name: "list_directory",
    description: "디렉토리 내용을 나열합니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "디렉토리 경로"
        }
      },
      required: ["path"]
    }
  }
];

// 도구 구현
async function readFile(filePath: string): Promise<string> {
  const content = await fs.readFile(filePath, "utf-8");
  return content;
}

async function writeFile(filePath: string, content: string): Promise<string> {
  await fs.mkdir(path.dirname(filePath), { recursive: true });
  await fs.writeFile(filePath, content, "utf-8");
  return `Successfully wrote ${content.length} bytes to ${filePath}`;
}

async function listDirectory(dirPath: string) {
  const entries = await fs.readdir(dirPath, { withFileTypes: true });

  const items = await Promise.all(
    entries.map(async (entry) => {
      const fullPath = path.join(dirPath, entry.name);
      const stats = await fs.stat(fullPath);

      return {
        name: entry.name,
        path: fullPath,
        type: entry.isDirectory() ? "directory" : "file",
        size: stats.size,
        modified: stats.mtime
      };
    })
  );

  return items;
}

// 도구 호출 핸들러
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  try {
    switch (name) {
      case "read_file": {
        const { path } = args as { path: string };
        const content = await readFile(path);
        return {
          content: [{ type: "text", text: content }]
        };
      }

      case "write_file": {
        const { path, content } = args as { path: string; content: string };
        const result = await writeFile(path, content);
        return {
          content: [{ type: "text", text: result }]
        };
      }

      case "list_directory": {
        const { path } = args as { path: string };
        const items = await listDirectory(path);
        return {
          content: [{ type: "text", text: JSON.stringify(items, null, 2) }]
        };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    const errorMessage = error instanceof Error ? error.message : String(error);
    return {
      content: [{ type: "text", text: `Error: ${errorMessage}` }],
      isError: true
    };
  }
});

import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
  Resource
} from "@modelcontextprotocol/sdk/types.js";

// 리소스 목록 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const resources: Resource[] = [
    {
      uri: "file:///etc/hosts",
      name: "System hosts",
      description: "시스템 호스트 파일",
      mimeType: "text/plain"
    },
    {
      uri: "config://app",
      name: "App config",
      description: "애플리케이션 설정",
      mimeType: "application/json"
    }
  ];

  return { resources };
});

// 리소스 읽기 핸들러
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;

  if (uri.startsWith("file://")) {
    const filePath = uri.slice(7);
    const content = await fs.readFile(filePath, "utf-8");

    return {
      contents: [
        {
          uri,
          mimeType: "text/plain",
          text: content
        }
      ]
    };
  }

  if (uri === "config://app") {
    const config = {
      version: "1.0.0",
      debug: false
    };

    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(config, null, 2)
        }
      ]
    };
  }

  throw new Error(`Unknown resource: ${uri}`);
});

#!/usr/bin/env node
// src/index.ts - 완전한 파일시스템 MCP 서버

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
  Tool,
  Resource
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs/promises";
import * as path from "path";
import { glob } from "glob";

// 허용된 디렉토리
const ALLOWED_DIRS = [
  path.join(process.env.HOME || "", "Documents"),
  path.join(process.env.HOME || "", "Projects")
];

function isPathAllowed(filePath: string): boolean {
  const absolute = path.resolve(filePath);
  return ALLOWED_DIRS.some((dir) => absolute.startsWith(dir));
}

// 서버 생성
const server = new Server(
  {
    name: "filesystem-server",
    version: "1.0.0"
  },
  {
    capabilities: {
      tools: {},
      resources: {}
    }
  }
);

// 도구 정의
const TOOLS: Tool[] = [
  {
    name: "read_file",
    description: "파일 내용을 읽습니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "파일 경로" }
      },
      required: ["path"]
    }
  },
  {
    name: "write_file",
    description: "파일에 내용을 씁니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "파일 경로" },
        content: { type: "string", description: "파일 내용" }
      },
      required: ["path", "content"]
    }
  },
  {
    name: "list_directory",
    description: "디렉토리 내용을 나열합니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "디렉토리 경로" }
      },
      required: ["path"]
    }
  },
  {
    name: "search_files",
    description: "파일을 검색합니다",
    inputSchema: {
      type: "object",
      properties: {
        directory: { type: "string", description: "검색 디렉토리" },
        pattern: { type: "string", description: "파일명 패턴 (glob)" }
      },
      required: ["directory", "pattern"]
    }
  }
];

// 핸들러 등록
server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: TOOLS }));

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

  try {
    switch (name) {
      case "read_file": {
        const { path: filePath } = args as { path: string };
        if (!isPathAllowed(filePath)) {
          throw new Error("Path not allowed");
        }
        const content = await fs.readFile(filePath, "utf-8");
        return { content: [{ type: "text", text: content }] };
      }

      case "write_file": {
        const { path: filePath, content } = args as { path: string; content: string };
        if (!isPathAllowed(filePath)) {
          throw new Error("Path not allowed");
        }
        await fs.mkdir(path.dirname(filePath), { recursive: true });
        await fs.writeFile(filePath, content, "utf-8");
        return { content: [{ type: "text", text: `Wrote to ${filePath}` }] };
      }

      case "list_directory": {
        const { path: dirPath } = args as { path: string };
        if (!isPathAllowed(dirPath)) {
          throw new Error("Path not allowed");
        }
        const entries = await fs.readdir(dirPath, { withFileTypes: true });
        const items = entries.map((e) => ({
          name: e.name,
          type: e.isDirectory() ? "directory" : "file"
        }));
        return { content: [{ type: "text", text: JSON.stringify(items, null, 2) }] };
      }

      case "search_files": {
        const { directory, pattern } = args as { directory: string; pattern: string };
        if (!isPathAllowed(directory)) {
          throw new Error("Path not allowed");
        }
        const files = await glob(pattern, { cwd: directory });
        return { content: [{ type: "text", text: JSON.stringify(files, null, 2) }] };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${error}` }],
      isError: true
    };
  }
});

// 리소스 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const resources: Resource[] = ALLOWED_DIRS
    .filter((dir) => fs.access(dir).then(() => true).catch(() => false))
    .map((dir) => ({
      uri: `file://${dir}`,
      name: path.basename(dir),
      description: `Directory: ${dir}`,
      mimeType: "text/directory"
    }));

  return { resources };
});

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;

  if (!uri.startsWith("file://")) {
    throw new Error("Only file:// URIs supported");
  }

  const filePath = uri.slice(7);
  if (!isPathAllowed(filePath)) {
    throw new Error("Path not allowed");
  }

  const content = await fs.readFile(filePath, "utf-8");

  return {
    contents: [{
      uri,
      mimeType: "text/plain",
      text: content
    }]
  };
});

// 서버 실행
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

[project]
name = "mcp-server-filesystem"
version = "1.0.0"
description = "MCP filesystem server"
authors = [{ name = "Your Name", email = "you@example.com" }]
requires-python = ">=3.10"
dependencies = [
    "mcp>=0.1.0"
]

[project.scripts]
mcp-server-filesystem = "mcp_server_filesystem:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# 개발 모드 설치
uv pip install -e .

# 또는 pip
pip install -e .

# 서버 실행
mcp-server-filesystem

{
  "name": "@myorg/mcp-server-filesystem",
  "version": "1.0.0",
  "description": "MCP filesystem server",
  "type": "module",
  "bin": {
    "mcp-server-filesystem": "./dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "prepare": "npm run build"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^0.1.0"
  },
  "devDependencies": {
    "typescript": "^5.3.0",
    "@types/node": "^20.0.0"
  }
}

# 빌드
npm run build

# 글로벌 설치
npm install -g .

# 서버 실행
mcp-server-filesystem

# Python (PyPI)
uv build
uv publish

# TypeScript (npm)
npm publish

# MCP Inspector 사용
npx @modelcontextprotocol/inspector python server.py

# 또는 TypeScript 서버
npx @modelcontextprotocol/inspector node dist/index.js

import pytest
from mcp.client import Client
from mcp.client.stdio import StdioClientTransport

@pytest.mark.asyncio
async def test_server():
    client = Client({"name": "test", "version": "1.0.0"}, {"capabilities": {"tools": {}}})

    transport = StdioClientTransport({
        "command": "python",
        "args": ["server.py"]
    })

    await client.connect(transport)

    # 도구 목록 테스트
    tools = await client.list_tools()
    assert len(tools) > 0

    # 도구 호출 테스트
    result = await client.call_tool("hello", {"name": "Test"})
    assert "Hello, Test!" in result

    await client.close()

# 프로젝트 디렉토리 생성
mkdir my-mcp-server
cd my-mcp-server

# 기본 구조 생성
mkdir -p src/my_mcp_server
mkdir -p tests
touch src/my_mcp_server/__init__.py
touch src/my_mcp_server/server.py
touch src/my_mcp_server/tools.py
touch src/my_mcp_server/resources.py
touch tests/__init__.py
touch tests/test_server.py

my-mcp-server/
├── pyproject.toml
├── README.md
├── src/
│   └── my_mcp_server/
│       ├── __init__.py
│       ├── server.py      # 메인 서버 진입점
│       ├── tools.py        # 도구 정의
│       └── resources.py    # 리소스 정의
└── tests/
    ├── __init__.py
    └── test_server.py

[project]
name = "my-mcp-server"
version = "0.1.0"
description = "내 첫 MCP 서버"
requires-python = ">=3.10"
dependencies = [
    "mcp>=1.0.0",
    "httpx>=0.25.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "pytest-asyncio>=0.21",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project.scripts]
my-mcp-server = "my_mcp_server.server:main"

# uv 설치 (아직 없는 경우)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 가상 환경 생성 및 의존성 설치
uv venv
source .venv/bin/activate

# pyproject.toml 기반 의존성 설치
uv pip install -e ".[dev]"

# 또는 uv 프로젝트 매니저로 직접 추가
uv add mcp
uv add httpx
uv add --dev pytest pytest-asyncio

"""MCP 서버 패키지"""
__version__ = "0.1.0"

"""MCP 서버 메인 모듈"""
import asyncio
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server

# 로거 설정
logger = logging.getLogger(__name__)

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

# 도구와 리소스를 별도 모듈에서 임포트하여 등록
from my_mcp_server import tools     # noqa: E402, F401
from my_mcp_server import resources  # noqa: E402, F401

async def run_server():
    """서버를 실행합니다."""
    logger.info("MCP 서버 시작 중...")

    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

def main():
    """CLI 진입점"""
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run_server())

if __name__ == "__main__":
    main()

"""MCP 서버 도구 정의"""
import json
import httpx
from typing import Any
from mcp.types import Tool, TextContent
from my_mcp_server.server import server

# --- 도구 목록 핸들러 ---
@server.list_tools()
async def list_tools() -> list[Tool]:
    """사용 가능한 도구 목록을 반환합니다."""
    return [
        Tool(
            name="get_weather",
            description="지정된 도시의 현재 날씨 정보를 조회합니다. "
                        "도시명은 한글 또는 영문으로 입력할 수 있습니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "날씨를 조회할 도시명 (예: 서울, Tokyo, New York)"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "온도 단위 (기본값: celsius)",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="calculate",
            description="수학 계산을 수행합니다. 사칙연산과 거듭제곱을 지원합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["add", "subtract", "multiply", "divide", "power"],
                        "description": "수행할 연산 종류"
                    },
                    "a": {
                        "type": "number",
                        "description": "첫 번째 피연산자"
                    },
                    "b": {
                        "type": "number",
                        "description": "두 번째 피연산자"
                    }
                },
                "required": ["operation", "a", "b"]
            }
        ),
        Tool(
            name="search_web",
            description="웹 검색을 수행하여 최신 정보를 조회합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 쿼리"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "최대 결과 수 (1~20)",
                        "minimum": 1,
                        "maximum": 20,
                        "default": 5
                    },
                    "language": {
                        "type": "string",
                        "description": "검색 결과 언어 코드 (예: ko, en, ja)",
                        "default": "ko"
                    }
                },
                "required": ["query"]
            }
        )
    ]

# --- 도구 실행 핸들러 ---
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """도구 호출을 처리합니다."""

    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")

        # 실제로는 날씨 API를 호출합니다
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.weatherapi.com/v1/current.json",
                params={"key": "YOUR_API_KEY", "q": city}
            )
            data = response.json()

        return [TextContent(
            type="text",
            text=json.dumps(data, ensure_ascii=False, indent=2)
        )]

    elif name == "calculate":
        op = arguments["operation"]
        a = arguments["a"]
        b = arguments["b"]

        operations = {
            "add": lambda: a + b,
            "subtract": lambda: a - b,
            "multiply": lambda: a * b,
            "divide": lambda: a / b if b != 0 else "오류: 0으로 나눌 수 없습니다",
            "power": lambda: a ** b,
        }
        result = operations[op]()

        return [TextContent(type="text", text=str(result))]

    else:
        raise ValueError(f"알 수 없는 도구: {name}")

"""MCP 서버 리소스 정의"""
import json
import os
from datetime import datetime
from mcp.types import Resource, TextContent
from my_mcp_server.server import server

# --- 리소스 목록 핸들러 ---
@server.list_resources()
async def list_resources() -> list[Resource]:
    """사용 가능한 리소스 목록을 반환합니다."""
    return [
        Resource(
            uri="config://app",
            name="앱 설정",
            description="현재 애플리케이션의 설정 정보",
            mimeType="application/json"
        ),
        Resource(
            uri="status://server",
            name="서버 상태",
            description="서버의 현재 상태 정보",
            mimeType="application/json"
        ),
        Resource(
            uri="file:///var/log/app.log",
            name="앱 로그",
            description="애플리케이션 최근 로그",
            mimeType="text/plain"
        )
    ]

# --- 리소스 읽기 핸들러 ---
@server.read_resource()
async def read_resource(uri: str) -> str:
    """리소스 내용을 읽습니다."""

    if uri == "config://app":
        config = {
            "app_name": "My MCP Server",
            "version": "0.1.0",
            "debug": os.environ.get("DEBUG", "false"),
            "log_level": os.environ.get("LOG_LEVEL", "INFO"),
        }
        return json.dumps(config, indent=2, ensure_ascii=False)

    elif uri == "status://server":
        status = {
            "status": "running",
            "uptime": "정상",
            "timestamp": datetime.now().isoformat(),
        }
        return json.dumps(status, indent=2, ensure_ascii=False)

    elif uri.startswith("file://"):
        path = uri[7:]
        if not os.path.exists(path):
            return f"파일을 찾을 수 없습니다: {path}"
        with open(path, "r", encoding="utf-8") as f:
            return f.read()

    raise ValueError(f"알 수 없는 리소스: {uri}")

"""MCP 서버 프롬프트 정의"""
from mcp.types import Prompt, PromptArgument, PromptMessage, TextContent
from my_mcp_server.server import server

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    """사용 가능한 프롬프트 템플릿 목록"""
    return [
        Prompt(
            name="code-review",
            description="코드 리뷰를 수행하는 프롬프트",
            arguments=[
                PromptArgument(
                    name="code",
                    description="리뷰할 코드",
                    required=True
                ),
                PromptArgument(
                    name="language",
                    description="프로그래밍 언어 (예: python, javascript)",
                    required=False
                )
            ]
        ),
        Prompt(
            name="explain-error",
            description="에러 메시지를 분석하고 해결책을 제안합니다",
            arguments=[
                PromptArgument(
                    name="error",
                    description="에러 메시지 또는 스택 트레이스",
                    required=True
                ),
                PromptArgument(
                    name="context",
                    description="에러가 발생한 상황 설명",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict | None = None) -> list[PromptMessage]:
    """프롬프트 템플릿을 렌더링합니다."""
    args = arguments or {}

    if name == "code-review":
        code = args.get("code", "")
        language = args.get("language", "알 수 없음")
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"""다음 {language} 코드를 리뷰해주세요.

검토 항목:
1. 코드 품질 및 가독성
2. 잠재적 버그나 에러
3. 성능 개선 가능성
4. 보안 취약점
5. 모범 사례 준수 여부

코드:
```
{code}
```"""
                )
            )
        ]

    elif name == "explain-error":
        error = args.get("error", "")
        context = args.get("context", "추가 컨텍스트 없음")
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"""다음 에러를 분석하고 해결책을 제안해주세요.

에러 메시지:
```
{error}
```

상황: {context}

다음을 포함하여 답변해주세요:
1. 에러의 원인
2. 구체적인 해결 방법
3. 향후 예방 방법"""
                )
            )
        ]

    raise ValueError(f"알 수 없는 프롬프트: {name}")

import logging
import traceback
from functools import wraps
from mcp.types import TextContent, INTERNAL_ERROR, INVALID_PARAMS

logger = logging.getLogger(__name__)

# MCP 표준 에러 코드
ERROR_CODES = {
    "PARSE_ERROR": -32700,
    "INVALID_REQUEST": -32600,
    "METHOD_NOT_FOUND": -32601,
    "INVALID_PARAMS": -32602,
    "INTERNAL_ERROR": -32603,
}

def safe_tool(func):
    """도구 함수에 에러 처리를 추가하는 데코레이터"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except ValueError as e:
            logger.warning(f"잘못된 입력: {e}")
            return [TextContent(
                type="text",
                text=f"입력 오류: {e}"
            )]
        except FileNotFoundError as e:
            logger.warning(f"파일 없음: {e}")
            return [TextContent(
                type="text",
                text=f"파일을 찾을 수 없습니다: {e}"
            )]
        except PermissionError as e:
            logger.error(f"권한 오류: {e}")
            return [TextContent(
                type="text",
                text=f"권한이 없습니다: {e}"
            )]
        except Exception as e:
            logger.error(f"예기치 못한 오류: {traceback.format_exc()}")
            return [TextContent(
                type="text",
                text=f"내부 오류가 발생했습니다: {type(e).__name__}: {e}"
            )]
    return wrapper

import logging
import sys

def setup_logging(level: str = "INFO") -> logging.Logger:
    """MCP 서버용 로깅을 설정합니다.

    중요: MCP는 stdout을 JSON-RPC에 사용하므로
    로그는 반드시 stderr로 출력합니다.
    """
    logger = logging.getLogger("my-mcp-server")
    logger.setLevel(getattr(logging, level.upper()))

    # stderr 핸들러 (stdout 사용 금지!)
    handler = logging.StreamHandler(sys.stderr)
    handler.setLevel(getattr(logging, level.upper()))

    # 포맷 설정
    formatter = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    return logger

# 사용 예시
logger = setup_logging("DEBUG")
logger.debug("디버그 메시지")
logger.info("서버가 시작되었습니다")
logger.warning("경고: 설정 파일이 없습니다")
logger.error("에러 발생")

# 직접 실행
python -m my_mcp_server.server

# 또는 pyproject.toml에 정의한 스크립트로 실행
my-mcp-server

# MCP Inspector로 테스트 (웹 UI 제공)
npx @modelcontextprotocol/inspector python -m my_mcp_server.server

# MCP Inspector가 브라우저를 열고 대화형 테스트 UI를 제공합니다
# 기본 주소: http://localhost:5173

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "python",
      "args": ["-m", "my_mcp_server.server"],
      "cwd": "/path/to/my-mcp-server",
      "env": {
        "DEBUG": "true",
        "LOG_LEVEL": "DEBUG"
      }
    }
  }
}

# 프로젝트 생성
mkdir my-mcp-server-ts
cd my-mcp-server-ts
npm init -y

# 디렉토리 구조 생성
mkdir -p src
mkdir -p tests

{
  "name": "my-mcp-server-ts",
  "version": "0.1.0",
  "description": "TypeScript MCP 서버",
  "type": "module",
  "main": "dist/index.js",
  "bin": {
    "my-mcp-server-ts": "dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "dev": "tsc --watch",
    "start": "node dist/index.js",
    "inspect": "npx @modelcontextprotocol/inspector node dist/index.js",
    "test": "vitest"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0"
  },
  "devDependencies": {
    "typescript": "^5.3.0",
    "@types/node": "^20.0.0",
    "vitest": "^1.0.0"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist", "tests"]
}

# 핵심 의존성 설치
npm install @modelcontextprotocol/sdk

# 개발 의존성 설치
npm install -D typescript @types/node vitest

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// 서버 인스턴스 생성
const server = new Server(
  {
    name: "my-mcp-server-ts",
    version: "0.1.0",
  },
  {
    capabilities: {
      tools: {},
      resources: {},
    },
  }
);

// 도구 목록 핸들러
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "get_weather",
        description: "지정된 도시의 현재 날씨 정보를 조회합니다",
        inputSchema: {
          type: "object" as const,
          properties: {
            city: {
              type: "string",
              description: "날씨를 조회할 도시명",
            },
            units: {
              type: "string",
              enum: ["celsius", "fahrenheit"],
              description: "온도 단위",
            },
          },
          required: ["city"],
        },
      },
      {
        name: "calculate",
        description: "수학 계산을 수행합니다",
        inputSchema: {
          type: "object" as const,
          properties: {
            operation: {
              type: "string",
              enum: ["add", "subtract", "multiply", "divide"],
              description: "수행할 연산",
            },
            a: { type: "number", description: "첫 번째 피연산자" },
            b: { type: "number", description: "두 번째 피연산자" },
          },
          required: ["operation", "a", "b"],
        },
      },
    ],
  };
});

// 도구 실행 핸들러
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "get_weather": {
      const city = args?.city as string;
      // 실제로는 날씨 API를 호출합니다
      return {
        content: [
          {
            type: "text",
            text: `${city}의 현재 날씨: 맑음, 22도`,
          },
        ],
      };
    }
    case "calculate": {
      const { operation, a, b } = args as {
        operation: string;
        a: number;
        b: number;
      };
      let result: number;
      switch (operation) {
        case "add": result = a + b; break;
        case "subtract": result = a - b; break;
        case "multiply": result = a * b; break;
        case "divide":
          if (b === 0) throw new Error("0으로 나눌 수 없습니다");
          result = a / b;
          break;
        default: throw new Error(`알 수 없는 연산: ${operation}`);
      }
      return {
        content: [{ type: "text", text: String(result) }],
      };
    }
    default:
      throw new Error(`알 수 없는 도구: ${name}`);
  }
});

// 리소스 목록 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "config://app",
        name: "앱 설정",
        description: "현재 애플리케이션 설정",
        mimeType: "application/json",
      },
    ],
  };
});

// 리소스 읽기 핸들러
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;

  if (uri === "config://app") {
    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(
            { appName: "My Server", version: "0.1.0", debug: true },
            null,
            2
          ),
        },
      ],
    };
  }

  throw new Error(`알 수 없는 리소스: ${uri}`);
});

// 서버 실행
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP 서버가 stdio에서 실행 중입니다");
}

main().catch(console.error);

# TypeScript 빌드
npm run build

# 서버 실행
npm start

# MCP Inspector로 테스트
npm run inspect

# Claude Desktop 설정 (claude_desktop_config.json)
# {
#   "mcpServers": {
#     "my-ts-server": {
#       "command": "node",
#       "args": ["/path/to/my-mcp-server-ts/dist/index.js"]
#     }
#   }
# }

# FastMCP 설치
pip install fastmcp
# 또는
uv add fastmcp

from fastmcp import FastMCP

# 서버 인스턴스 생성 (이름과 설정)
mcp = FastMCP(
    "My Fast Server",
    dependencies=["httpx", "beautifulsoup4"]
)

# --- 도구 등록 (데코레이터로 간단하게) ---

@mcp.tool()
def add(a: int, b: int) -> int:
    """두 수를 더합니다.

    Args:
        a: 첫 번째 숫자
        b: 두 번째 숫자

    Returns:
        두 수의 합
    """
    return a + b

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """두 수를 곱합니다."""
    return a * b

@mcp.tool()
async def fetch_url(url: str, timeout: int = 30) -> str:
    """URL에서 웹 페이지 내용을 가져옵니다.

    Args:
        url: 가져올 웹 페이지 URL
        timeout: 요청 타임아웃 (초, 기본값: 30)
    """
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=timeout)
        return response.text

# --- 리소스 등록 ---

@mcp.resource("config://app")
def get_config() -> str:
    """앱 설정을 반환합니다."""
    import json
    config = {"name": "My App", "version": "1.0", "debug": True}
    return json.dumps(config, indent=2)

@mcp.resource("status://server")
def get_status() -> str:
    """서버 상태를 반환합니다."""
    return "서버 정상 운영 중"

# --- 프롬프트 등록 ---

@mcp.prompt()
def review_prompt(code: str) -> str:
    """코드 리뷰 프롬프트를 생성합니다."""
    return f"다음 코드를 리뷰해주세요:\n\n```\n{code}\n```"

@mcp.prompt()
def debug_prompt(error: str, context: str = "") -> str:
    """디버깅 프롬프트를 생성합니다."""
    msg = f"다음 에러를 분석해주세요:\n\n{error}"
    if context:
        msg += f"\n\n상황: {context}"
    return msg

# FastMCP에서 이 함수를 등록하면:
@mcp.tool()
def search(
    query: str,
    max_results: int = 10,
    include_archived: bool = False
) -> list[str]:
    """데이터를 검색합니다.

    Args:
        query: 검색 쿼리 문자열
        max_results: 최대 결과 수
        include_archived: 보관된 항목 포함 여부
    """
    ...

# FastMCP가 자동 생성하는 inputSchema:
# {
#   "type": "object",
#   "properties": {
#     "query": {
#       "type": "string",
#       "description": "검색 쿼리 문자열"
#     },
#     "max_results": {
#       "type": "integer",
#       "default": 10,
#       "description": "최대 결과 수"
#     },
#     "include_archived": {
#       "type": "boolean",
#       "default": false,
#       "description": "보관된 항목 포함 여부"
#     }
#   },
#   "required": ["query"]
# }

# 직접 실행 (stdio 모드)
fastmcp run server.py

# MCP Inspector로 테스트
fastmcp dev server.py

# Claude Desktop에 설치
fastmcp install server.py --name "My Fast Server"

{
  "type": "object",
  "properties": {
    // 문자열 타입
    "name": {
      "type": "string",
      "description": "사용자 이름"
    },
    // 숫자 타입 (정수)
    "age": {
      "type": "integer",
      "description": "나이",
      "minimum": 0,
      "maximum": 150
    },
    // 숫자 타입 (실수)
    "score": {
      "type": "number",
      "description": "점수 (0.0 ~ 100.0)",
      "minimum": 0.0,
      "maximum": 100.0
    },
    // 불리언 타입
    "active": {
      "type": "boolean",
      "description": "활성화 여부",
      "default": true
    }
  },
  "required": ["name", "age"]
}

{
  "type": "object",
  "properties": {
    "language": {
      "type": "string",
      "enum": ["python", "javascript", "typescript", "rust", "go"],
      "description": "프로그래밍 언어"
    },
    "priority": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"],
      "description": "우선순위 수준",
      "default": "medium"
    },
    "format": {
      "type": "string",
      "enum": ["json", "csv", "xml", "yaml"],
      "description": "출력 형식"
    }
  },
  "required": ["language"]
}

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "검색 쿼리"
    },
    "filters": {
      "type": "object",
      "description": "검색 필터 조건",
      "properties": {
        "date_range": {
          "type": "object",
          "properties": {
            "start": {
              "type": "string",
              "description": "시작 날짜 (YYYY-MM-DD)"
            },
            "end": {
              "type": "string",
              "description": "종료 날짜 (YYYY-MM-DD)"
            }
          }
        },
        "category": {
          "type": "string",
          "description": "카테고리 필터"
        },
        "min_score": {
          "type": "number",
          "description": "최소 점수 필터",
          "minimum": 0
        }
      }
    },
    "options": {
      "type": "object",
      "description": "검색 옵션",
      "properties": {
        "page": {
          "type": "integer",
          "default": 1,
          "minimum": 1,
          "description": "페이지 번호"
        },
        "per_page": {
          "type": "integer",
          "default": 20,
          "minimum": 1,
          "maximum": 100,
          "description": "페이지당 결과 수"
        }
      }
    }
  },
  "required": ["query"]
}

{
  "type": "object",
  "properties": {
    // 문자열 배열
    "tags": {
      "type": "array",
      "items": {
        "type": "string"
      },
      "description": "태그 목록",
      "minItems": 1,
      "maxItems": 10
    },
    // 객체 배열
    "items": {
      "type": "array",
      "description": "항목 목록",
      "items": {
        "type": "object",
        "properties": {
          "name": {
            "type": "string",
            "description": "항목 이름"
          },
          "quantity": {
            "type": "integer",
            "minimum": 1,
            "description": "수량"
          }
        },
        "required": ["name", "quantity"]
      }
    },
    // 숫자 배열
    "scores": {
      "type": "array",
      "items": {
        "type": "number",
        "minimum": 0,
        "maximum": 100
      },
      "description": "점수 배열 (0~100)"
    }
  },
  "required": ["tags"]
}

{
  "type": "object",
  "properties": {
    "table": {
      "type": "string",
      "description": "쿼리할 테이블명 (예: users, orders, products)"
    },
    "columns": {
      "type": "array",
      "items": { "type": "string" },
      "description": "조회할 컬럼 목록 (빈 배열이면 전체 컬럼)"
    },
    "where": {
      "type": "array",
      "description": "WHERE 조건 목록",
      "items": {
        "type": "object",
        "properties": {
          "column": { "type": "string", "description": "컬럼명" },
          "operator": {
            "type": "string",
            "enum": ["=", "!=", ">", "<", ">=", "<=", "LIKE", "IN"],
            "description": "비교 연산자"
          },
          "value": {
            "description": "비교 값 (문자열, 숫자, 또는 배열)"
          }
        },
        "required": ["column", "operator", "value"]
      }
    },
    "order_by": {
      "type": "object",
      "description": "정렬 조건",
      "properties": {
        "column": { "type": "string", "description": "정렬 컬럼" },
        "direction": {
          "type": "string",
          "enum": ["ASC", "DESC"],
          "default": "ASC",
          "description": "정렬 방향"
        }
      }
    },
    "limit": {
      "type": "integer",
      "default": 50,
      "minimum": 1,
      "maximum": 1000,
      "description": "최대 결과 수 (기본값: 50, 최대: 1000)"
    }
  },
  "required": ["table"]
}

from fastmcp import FastMCP

mcp = FastMCP("Unit Converter")

@mcp.tool()
def celsius_to_fahrenheit(celsius: float) -> float:
    """섭씨를 화씨로 변환합니다."""
    return celsius * 9 / 5 + 32

@mcp.tool()
def kg_to_pounds(kg: float) -> float:
    """킬로그램을 파운드로 변환합니다."""
    return kg * 2.20462

@mcp.tool()
def km_to_miles(km: float) -> float:
    """킬로미터를 마일로 변환합니다."""
    return km * 0.621371

import os
import httpx
from fastmcp import FastMCP

mcp = FastMCP("GitHub API Proxy")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")

async def github_request(endpoint: str) -> dict:
    """GitHub API 요청 헬퍼"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.github.com{endpoint}",
            headers={
                "Authorization": f"token {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.v3+json"
            }
        )
        response.raise_for_status()
        return response.json()

@mcp.tool()
async def list_repos(username: str, sort: str = "updated") -> str:
    """GitHub 사용자의 레포지토리 목록을 조회합니다.

    Args:
        username: GitHub 사용자명
        sort: 정렬 기준 (created, updated, pushed, full_name)
    """
    data = await github_request(
        f"/users/{username}/repos?sort={sort}"
    )
    repos = [{"name": r["name"], "stars": r["stargazers_count"],
              "description": r["description"]} for r in data[:10]]
    import json
    return json.dumps(repos, ensure_ascii=False, indent=2)

@mcp.tool()
async def get_repo_issues(
    owner: str,
    repo: str,
    state: str = "open"
) -> str:
    """레포지토리의 이슈 목록을 조회합니다.

    Args:
        owner: 레포지토리 소유자
        repo: 레포지토리 이름
        state: 이슈 상태 (open, closed, all)
    """
    data = await github_request(
        f"/repos/{owner}/{repo}/issues?state={state}"
    )
    issues = [{"number": i["number"], "title": i["title"],
               "state": i["state"]} for i in data[:20]]
    import json
    return json.dumps(issues, ensure_ascii=False, indent=2)

from fastmcp import FastMCP
import json

mcp = FastMCP("System Monitor")

@mcp.tool()
async def system_overview() -> str:
    """시스템 전체 상태를 집계합니다.
    CPU, 메모리, 디스크 정보를 통합하여 반환합니다."""
    import psutil

    overview = {
        "cpu": {
            "percent": psutil.cpu_percent(interval=1),
            "count": psutil.cpu_count(),
        },
        "memory": {
            "total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
            "used_percent": psutil.virtual_memory().percent,
        },
        "disk": {
            "total_gb": round(psutil.disk_usage("/").total / (1024**3), 2),
            "used_percent": psutil.disk_usage("/").percent,
        },
    }
    return json.dumps(overview, indent=2)

@mcp.resource("monitor://dashboard")
def dashboard_data() -> str:
    """대시보드용 집계 데이터를 반환합니다."""
    import psutil
    data = {
        "cpu_percent": psutil.cpu_percent(),
        "memory_percent": psutil.virtual_memory().percent,
        "disk_percent": psutil.disk_usage("/").percent,
        "process_count": len(psutil.pids()),
    }
    return json.dumps(data, indent=2)

# Python 서버 테스트
npx @modelcontextprotocol/inspector python -m my_mcp_server.server

# FastMCP 서버 테스트
fastmcp dev server.py

# TypeScript 서버 테스트
npx @modelcontextprotocol/inspector node dist/index.js

# 환경 변수 전달
GITHUB_TOKEN=xxx npx @modelcontextprotocol/inspector python server.py

# Inspector가 http://localhost:5173에서 실행됩니다
# - Tools 탭: 도구 목록 확인 및 호출 테스트
# - Resources 탭: 리소스 목록 확인 및 읽기 테스트
# - Prompts 탭: 프롬프트 목록 확인 및 렌더링 테스트

"""MCP 서버 단위 테스트"""
import pytest
import json
from unittest.mock import AsyncMock, patch
from mcp.types import TextContent

# 서버 모듈에서 핸들러를 가져옵니다
from my_mcp_server.tools import call_tool, list_tools

# --- 도구 목록 테스트 ---

@pytest.mark.asyncio
async def test_list_tools():
    """도구 목록이 올바르게 반환되는지 테스트"""
    tools = await list_tools()

    # 도구 수 확인
    assert len(tools) >= 2

    # 도구 이름 확인
    tool_names = [t.name for t in tools]
    assert "get_weather" in tool_names
    assert "calculate" in tool_names

    # inputSchema 검증
    calc_tool = next(t for t in tools if t.name == "calculate")
    schema = calc_tool.inputSchema
    assert schema["type"] == "object"
    assert "operation" in schema["properties"]
    assert "operation" in schema["required"]

# --- 계산 도구 테스트 ---

@pytest.mark.asyncio
async def test_calculate_add():
    """덧셈 연산 테스트"""
    result = await call_tool(
        "calculate",
        {"operation": "add", "a": 5, "b": 3}
    )
    assert len(result) == 1
    assert result[0].text == "8"

@pytest.mark.asyncio
async def test_calculate_divide_by_zero():
    """0으로 나누기 에러 처리 테스트"""
    result = await call_tool(
        "calculate",
        {"operation": "divide", "a": 10, "b": 0}
    )
    assert "0으로 나눌 수 없습니다" in result[0].text

@pytest.mark.asyncio
async def test_unknown_tool():
    """알 수 없는 도구 호출 시 에러 테스트"""
    with pytest.raises(ValueError, match="알 수 없는 도구"):
        await call_tool("nonexistent", {})

# --- 리소스 테스트 ---

@pytest.mark.asyncio
async def test_read_config_resource():
    """config 리소스 읽기 테스트"""
    from my_mcp_server.resources import read_resource
    result = await read_resource("config://app")
    config = json.loads(result)
    assert "app_name" in config
    assert "version" in config

@pytest.mark.asyncio
async def test_unknown_resource():
    """알 수 없는 리소스 URI 테스트"""
    from my_mcp_server.resources import read_resource
    with pytest.raises(ValueError):
        await read_resource("unknown://test")

# 전체 테스트 실행
pytest tests/ -v

# 특정 테스트만 실행
pytest tests/test_server.py::test_calculate_add -v

# 커버리지 포함
pytest tests/ --cov=my_mcp_server --cov-report=html

"""MCP 서버 통합 테스트"""
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

@pytest.fixture
async def mcp_client():
    """MCP 클라이언트를 생성하여 서버에 연결합니다."""
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "my_mcp_server.server"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

@pytest.mark.asyncio
async def test_server_initialization(mcp_client):
    """서버 초기화 테스트"""
    # 세션이 정상적으로 초기화되면 성공
    assert mcp_client is not None

@pytest.mark.asyncio
async def test_list_tools_via_protocol(mcp_client):
    """프로토콜을 통한 도구 목록 조회 테스트"""
    result = await mcp_client.list_tools()
    assert len(result.tools) > 0

    tool_names = [t.name for t in result.tools]
    assert "calculate" in tool_names

@pytest.mark.asyncio
async def test_call_tool_via_protocol(mcp_client):
    """프로토콜을 통한 도구 호출 테스트"""
    result = await mcp_client.call_tool(
        "calculate",
        {"operation": "multiply", "a": 6, "b": 7}
    )
    assert result.content[0].text == "42"

@pytest.mark.asyncio
async def test_list_resources_via_protocol(mcp_client):
    """프로토콜을 통한 리소스 목록 조회 테스트"""
    result = await mcp_client.list_resources()
    assert len(result.resources) > 0

    uris = [r.uri for r in result.resources]
    assert "config://app" in uris

name: MCP Server Tests

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]

    steps:
    - uses: actions/checkout@v4

    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: ${{ matrix.python-version }}

    - name: Install uv
      run: curl -LsSf https://astral.sh/uv/install.sh | sh

    - name: Install dependencies
      run: uv pip install -e ".[dev]"

    - name: Run unit tests
      run: pytest tests/ -v --tb=short

    - name: Run tests with coverage
      run: pytest tests/ --cov=my_mcp_server --cov-report=xml

    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: coverage.xml

[project]
name = "my-mcp-server"
version = "1.0.0"
description = "MCP 서버 - 유용한 도구 모음"
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.10"
authors = [
    { name = "개발자", email = "dev@example.com" },
]
keywords = ["mcp", "ai", "claude", "tools"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Programming Language :: Python :: 3",
    "License :: OSI Approved :: MIT License",
]
dependencies = [
    "mcp>=1.0.0",
    "httpx>=0.25.0",
]

[project.scripts]
my-mcp-server = "my_mcp_server.server:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# 빌드 도구 설치
pip install build twine

# 패키지 빌드
python -m build

# 빌드 결과 확인 (dist/ 디렉토리)
ls dist/
# my_mcp_server-1.0.0-py3-none-any.whl
# my_mcp_server-1.0.0.tar.gz

# TestPyPI에 먼저 테스트 업로드
twine upload --repository testpypi dist/*

# 실제 PyPI에 업로드
twine upload dist/*

# 사용자는 이제 이렇게 설치할 수 있습니다
pip install my-mcp-server
# 또는
uvx my-mcp-server

{
  "name": "@myorg/mcp-server",
  "version": "1.0.0",
  "description": "MCP 서버 - 유용한 도구 모음",
  "type": "module",
  "main": "dist/index.js",
  "types": "dist/index.d.ts",
  "bin": {
    "my-mcp-server": "dist/index.js"
  },
  "files": ["dist"],
  "scripts": {
    "build": "tsc",
    "prepublishOnly": "npm run build"
  },
  "keywords": ["mcp", "ai", "claude"],
  "license": "MIT"
}

# 빌드
npm run build

# dist/index.js 파일 상단에 shebang 추가
# #!/usr/bin/env node

# npm 로그인
npm login

# 배포
npm publish

# 스코프 패키지의 경우 public 접근 설정
npm publish --access public

# 사용자는 이렇게 실행할 수 있습니다
npx @myorg/mcp-server

FROM python:3.12-slim

WORKDIR /app

# 의존성 먼저 설치 (캐시 활용)
COPY pyproject.toml .
RUN pip install --no-cache-dir .

# 소스 코드 복사
COPY src/ src/

# MCP 서버는 stdio 통신이므로
# ENTRYPOINT로 서버 명령 설정
ENTRYPOINT ["python", "-m", "my_mcp_server.server"]

# 이미지 빌드
docker build -t my-mcp-server .

# Claude Desktop 설정에서 Docker로 실행
# claude_desktop_config.json:
# {
#   "mcpServers": {
#     "my-server": {
#       "command": "docker",
#       "args": ["run", "-i", "--rm", "my-mcp-server"]
#     }
#   }
# }

{
  "name": "my-mcp-server",
  "description": "유용한 도구를 제공하는 MCP 서버",
  "version": "1.0.0",
  "runtime": "python",
  "entrypoint": "my_mcp_server.server",
  "capabilities": {
    "tools": true,
    "resources": true,
    "prompts": true
  },
  "author": "개발자",
  "license": "MIT",
  "repository": "https://github.com/user/my-mcp-server"
}

# Smithery CLI 설치
npx @smithery/cli login

# 서버 등록
npx @smithery/cli publish

# 사용자가 Smithery에서 설치
npx @smithery/cli install my-mcp-server

# 기본 실행
npx @modelcontextprotocol/inspector python -m my_mcp_server.server

# 환경 변수 전달
DEBUG=true LOG_LEVEL=DEBUG \
  npx @modelcontextprotocol/inspector python -m my_mcp_server.server

# Inspector에서 확인할 항목:
# 1. Connection 탭: 서버 연결 상태, 초기화 메시지
# 2. Tools 탭:
#    - 등록된 도구 목록과 inputSchema 확인
#    - 각 도구를 직접 호출하고 결과 확인
#    - 잘못된 인자를 전달하여 에러 처리 검증
# 3. Resources 탭:
#    - 리소스 URI 목록 확인
#    - 각 리소스의 내용을 읽어보기
# 4. Prompts 탭:
#    - 프롬프트 템플릿 목록 확인
#    - 인자를 전달하여 렌더링 결과 미리보기

import logging
import sys
import os

def configure_logging():
    """환경에 따른 로깅 설정"""
    level = os.environ.get("LOG_LEVEL", "INFO").upper()

    # 루트 로거 설정 (stderr로 출력)
    logging.basicConfig(
        level=getattr(logging, level),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        stream=sys.stderr,  # 반드시 stderr!
    )

    # 개발 모드: 상세 로깅
    if os.environ.get("DEBUG") == "true":
        logging.getLogger("my-mcp-server").setLevel(logging.DEBUG)
        logging.getLogger("mcp").setLevel(logging.DEBUG)

    # 프로덕션 모드: 경고 이상만
    if os.environ.get("ENV") == "production":
        logging.getLogger().setLevel(logging.WARNING)

# 도구에서 로깅 활용
logger = logging.getLogger("my-mcp-server.tools")

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    logger.info(f"도구 호출: {name}")
    logger.debug(f"인자: {arguments}")

    try:
        result = await _execute_tool(name, arguments)
        logger.info(f"도구 {name} 실행 성공")
        logger.debug(f"결과: {result}")
        return result
    except Exception as e:
        logger.error(f"도구 {name} 실행 실패: {e}", exc_info=True)
        raise

import logging
import json
import sys

# MCP 내부 로거를 DEBUG로 설정하면
# 프로토콜 메시지를 볼 수 있습니다
logging.basicConfig(
    level=logging.DEBUG,
    stream=sys.stderr,
    format="%(name)s: %(message)s"
)

# MCP SDK의 내부 로거 활성화
logging.getLogger("mcp.server").setLevel(logging.DEBUG)
logging.getLogger("mcp.server.stdio").setLevel(logging.DEBUG)

# 출력 예시 (stderr):
# mcp.server: Received request: {"jsonrpc":"2.0","id":1,"method":"tools/list"}
# mcp.server: Sending response: {"jsonrpc":"2.0","id":1,"result":{...}}

# pip 사용
pip install fastmcp

# uv 사용 (권장)
uv add fastmcp

# 개발 의존성 포함
uv add fastmcp[dev]

#!/usr/bin/env python3
"""FastMCP 기본 서버 예제"""

from fastmcp import FastMCP

# 서버 인스턴스 생성 — 이름만 지정하면 끝
mcp = FastMCP("my-server")

# 도구 등록: 데코레이터 + 타입 힌트 = 자동 inputSchema 생성
@mcp.tool()
def add(a: int, b: int) -> int:
    """두 정수를 더합니다."""
    return a + b

@mcp.tool()
def greet(name: str, language: str = "ko") -> str:
    """이름에 맞는 인사를 반환합니다.

    Args:
        name: 인사할 사람 이름
        language: 인사 언어 (ko, en, ja)
    """
    greetings = {
        "ko": f"안녕하세요, {name}님!",
        "en": f"Hello, {name}!",
        "ja": f"こんにちは、{name}さん！",
    }
    return greetings.get(language, greetings["ko"])

# 리소스 등록
@mcp.resource("config://app")
def get_config() -> str:
    """애플리케이션 설정을 반환합니다."""
    import json
    return json.dumps({"version": "2.0", "debug": False})

# 프롬프트 등록
@mcp.prompt()
def review_code(code: str, lang: str = "python") -> str:
    """코드 리뷰 프롬프트를 생성합니다."""
    return f"""다음 {lang} 코드를 리뷰해주세요:

```{lang}
{code}
```

품질, 보안, 성능 관점에서 분석해주세요."""

# 서버 실행
if __name__ == "__main__":
    mcp.run()

from fastmcp import FastMCP, Context

mcp = FastMCP("context-demo")

@mcp.tool()
async def process_files(
    directory: str,
    ctx: Context
) -> str:
    """디렉토리의 파일들을 처리합니다."""
    import os

    files = os.listdir(directory)
    total = len(files)

    # 로깅
    await ctx.info(f"처리할 파일 수: {total}")

    results = []
    for i, filename in enumerate(files):
        # 진행 상황 보고
        await ctx.report_progress(i, total)

        filepath = os.path.join(directory, filename)
        if os.path.isfile(filepath):
            size = os.path.getsize(filepath)
            results.append(f"{filename}: {size} bytes")

    await ctx.report_progress(total, total)
    return "\n".join(results)

from pydantic import BaseModel, Field
from typing import Optional, List
from fastmcp import FastMCP

mcp = FastMCP("structured-server")

# Pydantic 모델로 복잡한 입력 정의
class SearchQuery(BaseModel):
    """파일 검색 쿼리"""
    directory: str = Field(description="검색 대상 디렉토리")
    pattern: str = Field(description="파일명 패턴 (glob)")
    max_results: int = Field(default=20, ge=1, le=100,
                             description="최대 결과 수")
    include_hidden: bool = Field(default=False,
                                description="숨김 파일 포함 여부")

class SearchResult(BaseModel):
    """검색 결과"""
    path: str
    size: int
    modified: float

@mcp.tool()
def search_files(query: SearchQuery) -> List[SearchResult]:
    """파일을 검색합니다. Pydantic 모델로 구조화된 입력을 받습니다."""
    import os, glob

    search_path = os.path.join(query.directory, query.pattern)
    files = glob.glob(search_path, recursive=True)

    results = []
    for f in files[:query.max_results]:
        if not query.include_hidden and os.path.basename(f).startswith("."):
            continue
        stat = os.stat(f)
        results.append(SearchResult(
            path=f, size=stat.st_size, modified=stat.st_mtime
        ))

    return results

@mcp.resource("db://users/{user_id}")
def get_user(user_id: int) -> str:
    """사용자 정보를 반환합니다."""
    import json
    # 실제로는 DB에서 조회
    user = {"id": user_id, "name": "사용자", "role": "admin"}
    return json.dumps(user, ensure_ascii=False)

@mcp.resource("log://app/{date}")
def get_logs(date: str) -> str:
    """특정 날짜의 로그를 반환합니다."""
    log_path = f"/var/log/app/{date}.log"
    try:
        with open(log_path) as f:
            return f.read()
    except FileNotFoundError:
        return f"로그 없음: {date}"

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "검색할 키워드 또는 구문"
    },
    "max_results": {
      "type": "integer",
      "description": "반환할 최대 결과 수",
      "default": 10,
      "minimum": 1,
      "maximum": 100
    },
    "format": {
      "type": "string",
      "description": "출력 형식",
      "enum": ["json", "text", "csv"]
    }
  },
  "required": ["query"]
}

// 중첩 객체, 배열, enum을 활용한 복잡한 스키마
const deployTool: Tool = {
  name: "deploy_service",
  description: "서비스를 지정된 환경에 배포합니다",
  inputSchema: {
    type: "object",
    properties: {
      service: {
        type: "string",
        description: "배포할 서비스 이름"
      },
      environment: {
        type: "string",
        enum: ["dev", "staging", "production"],
        description: "배포 대상 환경"
      },
      config: {
        type: "object",
        description: "배포 설정",
        properties: {
          replicas: {
            type: "integer",
            minimum: 1,
            maximum: 10,
            description: "인스턴스 수"
          },
          env_vars: {
            type: "object",
            additionalProperties: { type: "string" },
            description: "환경 변수 (키-값 쌍)"
          }
        }
      },
      tags: {
        type: "array",
        items: { type: "string" },
        description: "배포 태그 목록"
      },
      dry_run: {
        type: "boolean",
        default: false,
        description: "실제 배포 없이 검증만 수행"
      }
    },
    required: ["service", "environment"]
  }
};

#!/usr/bin/env python3
"""모듈화된 MCP 서버 구조"""

from fastmcp import FastMCP

# 메인 서버
app = FastMCP("modular-server")

# ===== 파일 관련 도구 모듈 =====
import os
from pathlib import Path

ALLOWED_BASE = Path.home() / "Projects"

def validate_path(path: str) -> Path:
    """경로를 검증하고 절대 경로를 반환합니다."""
    resolved = Path(path).resolve()
    if not str(resolved).startswith(str(ALLOWED_BASE)):
        raise ValueError("허용되지 않은 경로입니다")
    return resolved

@app.tool()
def read_project_file(path: str) -> str:
    """프로젝트 파일을 읽습니다."""
    validated = validate_path(path)
    return validated.read_text(encoding="utf-8")

@app.tool()
def list_project_files(
    directory: str,
    extension: str = "*"
) -> list[str]:
    """프로젝트 디렉토리 내 파일을 나열합니다."""
    validated = validate_path(directory)
    pattern = f"*.{extension}" if extension != "*" else "*"
    return [str(f) for f in validated.glob(pattern)]

# ===== Git 관련 도구 모듈 =====
import subprocess

@app.tool()
def git_status(repo_path: str) -> str:
    """Git 저장소 상태를 반환합니다."""
    validated = validate_path(repo_path)
    result = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=str(validated),
        capture_output=True, text=True
    )
    return result.stdout or "작업 디렉토리가 깨끗합니다"

@app.tool()
def git_log(
    repo_path: str,
    count: int = 10
) -> str:
    """최근 Git 커밋 이력을 반환합니다."""
    validated = validate_path(repo_path)
    result = subprocess.run(
        ["git", "log", f"--oneline", f"-{count}"],
        cwd=str(validated),
        capture_output=True, text=True
    )
    return result.stdout

if __name__ == "__main__":
    app.run()

# Python 서버 Inspector로 실행
npx @modelcontextprotocol/inspector python server.py

# FastMCP 서버
npx @modelcontextprotocol/inspector uv run server.py

# TypeScript 서버
npx @modelcontextprotocol/inspector node dist/index.js

# 환경 변수 전달
API_KEY=xxx npx @modelcontextprotocol/inspector python server.py

import pytest
import json
from unittest.mock import patch, MagicMock

# 서버 모듈에서 도구 함수를 직접 임포트하여 테스트
from server import read_project_file, list_project_files

class TestFileTools:
    """파일 관련 도구 테스트"""

    def test_read_file_success(self, tmp_path):
        """파일 읽기 성공 테스트"""
        test_file = tmp_path / "test.txt"
        test_file.write_text("테스트 내용")

        # 경로 검증을 임시로 우회
        with patch("server.validate_path",
                    return_value=test_file):
            result = read_project_file(str(test_file))

        assert result == "테스트 내용"

    def test_read_file_not_found(self):
        """존재하지 않는 파일 읽기 테스트"""
        with pytest.raises(FileNotFoundError):
            read_project_file("/nonexistent/file.txt")

    def test_path_traversal_blocked(self):
        """경로 탐색 공격 차단 테스트"""
        with pytest.raises(ValueError, match="허용되지 않은"):
            read_project_file("../../etc/passwd")

import { describe, it, expect } from "@jest/globals";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from
    "@modelcontextprotocol/sdk/client/stdio.js";

describe("MCP Server", () => {
  let client: Client;

  beforeAll(async () => {
    const transport = new StdioClientTransport({
      command: "node",
      args: ["dist/index.js"]
    });
    client = new Client(
      { name: "test-client", version: "1.0.0" },
      { capabilities: {} }
    );
    await client.connect(transport);
  });

  afterAll(async () => {
    await client.close();
  });

  it("도구 목록을 반환한다", async () => {
    const result = await client.listTools();
    expect(result.tools.length).toBeGreaterThan(0);
  });

  it("hello 도구를 호출한다", async () => {
    const result = await client.callTool({
      name: "hello",
      arguments: { name: "테스트" }
    });
    expect(result.content[0].text).toContain("테스트");
  });
});

import sys
import logging

# stderr로 로깅 설정 (stdout은 MCP 프로토콜 전용)
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(message)s",
    stream=sys.stderr  # 핵심: stderr로 출력
)
logger = logging.getLogger("mcp-server")

@mcp.tool()
def process_data(data: str) -> str:
    """데이터를 처리합니다."""
    logger.debug(f"입력 데이터: {data[:100]}...")

    try:
        result = do_processing(data)
        logger.info(f"처리 완료: {len(result)} bytes")
        return result
    except Exception as e:
        logger.error(f"처리 실패: {e}", exc_info=True)
        raise

# macOS - Claude Desktop MCP 로그 위치
tail -f ~/Library/Logs/Claude/mcp*.log

# 특정 서버 로그만 필터링
tail -f ~/Library/Logs/Claude/mcp*.log | grep "my-server"

# Windows
# %APPDATA%\Claude\logs\mcp*.log

#!/usr/bin/env python3
"""JSON-RPC 메시지를 로깅하는 디버그 래퍼"""

import sys
import json
import subprocess
import threading

def log_stream(name: str, stream, output):
    """스트림의 내용을 로깅하면서 전달합니다."""
    for line in stream:
        # stderr에 로깅
        print(
            f"[{name}] {line.rstrip()}",
            file=sys.stderr
        )
        # 원래 스트림으로 전달
        output.write(line)
        output.flush()

# 실제 서버를 서브프로세스로 실행
proc = subprocess.Popen(
    ["python", "server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# stdin 로깅 (클라이언트 -> 서버)
t_in = threading.Thread(
    target=log_stream,
    args=("REQ", sys.stdin, proc.stdin)
)
# stdout 로깅 (서버 -> 클라이언트)
t_out = threading.Thread(
    target=log_stream,
    args=("RES", proc.stdout, sys.stdout)
)

t_in.start()
t_out.start()
t_in.join()
t_out.join()