Split View: MCP 서버 직접 만들기: 원리부터 구축까지 — Grafana, Mermaid, PPT, DB 연동 MCP 생태계 완전 해부

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "Seoul"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Seoul: 15 degrees, partly cloudy"
      }
    ]
  }
}

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params: city is required"
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "roots": {
        "listChanged": true
      },
      "sampling": {}
    },
    "clientInfo": {
      "name": "Claude Desktop",
      "version": "1.5.0"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": {
        "listChanged": true
      },
      "resources": {
        "subscribe": true,
        "listChanged": true
      },
      "prompts": {
        "listChanged": true
      }
    },
    "serverInfo": {
      "name": "Weather MCP Server",
      "version": "1.0.0"
    }
  }
}

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

Client                    Server Process
  |                           |
  |--- spawn process -------->|
  |                           |
  |--- stdin: JSON-RPC ------>|
  |<-- stdout: JSON-RPC -----|
  |                           |
  |--- stdin: JSON-RPC ------>|
  |<-- stdout: JSON-RPC -----|
  |                           |
  |--- kill process --------->|

Client                    HTTP Server
  |                           |
  |--- POST /mcp ----------->|
  |    (initialize)           |
  |<-- SSE: result -----------|
  |<-- Mcp-Session-Id --------|
  |                           |
  |--- POST /mcp ----------->|
  |    (tools/call)           |
  |<-- SSE: result -----------|
  |                           |
  |--- DELETE /mcp ---------->|
  |    (session close)        |

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/read",
  "params": {
    "uri": "file:///project/schema.sql"
  }
}

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "contents": [
      {
        "uri": "file:///project/schema.sql",
        "mimeType": "text/plain",
        "text": "CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(100));"
      }
    ]
  }
}

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/list"
}

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "tools": [
      {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "inputSchema": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "City name"
            }
          },
          "required": ["city"]
        }
      }
    ]
  }
}

{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "prompts/get",
  "params": {
    "name": "code_review",
    "arguments": {
      "language": "python"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 4,
  "result": {
    "description": "Code review prompt for Python",
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "Please review the following Python code for best practices, bugs, and performance issues."
        }
      }
    ]
  }
}

{
  "capabilities": {
    "tools": {
      "listChanged": true
    }
  }
}

{
  "capabilities": {
    "tools": { "listChanged": true },
    "resources": { "subscribe": true, "listChanged": true },
    "prompts": { "listChanged": true },
    "logging": {}
  }
}

{
  "capabilities": {
    "roots": { "listChanged": true },
    "sampling": {}
  }
}

{
  "jsonrpc": "2.0",
  "id": 10,
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "Summarize this error log: Connection timeout after 30s..."
        }
      }
    ],
    "maxTokens": 500,
    "modelPreferences": {
      "hints": [{ "name": "claude-sonnet-4-20250514" }]
    }
  }
}

pip install fastmcp

from fastmcp import FastMCP

# MCP 서버 인스턴스 생성
mcp = FastMCP("Weather Server")


@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: Name of the city (e.g., Seoul, Tokyo, New York)
    """
    # 실제로는 날씨 API를 호출하겠지만 여기서는 간단히 시뮬레이션
    weather_data = {
        "Seoul": "15 degrees, partly cloudy",
        "Tokyo": "18 degrees, sunny",
        "New York": "10 degrees, rainy",
        "London": "8 degrees, foggy",
    }
    result = weather_data.get(city, f"Weather data not available for {city}")
    return f"Weather in {city}: {result}"


@mcp.tool()
def get_forecast(city: str, days: int = 3) -> str:
    """Get weather forecast for a city.

    Args:
        city: Name of the city
        days: Number of days to forecast (1-7)
    """
    if days < 1 or days > 7:
        return "Days must be between 1 and 7"
    return f"{days}-day forecast for {city}: Mostly sunny with temperatures 10-20C"


@mcp.resource("weather://cities")
def list_cities() -> str:
    """List all supported cities."""
    cities = ["Seoul", "Tokyo", "New York", "London", "Paris"]
    return "\n".join(cities)


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

# 직접 실행 (stdio 모드)
python weather_server.py

# 또는 FastMCP CLI로 실행
fastmcp run weather_server.py

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

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    }
  }
}

import sqlite3
import json
from fastmcp import FastMCP

mcp = FastMCP("Database Server")

DB_PATH = "myapp.db"


def get_connection():
    """Get database connection."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


