MCP 서버 개발

# pip 사용
pip install mcp

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

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

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

# 권장: FastMCP 사용 (공식 고수준 API)
# from mcp.server.fastmcp import FastMCP
# mcp = FastMCP("my-mcp-server")
#
# 아래는 저수준 SDK 방식입니다 (고급 사용 사례용)
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 파일시스템 서버
"""

# 권장: FastMCP 사용 (공식 고수준 API)
# from mcp.server.fastmcp import FastMCP
# mcp = FastMCP("filesystem-server")
#
# 아래는 저수준 SDK 방식입니다 (고급 사용 사례용)
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": "^1.29.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.29.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, es)
    """
    greetings = {
        "ko": f"안녕하세요, {name}님!",
        "en": f"Hello, {name}!",
        "es": f"Hola, {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()