Python MCP или как подключить свою LLM ко всему миру - Создаем MCP-сервер

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
 
# Создаём экземпляр сервера с уникальным именем
server = Server("my-first-server")
 
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__":
    asyncio.run(main())

@server.list_tools()
async def handle_list_tools():
    """Возвращаем список доступных инструментов"""
    return [
        Tool(
            name="get_weather",
            description="Получает текущую погоду для указанного города",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "Название города на английском"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Единицы измерения температуры",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    """Выполняем запрошенный инструмент"""
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        
        # Тут должен быть реальный API вызов, 
        # но для примера возвращаем заглушку
        temp = 22 if units == "celsius" else 72
        
        return [
            TextContent(
                type="text",
                text=f"Температура в {city}: {temp}°{units[0].upper()}"
            )
        ]
    
    raise ValueError(f"Неизвестный инструмент: {name}")

from mcp.types import Resource
 
@server.list_resources()
async def handle_list_resources():
    """Перечисляем доступные ресурсы"""
    return [
        Resource(
            uri="file:///docs/readme.md",
            name="README документ",
            description="Основная документация проекта",
            mimeType="text/markdown"
        ),
        Resource(
            uri="db://users/{user_id}",
            name="Профиль пользователя",
            description="Данные пользователя из базы по ID",
            mimeType="application/json"
        )
    ]

import aiofiles
 
@server.read_resource()
async def handle_read_resource(uri: str):
    """Читаем содержимое запрошенного ресурса"""
    if uri.startswith("file://"):
        # Извлекаем путь из URI
        file_path = uri.replace("file://", "")
        
        # Асинхронно читаем файл
        async with aiofiles.open(file_path, mode='r') as f:
            content = await f.read()
        
        return [
            TextContent(
                type="text",
                text=content
            )
        ]
    
    elif uri.startswith("db://users/"):
        # Парсим ID из URI
        user_id = uri.split("/")[-1]
        
        # Тут должен быть реальный запрос к БД
        user_data = {
            "id": user_id,
            "name": "Иван Иванов",
            "email": "ivan@example.com"
        }
        
        import json
        return [
            TextContent(
                type="text",
                text=json.dumps(user_data, ensure_ascii=False, indent=2)
            )
        ]
    
    raise ValueError(f"Неизвестная схема URI: {uri}")

import base64
from mcp.types import BlobContent
 
@server.read_resource()
async def handle_read_resource(uri: str):
    if uri.startswith("image://"):
        image_path = uri.replace("image://", "")
        
        async with aiofiles.open(image_path, mode='rb') as f:
            image_bytes = await f.read()
        
        return [
            BlobContent(
                type="blob",
                blob=base64.b64encode(image_bytes).decode('utf-8'),
                mimeType="image/png"
            )
        ]

from pathlib import Path
import aiofiles.os
 
@server.list_tools()
async def handle_list_tools():
    return [
        Tool(
            name="create_directory",
            description="Создаёт новую директорию по указанному пути",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Абсолютный или относительный путь к создаваемой папке"
                    },
                    "parents": {
                        "type": "boolean",
                        "description": "Создавать ли родительские директории если их нет",
                        "default": True
                    }
                },
                "required": ["path"]
            }
        )
    ]
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "create_directory":
        path = Path(arguments["path"])
        parents = arguments.get("parents", True)
        
        try:
            await aiofiles.os.makedirs(path, exist_ok=parents)
            return [
                TextContent(
                    type="text",
                    text=f"Директория создана: {path.absolute()}"
                )
            ]
        except FileExistsError:
            return [
                TextContent(
                    type="text",
                    text=f"Директория уже существует: {path}"
                )
            ]
        except PermissionError:
            raise ValueError(f"Недостаточно прав для создания {path}")

import aiohttp
from typing import Optional
 
class SlackClient:
    def __init__(self, token: str):
        self.token = token
        self.base_url = "https://slack.com/api"
    
    async def post_message(
        self, 
        channel: str, 
        text: str, 
        thread_ts: Optional[str] = None
    ) -> dict:
        """Отправляет сообщение в канал Slack"""
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat.postMessage",
                headers={"Authorization": f"Bearer {self.token}"},
                json={
                    "channel": channel,
                    "text": text,
                    "thread_ts": thread_ts
                }
            ) as response:
                return await response.json()
 
# Инициализируем клиент при старте сервера
slack = SlackClient(token="xoxb-your-token")
 
@server.list_tools()
async def handle_list_tools():
    return [
        Tool(
            name="send_slack_message",
            description="Отправляет сообщение в указанный канал Slack",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {
                        "type": "string",
                        "description": "ID или название канала (#general, C1234567)"
                    },
                    "message": {
                        "type": "string",
                        "description": "Текст отправляемого сообщения"
                    },
                    "thread_id": {
                        "type": "string",
                        "description": "ID треда для ответа (опционально)"
                    }
                },
                "required": ["channel", "message"]
            }
        )
    ]
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "send_slack_message":
        channel = arguments["channel"]
        message = arguments["message"]
        thread_id = arguments.get("thread_id")
        
        result = await slack.post_message(
            channel=channel,
            text=message,
            thread_ts=thread_id
        )
        
        if result.get("ok"):
            return [
                TextContent(
                    type="text",
                    text=f"Сообщение отправлено в {channel}"
                )
            ]
        else:
            error = result.get("error", "Неизвестная ошибка")
            raise ValueError(f"Slack API ошибка: {error}")

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_external_api(url: str) -> dict:
    """Вызов с автоматическими повторами при ошибках"""
    async with aiohttp.ClientSession() as session:
        async with asyncio.timeout(30):  # таймаут 30 секунд
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()

from typing import Callable, Any
from inspect import signature, Parameter
import functools
 
def mcp_tool(description: str):
    """Декоратор для автоматической регистрации MCP-инструмента"""
    def decorator(func: Callable) -> Callable:
        # Извлекаем сигнатуру функции
        sig = signature(func)
        
        # Строим JSON Schema из аннотаций типов
        properties = {}
        required = []
        
        for param_name, param in sig.parameters.items():
            if param_name in ('self', 'cls'):
                continue
                
            param_schema = {"type": "string"}  # дефолтный тип
            
            # Определяем тип из аннотации
            if param.annotation == int:
                param_schema["type"] = "integer"
            elif param.annotation == bool:
                param_schema["type"] = "boolean"
            elif param.annotation == float:
                param_schema["type"] = "number"
            
            # Добавляем описание из докстринга если есть
            if func.__doc__:
                # Парсим докстринг в поисках описания параметра
                for line in func.__doc__.split('\n'):
                    if f':param {param_name}:' in line:
                        param_schema["description"] = line.split(':', 2)[2].strip()
            
            properties[param_name] = param_schema
            
            # Параметры без дефолтного значения - обязательные
            if param.default == Parameter.empty:
                required.append(param_name)
        
        # Сохраняем метаданные в атрибутах функции
        func._mcp_tool_name = func.__name__
        func._mcp_tool_description = description
        func._mcp_tool_schema = {
            "type": "object",
            "properties": properties,
            "required": required
        }
        
        return func
    
    return decorator

@mcp_tool("Запускает Docker-контейнер с указанным образом")
async def start_container(
    image: str,
    name: str,
    port: int = 8080,
    detached: bool = True
) -> str:
    """
    Запускает новый Docker-контейнер
    
    :param image: Название Docker-образа для запуска
    :param name: Имя для создаваемого контейнера
    :param port: Порт для проброса на хост
    :param detached: Запустить в фоновом режиме
    """
    # Реализация запуска контейнера
    return f"Контейнер {name} запущен из образа {image}"

def collect_tools(module) -> list[Tool]:
    """Собирает все функции с декоратором @mcp_tool"""
    tools = []
    
    for name in dir(module):
        obj = getattr(module, name)
        if callable(obj) and hasattr(obj, '_mcp_tool_name'):
            tools.append(Tool(
                name=obj._mcp_tool_name,
                description=obj._mcp_tool_description,
                inputSchema=obj._mcp_tool_schema
            ))
    
    return tools
 
# При инициализации сервера
@server.list_tools()
async def handle_list_tools():
    return collect_tools(sys.modules[__name__])

def build_tool_map(module) -> dict[str, Callable]:
    """Строит словарь имя -> функция для всех инструментов"""
    tool_map = {}
    
    for name in dir(module):
        obj = getattr(module, name)
        if callable(obj) and hasattr(obj, '_mcp_tool_name'):
            tool_map[obj._mcp_tool_name] = obj
    
    return tool_map
 
TOOLS = build_tool_map(sys.modules[__name__])
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name not in TOOLS:
        raise ValueError(f"Неизвестный инструмент: {name}")
    
    tool_func = TOOLS[name]
    result = await tool_func(**arguments)
    
    return [TextContent(type="text", text=str(result))]

from mcp.types import Prompt, PromptArgument
 
@server.list_prompts()
async def handle_list_prompts():
    """Возвращаем список доступных промптов"""
    return [
        Prompt(
            name="analyze_code",
            description="Анализирует код на потенциальные проблемы",
            arguments=[
                PromptArgument(
                    name="language",
                    description="Язык программирования",
                    required=True
                ),
                PromptArgument(
                    name="focus",
                    description="Аспект для анализа: security, performance, style",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict):
    """Генерирует промпт с подставленными параметрами"""
    if name == "analyze_code":
        language = arguments["language"]
        focus = arguments.get("focus", "general")
        
        # Базовая инструкция
        base_instruction = f"""Ты - эксперт по {language}. 
Проанализируй предоставленный код и дай детальную обратную связь."""
        
        # Специфичные инструкции в зависимости от фокуса
        focus_instructions = {
            "security": f"""
Обрати особое внимание на:
SQL-инъекции и другие векторы атак
Небезопасную работу с пользовательским вводом
Утечки конфиденциальных данных
Уязвимости в зависимостях
""",
            "performance": f"""
Сфокусируйся на:
Неэффективных алгоритмах и структурах данных
Избыточных вычислениях и запросах к БД
Утечках памяти
Возможностях для кеширования
""",
            "style": f"""
Проверь соответствие:
Конвенциям именования для {language}
Стандартам форматирования кода
Принципам SOLID и чистой архитектуры
Наличию документации
"""
        }
        
        instruction = base_instruction + focus_instructions.get(focus, "")
        
        return [
            TextContent(
                type="text",
                text=instruction
            )
        ]

class DynamicPromptGenerator:
    def __init__(self, project_path: str):
        self.project_path = Path(project_path)
        self.style_guide = self._load_style_guide()
        self.common_issues = self._load_issue_database()
    
    def _load_style_guide(self) -> str:
        """Загружает корпоративный style guide"""
        style_file = self.project_path / ".style-guide.md"
        if style_file.exists():
            return style_file.read_text()
        return ""
    
    async def generate_review_prompt(
        self, 
        language: str,
        file_path: str
    ) -> str:
        """Создаёт контекстуальный промпт для ревью"""
        
        # Находим похожие файлы с хорошими примерами
        similar_files = await self._find_similar_patterns(file_path)
        
        # Собираем типичные проблемы для этого типа файлов
        relevant_issues = self._get_relevant_issues(language, file_path)
        
        prompt = f"""Ты проводишь code review для {file_path}.
 
СТАНДАРТЫ ПРОЕКТА:
{self.style_guide}
 
ТИПИЧНЫЕ ПРОБЛЕМЫ В ПОДОБНЫХ ФАЙЛАХ:
"""
        for issue in relevant_issues[:5]:  # топ-5 проблем
            prompt += f"- {issue['description']} (встречается в {issue['frequency']}% случаев)
"
        
        if similar_files:
            prompt += f"""
 
ПРИМЕРЫ ХОРОШЕГО КОДА ИЗ ПРОЕКТА:
"""
            for example_file in similar_files[:3]:
                snippet = await self._extract_relevant_snippet(example_file)
                prompt += f"""
Файл: {example_file}
{language}
{snippet}
"""
        
        return prompt
 
# Интегрируем в обработчик
generator = DynamicPromptGenerator("/path/to/project")
 
@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict):
    if name == "code_review":
        language = arguments["language"]
        file_path = arguments["file_path"]
        
        prompt_text = await generator.generate_review_prompt(
            language, 
            file_path
        )
        
        return [TextContent(type="text", text=prompt_text)]

class PromptComposer:
    def __init__(self):
        self.snippets = {
            "base_rules": "Следуй принципам чистого кода...",
            "security_focus": "Особое внимание к безопасности...",
            "performance_tips": "Оптимизируй критичные участки...",
        }
    
    def compose(self, base: str, additions: list[str]) -> str:
        """Собирает финальный промпт из компонентов"""
        parts = [self.snippets[base]]
        parts.extend(self.snippets[add] for add in additions)
        return "
 
".join(parts)
 
composer = PromptComposer()
 
@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict):
    if name == "review_endpoint":
        # Базовый промпт + специфичные дополнения
        focus_areas = arguments.get("focus", [])
        
        prompt = composer.compose(
            "base_rules",
            ["security_focus"] + focus_areas
        )
        
        return [TextContent(type="text", text=prompt)]

from typing import AsyncIterator
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> AsyncIterator:
    if name == "export_logs":
        start_date = arguments["start_date"]
        end_date = arguments["end_date"]
        
        # Открываем файл в streaming режиме
        async with aiofiles.open("app.log", mode='r') as log_file:
            buffer = []
            async for line in log_file:
                # Проверяем попадает ли строка в диапазон дат
                if is_in_date_range(line, start_date, end_date):
                    buffer.append(line)
                    
                    # Отправляем порцию каждые 100 строк
                    if len(buffer) >= 100:
                        yield [TextContent(
                            type="text",
                            text="".join(buffer)
                        )]
                        buffer.clear()
            
            # Отправляем остаток
            if buffer:
                yield [TextContent(
                    type="text", 
                    text="".join(buffer)
                )]

import asyncpg
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "query_analytics":
        query = arguments["sql"]
        
        conn = await asyncpg.connect("postgresql://...")
        
        # Используем cursor для streaming-чтения
        async with conn.transaction():
            cursor = await conn.cursor(query)
            
            batch = []
            async for record in cursor:
                batch.append(dict(record))
                
                if len(batch) >= 1000:
                    yield [TextContent(
                        type="text",
                        text=json.dumps(batch, ensure_ascii=False)
                    )]
                    batch.clear()
            
            if batch:
                yield [TextContent(
                    type="text",
                    text=json.dumps(batch, ensure_ascii=False)
                )]
        
        await conn.close()

try:
    # Streaming логика
    async for chunk in process_data():
        yield chunk
except Exception as e:
    # Отправляем сообщение об ошибке
    yield [TextContent(
        type="text",
        text=f"ERROR: {str(e)}"
    )]
    raise

from functools import wraps
from typing import Callable, Any
import asyncio
 
class AsyncLRUCache:
    def __init__(self, maxsize: int = 128):
        self.cache: dict[str, Any] = {}
        self.maxsize = maxsize
        self.lock = asyncio.Lock()
    
    def _make_key(self, args: tuple, kwargs: dict) -> str:
        """Создаём ключ из аргументов функции"""
        key_parts = [str(arg) for arg in args]
        key_parts.extend(f"{k}={v}" for k, v in sorted(kwargs.items()))
        return ":".join(key_parts)
    
    async def get(self, key: str) -> Any:
        """Получаем значение из кеша"""
        async with self.lock:
            return self.cache.get(key)
    
    async def set(self, key: str, value: Any) -> None:
        """Сохраняем в кеш с учётом лимита размера"""
        async with self.lock:
            if len(self.cache) >= self.maxsize:
                # Удаляем самую старую запись (простейшая стратегия)
                first_key = next(iter(self.cache))
                del self.cache[first_key]
            self.cache[key] = value
 
def async_lru_cache(maxsize: int = 128):
    """Декоратор для кеширования асинхронных функций"""
    cache = AsyncLRUCache(maxsize)
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = cache._make_key(args, kwargs)
            
            # Проверяем есть ли в кеше
            cached = await cache.get(cache_key)
            if cached is not None:
                return cached
            
            # Вычисляем и кешируем
            result = await func(*args, **kwargs)
            await cache.set(cache_key, result)
            return result
        
        return wrapper
    return decorator

@async_lru_cache(maxsize=256)
async def parse_python_file(file_path: str) -> dict:
    """Парсит Python файл и возвращает метаданные"""
    async with aiofiles.open(file_path, 'r') as f:
        content = await f.read()
    
    tree = ast.parse(content)
    # Извлекаем функции, классы, импорты
    functions = [node.name for node in ast.walk(tree) 
                 if isinstance(node, ast.FunctionDef)]
    classes = [node.name for node in ast.walk(tree) 
               if isinstance(node, ast.ClassDef)]
    
    return {
        "functions": functions,
        "classes": classes,
        "lines": len(content.split('\n'))
    }

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
 
class CacheInvalidator(FileSystemEventHandler):
    def __init__(self, cache: AsyncLRUCache):
        self.cache = cache
    
    def on_modified(self, event):
        """При изменении файла сбрасываем его из кеша"""
        if event.src_path.endswith('.py'):
            # Находим и удаляем все записи связанные с этим файлом
            asyncio.create_task(self._invalidate(event.src_path))
    
    async def _invalidate(self, file_path: str):
        async with self.cache.lock:
            keys_to_remove = [
                key for key in self.cache.cache.keys() 
                if file_path in key
            ]
            for key in keys_to_remove:
                del self.cache.cache[key]
 
# Запускаем наблюдатель за файловой системой
cache = AsyncLRUCache(maxsize=512)
observer = Observer()
observer.schedule(CacheInvalidator(cache), path="./src", recursive=True)
observer.start()

import aioredis
import json
import hashlib
 
class RedisCache:
    def __init__(self, redis_url: str = "redis://localhost"):
        self.redis = None
        self.redis_url = redis_url
    
    async def connect(self):
        """Устанавливаем соединение с Redis"""
        self.redis = await aioredis.from_url(self.redis_url)
    
    def _make_key(self, prefix: str, *args, **kwargs) -> str:
        """Создаём хеш от аргументов для ключа"""
        data = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
        hash_val = hashlib.sha256(data.encode()).hexdigest()[:16]
        return f"{prefix}:{hash_val}"
    
    async def get_or_compute(
        self, 
        key_prefix: str,
        compute_func: Callable,
        ttl: int = 3600,  # время жизни в секундах
        *args,
        **kwargs
    ) -> Any:
        """Получает из кеша или вычисляет если нет"""
        cache_key = self._make_key(key_prefix, *args, **kwargs)
        
        # Проверяем кеш
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Вычисляем
        result = await compute_func(*args, **kwargs)
        
        # Сохраняем с TTL
        await self.redis.setex(
            cache_key,
            ttl,
            json.dumps(result, ensure_ascii=False)
        )
        
        return result
 
# Использование
redis_cache = RedisCache()
await redis_cache.connect()
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    if name == "analyze_repository":
        repo_url = arguments["url"]
        
        result = await redis_cache.get_or_compute(
            "repo_analysis",
            analyze_repo_expensive,
            ttl=7200,  # кешируем на 2 часа
            repo_url
        )
        
        return [TextContent(type="text", text=json.dumps(result))]

class QueryBatcher:
    def __init__(self, batch_size: int = 100, flush_interval: float = 0.1):
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.pending: list[tuple] = []
        self.results: dict[str, asyncio.Future] = {}
        self._flush_task = None
    
    async def add(self, query_id: str, query: str) -> Any:
        """Добавляет запрос в батч"""
        future = asyncio.Future()
        self.results[query_id] = future
        self.pending.append((query_id, query))
        
        if len(self.pending) >= self.batch_size:
            await self._flush()
        elif not self._flush_task:
            # Автофлаш через интервал
            self._flush_task = asyncio.create_task(self._auto_flush())
        
        return await future
    
    async def _flush(self):
        """Выполняет накопленные запросы одним батчем"""
        if not self.pending:
            return
        
        batch = self.pending[:]
        self.pending.clear()
        
        # Выполняем все запросы одной транзакцией
        conn = await asyncpg.connect("postgresql://...")
        async with conn.transaction():
            for query_id, query in batch:
                try:
                    result = await conn.fetch(query)
                    self.results[query_id].set_result(result)
                except Exception as e:
                    self.results[query_id].set_exception(e)
        
        await conn.close()
        
        if self._flush_task:
            self._flush_task.cancel()
            self._flush_task = None
    
    async def _auto_flush(self):
        """Автоматический flush по таймеру"""
        await asyncio.sleep(self.flush_interval)
        await self._flush()

import time
import logging
from typing import Callable, Any
from functools import wraps
import json
 
logger = logging.getLogger("mcp_server")
 
class RequestMiddleware:
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "failed_requests": 0,
            "avg_duration": 0.0
        }
        self.request_history = []
    
    def log_request(self, func: Callable) -> Callable:
        """Декоратор для логирования запросов"""
        @wraps(func)
        async def wrapper(*args, **kwargs):
            request_id = self._generate_request_id()
            start_time = time.time()
            
            # Логируем входящий запрос
            logger.info(
                f"[{request_id}] Запрос: {func.__name__}",
                extra={
                    "args": str(args)[:200],  # ограничиваем размер
                    "kwargs": {k: str(v)[:100] for k, v in kwargs.items()}
                }
            )
            
            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time
                
                # Обновляем метрики
                self.metrics["total_requests"] += 1
                self._update_avg_duration(duration)
                
                # Логируем успех
                logger.info(
                    f"[{request_id}] Успех за {duration:.3f}с",
                    extra={"result_size": len(str(result))}
                )
                
                # Сохраняем в историю
                self._save_to_history(
                    request_id,
                    func.__name__,
                    args,
                    kwargs,
                    result,
                    duration,
                    success=True
                )
                
                return result
                
            except Exception as e:
                duration = time.time() - start_time
                self.metrics["failed_requests"] += 1
                
                # Логируем ошибку с полным трейсбеком
                logger.error(
                    f"[{request_id}] Ошибка: {str(e)}",
                    exc_info=True,
                    extra={"duration": duration}
                )
                
                self._save_to_history(
                    request_id,
                    func.__name__,
                    args,
                    kwargs,
                    str(e),
                    duration,
                    success=False
                )
                
                raise
        
        return wrapper
    
    def _generate_request_id(self) -> str:
        """Генерирует уникальный ID запроса"""
        import uuid
        return str(uuid.uuid4())[:8]
    
    def _update_avg_duration(self, duration: float):
        """Обновляет среднее время выполнения"""
        total = self.metrics["total_requests"]
        current_avg = self.metrics["avg_duration"]
        self.metrics["avg_duration"] = (
            (current_avg * (total - 1) + duration) / total
        )
    
    def _save_to_history(
        self,
        request_id: str,
        func_name: str,
        args: tuple,
        kwargs: dict,
        result: Any,
        duration: float,
        success: bool
    ):
        """Сохраняет запрос в историю"""
        entry = {
            "id": request_id,
            "function": func_name,
            "timestamp": time.time(),
            "duration": duration,
            "success": success
        }
        
        # Ограничиваем размер истории
        if len(self.request_history) >= 1000:
            self.request_history.pop(0)
        
        self.request_history.append(entry)
 
# Создаём глобальный middleware
middleware = RequestMiddleware()

@middleware.log_request
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    # Основная логика без изменений
    if name == "query_database":
        result = await execute_query(arguments["sql"])
        return [TextContent(type="text", text=json.dumps(result))]

from prometheus_client import Counter, Histogram, Gauge
 
# Определяем метрики
request_counter = Counter(
    'mcp_requests_total',
    'Общее количество запросов',
    ['function', 'status']
)
 
request_duration = Histogram(
    'mcp_request_duration_seconds',
    'Длительность выполнения запросов',
    ['function']
)
 
active_requests = Gauge(
    'mcp_active_requests',
    'Количество запросов в обработке'
)
 
class PrometheusMiddleware:
    def monitor(self, func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            active_requests.inc()
            
            with request_duration.labels(
                function=func.__name__
            ).time():
                try:
                    result = await func(*args, **kwargs)
                    request_counter.labels(
                        function=func.__name__,
                        status='success'
                    ).inc()
                    return result
                except Exception as e:
                    request_counter.labels(
                        function=func.__name__,
                        status='error'
                    ).inc()
                    raise
                finally:
                    active_requests.dec()
        
        return wrapper

import asyncio
from typing import Callable
 
class RequestHandler:
    def __init__(self, max_concurrent: int = 50):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.handlers = {}
    
    def register(self, method: str, handler: Callable):
        """Регистрируем обработчик для метода"""
        self.handlers[method] = handler
    
    async def handle(self, method: str, params: dict):
        """Обрабатываем запрос с ограничением конкурентности"""
        if method not in self.handlers:
            raise ValueError(f"Неизвестный метод: {method}")
        
        async with self.semaphore:
            handler = self.handlers[method]
            return await handler(**params)
 
handler = RequestHandler(max_concurrent=20)
 
# Регистрируем обработчики
handler.register("tools/list", handle_list_tools)
handler.register("tools/call", handle_call_tool)
handler.register("resources/list", handle_list_resources)

from pydantic import BaseModel, validator
 
class ToolCallParams(BaseModel):
    name: str
    arguments: dict
    
    @validator('name')
    def validate_name(cls, v):
        if not v or len(v) > 100:
            raise ValueError("Некорректное имя инструмента")
        return v
    
    @validator('arguments')
    def validate_args(cls, v):
        # Ограничиваем размер JSON
        import json
        if len(json.dumps(v)) > 1_000_000:  # 1MB
            raise ValueError("Слишком большой payload")
        return v
 
async def handle_call_tool_validated(params: dict):
    """Обработчик с валидацией через Pydantic"""
    validated = ToolCallParams(**params)
    return await handle_call_tool(
        validated.name, 
        validated.arguments
    )