@mcp.resource("db://schema")
def get_schema() -> str:
    """Get the database schema information."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT sql FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    tables = cursor.fetchall()
    conn.close()
    schema_lines = []
    for table in tables:
        if table[0]:
            schema_lines.append(table[0])
    return "\n\n".join(schema_lines)


@mcp.resource("db://tables")
def list_tables() -> str:
    """List all tables in the database."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    tables = [row[0] for row in cursor.fetchall()]
    conn.close()
    return json.dumps(tables, indent=2)


@mcp.tool()
def query_database(sql: str) -> str:
    """Execute a read-only SQL query and return results.

    Args:
        sql: SQL SELECT query to execute
    """
    # 보안: SELECT만 허용
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are allowed for safety"

    conn = get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(sql)
        rows = cursor.fetchall()
        if not rows:
            return "No results found"

        columns = [description[0] for description in cursor.description]
        results = []
        for row in rows:
            results.append(dict(zip(columns, row)))

        return json.dumps(results, indent=2, ensure_ascii=False)
    except Exception as e:
        return f"Query error: {str(e)}"
    finally:
        conn.close()


@mcp.tool()
def insert_record(table: str, data: str) -> str:
    """Insert a record into a table.

    Args:
        table: Table name
        data: JSON string of column-value pairs
    """
    try:
        record = json.loads(data)
    except json.JSONDecodeError:
        return "Error: Invalid JSON data"

    columns = ", ".join(record.keys())
    placeholders = ", ".join(["?" for _ in record])
    values = list(record.values())

    conn = get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(
            f"INSERT INTO {table} ({columns}) VALUES ({placeholders})",
            values,
        )
        conn.commit()
        return f"Successfully inserted record with id {cursor.lastrowid}"
    except Exception as e:
        return f"Insert error: {str(e)}"
    finally:
        conn.close()


@mcp.tool()
def update_record(table: str, record_id: int, data: str) -> str:
    """Update a record in a table.

    Args:
        table: Table name
        record_id: ID of the record to update
        data: JSON string of column-value pairs to update
    """
    try:
        updates = json.loads(data)
    except json.JSONDecodeError:
        return "Error: Invalid JSON data"

    set_clause = ", ".join([f"{k} = ?" for k in updates.keys()])
    values = list(updates.values()) + [record_id]

    conn = get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(
            f"UPDATE {table} SET {set_clause} WHERE id = ?", values
        )
        conn.commit()
        return f"Successfully updated {cursor.rowcount} record(s)"
    except Exception as e:
        return f"Update error: {str(e)}"
    finally:
        conn.close()


@mcp.tool()
def delete_record(table: str, record_id: int) -> str:
    """Delete a record from a table.

    Args:
        table: Table name
        record_id: ID of the record to delete
    """
    conn = get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(f"DELETE FROM {table} WHERE id = ?", [record_id])
        conn.commit()
        return f"Successfully deleted {cursor.rowcount} record(s)"
    except Exception as e:
        return f"Delete error: {str(e)}"
    finally:
        conn.close()


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

import os
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("Filesystem Server")

# 보안: 허용된 디렉토리만 접근 가능
ALLOWED_DIRS = [
    Path("/Users/dev/projects"),
    Path("/Users/dev/documents"),
]


def is_path_allowed(path: str) -> bool:
    """Check if the path is within allowed directories."""
    resolved = Path(path).resolve()
    return any(
        resolved == allowed or allowed in resolved.parents
        for allowed in ALLOWED_DIRS
    )


