Инструменты разработчика

Server: my-first-server
Version: 1.0.0

Tools (2):
  - add: Сложить два числа
  - greet: Поприветствовать по имени

Resources (1):
  - info://about

Prompts (0):
  (none)

import logging

# Для разработки
logging.basicConfig(level=logging.DEBUG, stream=sys.stderr)

# Для production
logging.basicConfig(level=logging.WARNING, stream=sys.stderr)

import pytest
from mcp.server.fastmcp import FastMCP

# Создаём тестовый сервер
mcp = FastMCP("test-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Сложить два числа"""
    return a + b

@mcp.tool()
def divide(a: float, b: float) -> float:
    """Разделить a на b"""
    if b == 0:
        raise ValueError("Деление на ноль")
    return a / b

# Тесты
class TestTools:
    def test_add(self):
        result = add(5, 3)
        assert result == 8

    def test_add_negative(self):
        result = add(-5, 3)
        assert result == -2

    def test_divide(self):
        result = divide(10, 2)
        assert result == 5.0

    def test_divide_by_zero(self):
        with pytest.raises(ValueError, match="Деление на ноль"):
            divide(10, 0)

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { spawn, ChildProcess } from 'child_process';

describe('MCP Server Integration', () => {
  let serverProcess: ChildProcess;

  beforeAll(async () => {
    serverProcess = spawn('npm', ['run', 'start'], {
      stdio: ['pipe', 'pipe', 'pipe']
    });
    // Ждём запуска
    await new Promise(resolve => setTimeout(resolve, 1000));
  });

  afterAll(() => {
    serverProcess.kill();
  });

  it('should respond to initialize', async () => {
    const request = JSON.stringify({
      jsonrpc: '2.0',
      id: 1,
      method: 'initialize',
      params: {
        protocolVersion: '2024-11-05',
        capabilities: {},
        clientInfo: { name: 'test', version: '1.0.0' }
      }
    });

    serverProcess.stdin!.write(request + '\n');

    const response = await new Promise<string>(resolve => {
      serverProcess.stdout!.once('data', data => resolve(data.toString()));
    });

    const parsed = JSON.parse(response);
    expect(parsed.result).toBeDefined();
    expect(parsed.result.serverInfo).toBeDefined();
  });
});

// Запрос: initialize
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{
  "protocolVersion":"2024-11-05",
  "capabilities":{},
  "clientInfo":{"name":"test","version":"1.0"}
}}

// Ответ
{"jsonrpc":"2.0","id":1,"result":{
  "protocolVersion":"2024-11-05",
  "serverInfo":{"name":"my-server","version":"1.0.0"},
  "capabilities":{"tools":{}}
}}

// Запрос: tools/list
{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

// Ответ
{"jsonrpc":"2.0","id":2,"result":{"tools":[
  {"name":"add","description":"Сложить два числа","inputSchema":{...}}
]}}

// Запрос: tools/call
{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{
  "name":"add",
  "arguments":{"a":5,"b":3}
}}

// Ответ
{"jsonrpc":"2.0","id":3,"result":{
  "content":[{"type":"text","text":"8"}]
}}

#!/usr/bin/env python3
"""health_check.py - Проверка работоспособности MCP сервера"""

import subprocess
import json
import sys

def check_server(command: list[str]) -> bool:
    """Проверяет, что сервер отвечает на initialize"""

    init_request = json.dumps({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "health-check", "version": "1.0"}
        }
    }) + "\n"

    try:
        result = subprocess.run(
            command,
            input=init_request,
            capture_output=True,
            text=True,
            timeout=5
        )

        response = json.loads(result.stdout.strip())

        if "result" in response and "serverInfo" in response["result"]:
            print(f"✅ Server OK: {response['result']['serverInfo']}")
            return True
        else:
            print(f"❌ Invalid response: {response}")
            return False

    except subprocess.TimeoutExpired:
        print("❌ Server timeout")
        return False
    except json.JSONDecodeError as e:
        print(f"❌ Invalid JSON: {e}")
        return False
    except Exception as e:
        print(f"❌ Error: {e}")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python health_check.py <command>")
        print("Example: python health_check.py python server.py")
        sys.exit(1)

    command = sys.argv[1:]
    success = check_server(command)
    sys.exit(0 if success else 1)

FROM python:3.11-slim

WORKDIR /app

# Установка зависимостей для hot-reload
RUN pip install watchdog

COPY requirements.txt .
RUN pip install -r requirements.txt

# Копируем код
COPY . .

# Hot-reload с watchmedo
CMD ["watchmedo", "auto-restart", "--patterns=*.py", "--recursive", "--", "python", "server.py"]