@mcp.tool()
def read_file(path: str) -> str:
    """Read contents of a file.

    Args:
        path: Absolute file path
    """
    if not is_path_allowed(path):
        return "Error: Access denied. Path is outside allowed directories."

    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found: {path}"
    except UnicodeDecodeError:
        return "Error: File is not a text file"


@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write content to a file.

    Args:
        path: Absolute file path
        content: Content to write
    """
    if not is_path_allowed(path):
        return "Error: Access denied. Path is outside allowed directories."

    try:
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully wrote to {path}"
    except Exception as e:
        return f"Error writing file: {str(e)}"


@mcp.tool()
def list_directory(path: str) -> str:
    """List contents of a directory.

    Args:
        path: Absolute directory path
    """
    if not is_path_allowed(path):
        return "Error: Access denied. Path is outside allowed directories."

    try:
        entries = []
        for entry in sorted(Path(path).iterdir()):
            entry_type = "dir" if entry.is_dir() else "file"
            size = entry.stat().st_size if entry.is_file() else 0
            entries.append(f"[{entry_type}] {entry.name} ({size} bytes)")
        return "\n".join(entries)
    except FileNotFoundError:
        return f"Error: Directory not found: {path}"


@mcp.tool()
def search_files(directory: str, pattern: str) -> str:
    """Search for files matching a glob pattern.

    Args:
        directory: Directory to search in
        pattern: Glob pattern (e.g., '*.py', '**/*.md')
    """
    if not is_path_allowed(directory):
        return "Error: Access denied."

    matches = list(Path(directory).glob(pattern))
    if not matches:
        return "No files found matching the pattern"
    return "\n".join(str(m) for m in matches[:50])


@mcp.resource("fs://allowed-dirs")
def get_allowed_dirs() -> str:
    """List allowed directories."""
    return "\n".join(str(d) for d in ALLOWED_DIRS)


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

npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node

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

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'

// 서버 인스턴스 생성
const server = new McpServer({
  name: 'My MCP Server',
  version: '1.0.0',
})

// Tool 등록
server.tool(
  'hello',
  'Say hello to someone',
  {
    name: z.string().describe('Name of the person'),
  },
  async ({ name }) => {
    return {
      content: [
        {
          type: 'text',
          text: `Hello, ${name}! Welcome to MCP.`,
        },
      ],
    }
  }
)

// Resource 등록
server.resource('info', 'server://info', async (uri) => {
  return {
    contents: [
      {
        uri: uri.href,
        mimeType: 'text/plain',
        text: 'This is My MCP Server v1.0.0',
      },
    ],
  }
})

// Transport 연결 및 실행
async function main() {
  const transport = new StdioServerTransport()
  await server.connect(transport)
  console.error('MCP Server running on stdio')
}

main().catch(console.error)

npx tsc
node dist/index.js

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'

const server = new McpServer({
  name: 'GitHub Issue Server',
  version: '1.0.0',
})

const GITHUB_TOKEN = process.env.GITHUB_TOKEN
const BASE_URL = 'https://api.github.com'

// 공통 헤더
function getHeaders() {
  return {
    Authorization: `Bearer ${GITHUB_TOKEN}`,
    Accept: 'application/vnd.github.v3+json',
    'Content-Type': 'application/json',
    'User-Agent': 'MCP-GitHub-Server',
  }
}

// Tool: 이슈 목록 조회
server.tool(
  'list_issues',
  'List issues in a GitHub repository',
  {
    owner: z.string().describe('Repository owner'),
    repo: z.string().describe('Repository name'),
    state: z.enum(['open', 'closed', 'all']).default('open').describe('Issue state filter'),
    labels: z.string().optional().describe('Comma-separated label names'),
    per_page: z.number().min(1).max(100).default(10).describe('Results per page'),
  },
  async ({ owner, repo, state, labels, per_page }) => {
    const params = new URLSearchParams({
      state,
      per_page: String(per_page),
    })
    if (labels) params.set('labels', labels)

    const response = await fetch(`${BASE_URL}/repos/${owner}/${repo}/issues?${params}`, {
      headers: getHeaders(),
    })

    if (!response.ok) {
      return {
        content: [
          {
            type: 'text' as const,
            text: `Error: ${response.status} ${response.statusText}`,
          },
        ],
      }
    }

    const issues = await response.json()
    const formatted = issues.map(
      (issue: Record<string, unknown>) =>
        `#${issue.number} [${issue.state}] ${issue.title}\n  Labels: ${
          (issue.labels as Array<Record<string, string>>).map((l) => l.name).join(', ') || 'none'
        }\n  Created: ${issue.created_at}`
    )

    return {
      content: [
        {
          type: 'text' as const,
          text: formatted.join('\n\n') || 'No issues found',
        },
      ],
    }
  }
)

// Tool: 이슈 생성
server.tool(
  'create_issue',
  'Create a new issue in a GitHub repository',
  {
    owner: z.string().describe('Repository owner'),
    repo: z.string().describe('Repository name'),
    title: z.string().describe('Issue title'),
    body: z.string().optional().describe('Issue body (Markdown)'),
    labels: z.array(z.string()).optional().describe('Labels to add'),
    assignees: z.array(z.string()).optional().describe('Usernames to assign'),
  },
  async ({ owner, repo, title, body, labels, assignees }) => {
    const response = await fetch(`${BASE_URL}/repos/${owner}/${repo}/issues`, {
      method: 'POST',
      headers: getHeaders(),
      body: JSON.stringify({ title, body, labels, assignees }),
    })

    if (!response.ok) {
      const error = await response.text()
      return {
        content: [
          {
            type: 'text' as const,
            text: `Error creating issue: ${response.status} - ${error}`,
          },
        ],
      }
    }

    const issue = await response.json()
    return {
      content: [
        {
          type: 'text' as const,
          text: `Created issue #${issue.number}: ${issue.title}\nURL: ${issue.html_url}`,
        },
      ],
    }
  }
)

// Tool: 이슈 닫기
server.tool(
  'close_issue',
  'Close an issue in a GitHub repository',
  {
    owner: z.string().describe('Repository owner'),
    repo: z.string().describe('Repository name'),
    issue_number: z.number().describe('Issue number'),
    comment: z.string().optional().describe('Optional closing comment'),
  },
  async ({ owner, repo, issue_number, comment }) => {
    // 코멘트가 있으면 먼저 추가
    if (comment) {
      await fetch(`${BASE_URL}/repos/${owner}/${repo}/issues/${issue_number}/comments`, {
        method: 'POST',
        headers: getHeaders(),
        body: JSON.stringify({ body: comment }),
      })
    }

    const response = await fetch(`${BASE_URL}/repos/${owner}/${repo}/issues/${issue_number}`, {
      method: 'PATCH',
      headers: getHeaders(),
      body: JSON.stringify({ state: 'closed' }),
    })

    if (!response.ok) {
      return {
        content: [
          {
            type: 'text' as const,
            text: `Error closing issue: ${response.status}`,
          },
        ],
      }
    }

    return {
      content: [
        {
          type: 'text' as const,
          text: `Closed issue #${issue_number} successfully`,
        },
      ],
    }
  }
)

// Resource: 리포지토리 정보
server.resource('repo-info', 'github://repo-info', async (uri) => {
  return {
    contents: [
      {
        uri: uri.href,
        mimeType: 'text/plain',
        text: 'GitHub Issue MCP Server - Manage issues via AI',
      },
    ],
  }
})

// Prompt: 이슈 트리아지
server.prompt(
  'triage_issues',
  'Generate a prompt for triaging open issues',
  { owner: z.string(), repo: z.string() },
  ({ owner, repo }) => ({
    messages: [
      {
        role: 'user' as const,
        content: {
          type: 'text' as const,
          text: `Please triage the open issues in ${owner}/${repo}. For each issue:
1. Assess priority (critical/high/medium/low)
2. Suggest appropriate labels
3. Recommend if it should be assigned to someone
4. Provide a brief action plan`,
        },
      },
    ],
  })
)

async function main() {
  const transport = new StdioServerTransport()
  await server.connect(transport)
  console.error('GitHub Issue MCP Server running')
}

main().catch(console.error)

{
  "name": "github-issue-mcp",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.12.0",
    "zod": "^3.24.0"
  },
  "devDependencies": {
    "typescript": "^5.7.0",
    "@types/node": "^22.0.0"
  }
}

{
  "mcpServers": {
    "github-issues": {
      "command": "node",
      "args": ["/path/to/github-issue-mcp/dist/index.js"],
      "env": {
        "GITHUB_TOKEN": "ghp_your_token_here"
      }
    }
  }
}

# npx로 바로 실행
npx @modelcontextprotocol/inspector

# 특정 서버를 지정하여 실행
npx @modelcontextprotocol/inspector python weather_server.py

# FastMCP의 dev 모드 (Inspector 자동 연결)
fastmcp dev weather_server.py

# 설정 파일 열기
code ~/Library/Application\ Support/Claude/claude_desktop_config.json

%APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/absolute/path/to/weather_server.py"]
    },
    "database": {
      "command": "python",
      "args": ["/absolute/path/to/db_server.py"],
      "env": {
        "DB_PATH": "/data/myapp.db"
      }
    },
    "github": {
      "command": "node",
      "args": ["/absolute/path/to/github-mcp/dist/index.js"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxx"
      }
    }
  }
}

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["./tools/my_mcp_server.py"]
    }
  }
}

claude mcp add my-server python ./tools/my_mcp_server.py

import sys
import logging

# stderr로 로그 출력 설정
logging.basicConfig(
    level=logging.DEBUG,
    stream=sys.stderr,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

logger = logging.getLogger("my-mcp-server")

@mcp.tool()
def my_tool(param: str) -> str:
    logger.debug(f"Tool called with param: {param}")
    try:
        result = do_something(param)
        logger.info(f"Tool succeeded: {result}")
        return result
    except Exception as e:
        logger.error(f"Tool failed: {e}")
        return f"Error: {str(e)}"

// stderr로 로그 출력
function log(level: string, message: string) {
  console.error(`[${new Date().toISOString()}] [${level}] ${message}`)
}

server.tool('my_tool', 'Description', { param: z.string() }, async ({ param }) => {
  log('DEBUG', `Tool called with: ${param}`)
  try {
    const result = await doSomething(param)
    log('INFO', `Success: ${result}`)
    return { content: [{ type: 'text', text: result }] }
  } catch (error) {
    log('ERROR', `Failed: ${error}`)
    return { content: [{ type: 'text', text: `Error: ${error}` }] }
  }
})

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("Robust Server")

@mcp.tool()
def safe_operation(param: str) -> str:
    """A tool with proper error handling."""
    # 입력 검증
    if not param or len(param) > 1000:
        raise ToolError("Parameter must be 1-1000 characters")

    try:
        result = external_api_call(param)
        return result
    except ConnectionError:
        raise ToolError("Failed to connect to external service. Please try again.")
    except TimeoutError:
        raise ToolError("Request timed out. The service may be temporarily unavailable.")
    except Exception as e:
        # 예상치 못한 에러는 상세 정보를 숨기고 일반 메시지만 반환
        logging.error(f"Unexpected error: {e}", exc_info=True)
        raise ToolError("An unexpected error occurred. Check server logs for details.")

# 프로젝트 생성
npm create cloudflare@latest -- my-mcp-server
cd my-mcp-server
npm install @modelcontextprotocol/sdk

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'

export default {
  async fetch(request: Request): Promise<Response> {
    const server = new McpServer({
      name: 'Cloudflare MCP Server',
      version: '1.0.0',
    })

    // Tools 등록
    server.tool('hello', 'Say hello', { name: z.string() }, async ({ name }) => ({
      content: [{ type: 'text', text: `Hello ${name} from the edge!` }],
    }))

    // Streamable HTTP transport 처리
    // 실제로는 MCP SDK의 HTTP transport handler를 사용
    return new Response('MCP Server Running', { status: 200 })
  },
}

# 배포
npx wrangler deploy

FROM node:22-slim

WORKDIR /app

COPY package*.json ./
RUN npm ci --production

COPY dist/ ./dist/

# MCP 서버는 stdio로 통신
CMD ["node", "dist/index.js"]

docker build -t my-mcp-server .
docker run -i my-mcp-server

{
  "mcpServers": {
    "my-server": {
      "command": "docker",
      "args": ["run", "-i", "--rm", "my-mcp-server"]
    }
  }
}

{
  "name": "@myorg/mcp-weather",
  "version": "1.0.0",
  "bin": {
    "mcp-weather": "./dist/index.js"
  }
}

npm publish --access public

{
  "mcpServers": {
    "weather": {
      "command": "npx",
      "args": ["-y", "@myorg/mcp-weather"]
    }
  }
}

pip install build twine
python -m build
twine upload dist/*

{
  "mcpServers": {
    "weather": {
      "command": "uvx",
      "args": ["mcp-weather"]
    }
  }
}

# smithery.yaml
name: my-weather-mcp
description: Weather MCP server for AI assistants
icon: cloud-sun
startCommand:
  type: stdio
  configSchema:
    type: object
    properties:
      API_KEY:
        type: string
        description: Weather API key
    required:
      - API_KEY
  commandFunction:
    - node
    - dist/index.js

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'

const server = new McpServer({
  name: 'Authenticated Server',
  version: '1.0.0',
})

// OAuth 2.0 미들웨어 (Streamable HTTP transport)
async function validateToken(token: string): Promise<boolean> {
  // OAuth provider에 토큰 검증 요청
  const response = await fetch('https://auth.example.com/introspect', {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
    body: `token=${token}`,
  })
  const data = await response.json()
  return data.active === true
}

// HTTP handler에서 인증 확인
async function handleRequest(req: Request): Promise<Response> {
  const authHeader = req.headers.get('Authorization')
  if (!authHeader || !authHeader.startsWith('Bearer ')) {
    return new Response('Unauthorized', { status: 401 })
  }

  const token = authHeader.slice(7)
  if (!(await validateToken(token))) {
    return new Response('Invalid token', { status: 403 })
  }

  // MCP 요청 처리 계속...
  return new Response('OK')
}

사용자: "지난 24시간 동안 CPU 사용률이 80%를 넘은 서버를 찾아줘"
Claude: [Grafana MCP의 query_metrics 도구 호출]
        → PromQL: avg by (instance)(cpu_usage_percent > 80)
        → 결과: server-03 (평균 87%), server-07 (평균 92%)

사용자: "현재 프로젝트의 아키텍처를 시퀀스 다이어그램으로 그려줘"
Claude: [Mermaid MCP의 create_diagram 도구 호출]
        → sequenceDiagram 자동 생성
        → SVG/PNG 파일로 저장

from fastmcp import FastMCP
import glob
import yaml

mcp = FastMCP("Blog Manager")

BLOG_DIR = "/path/to/blog/data"

@mcp.tool()
def list_posts(tag: str = "") -> str:
    """List all blog posts, optionally filtered by tag."""
    posts = glob.glob(f"{BLOG_DIR}/**/*.mdx", recursive=True)
    results = []
    for post_path in sorted(posts, reverse=True):
        with open(post_path) as f:
            content = f.read()
            # frontmatter 파싱
            if content.startswith("---"):
                fm_end = content.index("---", 3)
                frontmatter = yaml.safe_load(content[3:fm_end])
                if tag and tag not in frontmatter.get("tags", []):
                    continue
                results.append(
                    f"- {frontmatter['title']} ({frontmatter['date']})"
                )
    return "\n".join(results[:20])

@mcp.tool()
def create_post(title: str, tags: str, content: str) -> str:
    """Create a new blog post."""
    # 구현...
    return "Post created successfully"

from fastmcp import FastMCP
import httpx

mcp = FastMCP("Smart Home")

HA_URL = "http://homeassistant.local:8123"
HA_TOKEN = "your_long_lived_token"

@mcp.tool()
def control_light(entity_id: str, action: str) -> str:
    """Control a smart light.

    Args:
        entity_id: Home Assistant entity ID (e.g., light.living_room)
        action: 'on' or 'off'
    """
    service = "turn_on" if action == "on" else "turn_off"
    response = httpx.post(
        f"{HA_URL}/api/services/light/{service}",
        headers={"Authorization": f"Bearer {HA_TOKEN}"},
        json={"entity_id": entity_id},
    )
    return f"Light {entity_id}: {action} (status: {response.status_code})"

@mcp.tool()
def get_temperature(entity_id: str) -> str:
    """Get temperature from a sensor."""
    response = httpx.get(
        f"{HA_URL}/api/states/{entity_id}",
        headers={"Authorization": f"Bearer {HA_TOKEN}"},
    )
    data = response.json()
    return f"Temperature: {data['state']}{data['attributes'].get('unit_of_measurement', '')}"

from pydantic import BaseModel, Field, validator

class QueryInput(BaseModel):
    sql: str = Field(..., max_length=5000)
    timeout: int = Field(default=30, ge=1, le=300)

    @validator("sql")
    def validate_sql(cls, v):
        forbidden = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
        upper = v.upper()
        for keyword in forbidden:
            if keyword in upper:
                raise ValueError(f"Forbidden SQL keyword: {keyword}")
        return v

const querySchema = z.object({
  sql: z
    .string()
    .max(5000)
    .refine(
      (sql) => {
        const forbidden = ['DROP', 'DELETE', 'TRUNCATE', 'ALTER']
        const upper = sql.toUpperCase()
        return !forbidden.some((kw) => upper.includes(kw))
      },
      { message: 'Forbidden SQL keyword detected' }
    ),
  timeout: z.number().min(1).max(300).default(30),
})

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int = 10, window: int = 60):
        self.max_calls = max_calls
        self.window = window
        self.calls = defaultdict(list)

    def check(self, key: str) -> bool:
        now = time.time()
        self.calls[key] = [
            t for t in self.calls[key] if now - t < self.window
        ]
        if len(self.calls[key]) >= self.max_calls:
            return False
        self.calls[key].append(now)
        return True

limiter = RateLimiter(max_calls=20, window=60)

@mcp.tool()
def rate_limited_tool(param: str) -> str:
    if not limiter.check("default"):
        return "Rate limit exceeded. Please wait before trying again."
    return do_work(param)

import os

# 환경 변수에서 시크릿 로드 (하드코딩 금지)
DB_PASSWORD = os.environ.get("DB_PASSWORD")
API_KEY = os.environ.get("API_KEY")

if not DB_PASSWORD or not API_KEY:
    raise RuntimeError("Required environment variables not set")

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["server.py"],
      "env": {
        "DB_PASSWORD": "secret_from_keychain",
        "API_KEY": "sk-xxx"
      }
    }
  }
}

import json
import sys
from datetime import datetime

def audit_log(tool_name: str, params: dict, result: str, success: bool):
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "tool": tool_name,
        "params": params,
        "success": success,
        "result_length": len(result),
    }
    print(json.dumps(log_entry), file=sys.stderr)

# docker-compose.yml
services:
  mcp-server:
    build: .
    read_only: true
    security_opt:
      - no-new-privileges:true
    tmpfs:
      - /tmp:size=100M
    mem_limit: 512m
    cpus: '0.5'
    networks:
      - mcp-net
    volumes:
      - ./allowed-data:/data:ro

networks:
  mcp-net:
    driver: bridge
    internal: true # 외부 인터넷 접근 차단