Python MCP или как подключить свою LLM ко всему миру - Развертывание MCP-серверов, Универсальный MCP-сервер с разными источниками данных

FROM python:3.11-slim
 
# Рабочая директория
WORKDIR /app
 
# Устанавливаем системные зависимости если нужны
RUN apt-get update && apt-get install -y \
    gcc \
    postgresql-client \
    && rm -rf /var/lib/apt/lists/*
 
# Копируем requirements отдельно для кеширования слоя
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Копируем код сервера
COPY . .
 
# Непривилегированный пользователь для безопасности
RUN useradd -m -u 1000 mcpuser && \
    chown -R mcpuser:mcpuser /app
USER mcpuser
 
# Healthcheck для контроля состояния
HEALTHCHECK --interval=30s --timeout=3s --start-period=10s \
    CMD python -c "import sys; sys.exit(0)"
 
# Переменные окружения по умолчанию
ENV PYTHONUNBUFFERED=1 \
    MCP_PORT=8080
 
EXPOSE 8080
 
CMD ["python", "server.py"]

# Стадия сборки
FROM python:3.11 as builder
 
WORKDIR /build
COPY requirements.txt .
 
# Устанавливаем зависимости в отдельную директорию
RUN pip install --user --no-cache-dir -r requirements.txt
 
# Production стадия
FROM python:3.11-slim
 
WORKDIR /app
 
# Копируем только установленные пакеты
COPY --from=builder /root/.local /root/.local
COPY . .
 
# Добавляем pip packages в PATH
ENV PATH=/root/.local/bin:$PATH
 
USER 1000
 
CMD ["python", "server.py"]

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
 
# Определяем метрики
request_count = Counter(
    'mcp_requests_total',
    'Всего запросов',
    ['tool_name', 'status']
)
 
request_duration = Histogram(
    'mcp_request_duration_seconds',
    'Длительность запросов',
    ['tool_name']
)
 
active_connections = Gauge(
    'mcp_active_connections',
    'Активные соединения'
)
 
error_count = Counter(
    'mcp_errors_total',
    'Всего ошибок',
    ['error_type']
)
 
# Запускаем HTTP-сервер для Prometheus
start_http_server(9090)
 
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    active_connections.inc()
    start_time = time.time()
    
    try:
        result = await execute_tool(name, arguments)
        request_count.labels(tool_name=name, status='success').inc()
        return result
        
    except Exception as e:
        request_count.labels(tool_name=name, status='error').inc()
        error_count.labels(error_type=type(e).__name__).inc()
        raise
        
    finally:
        duration = time.time() - start_time
        request_duration.labels(tool_name=name).observe(duration)
        active_connections.dec()

universal-mcp-server/
├── adapters/
│   ├── __init__.py
│   ├── postgres.py
│   ├── mongodb.py
│   ├── filesystem.py
│   └── external_api.py
├── middleware/
│   ├── __init__.py
│   ├── auth.py
│   ├── logging.py
│   └── rate_limit.py
├── config.py
├── server.py
├── requirements.txt
└── docker-compose.yml

# config.py
from pydantic_settings import BaseSettings
from typing import Optional
 
class Settings(BaseSettings):
    # База данных
    postgres_host: str = "localhost"
    postgres_port: int = 5432
    postgres_db: str = "mcp_demo"
    postgres_user: str = "mcp_user"
    postgres_password: str
    
    # MongoDB
    mongodb_url: str = "mongodb://localhost:27017"
    mongodb_database: str = "mcp_demo"
    
    # Файловая система
    fs_base_path: str = "./data"
    fs_max_file_size: int = 10_000_000
    
    # API
    external_api_key: Optional[str] = None
    external_api_base_url: str = "https://api.example.com"
    
    # Сервер
    mcp_port: int = 8080
    log_level: str = "INFO"
    
    # Безопасность
    jwt_secret: str
    rate_limit_requests: int = 100
    rate_limit_window: int = 60
    
    class Config:
        env_file = ".env"
 
settings = Settings()

# server.py
import asyncio
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json
 
from config import settings
from adapters.postgres import PostgresAdapter
from adapters.mongodb import MongoAdapter
from adapters.filesystem import FileSystemAdapter
from adapters.external_api import ExternalAPIClient
from middleware.logging import setup_logging, RequestLogger
from middleware.rate_limit import RateLimitMiddleware
from middleware.auth import AuthMiddleware
 
# Настраиваем логирование
setup_logging(settings.log_level)
logger = logging.getLogger(__name__)
 
# Создаём сервер
server = Server("universal-mcp-server")
 
# Инициализируем адаптеры
db_adapter = PostgresAdapter(
    host=settings.postgres_host,
    port=settings.postgres_port,
    database=settings.postgres_db,
    user=settings.postgres_user,
    password=settings.postgres_password
)
 
mongo_adapter = MongoAdapter(
    connection_string=settings.mongodb_url,
    database_name=settings.mongodb_database
)
 
fs_adapter = FileSystemAdapter(
    base_path=settings.fs_base_path,
    max_file_size=settings.fs_max_file_size
)
 
api_client = ExternalAPIClient(
    base_url=settings.external_api_base_url,
    api_key=settings.external_api_key
)
 
# Middleware
request_logger = RequestLogger()
rate_limiter = RateLimitMiddleware(
    max_requests=settings.rate_limit_requests,
    time_window=settings.rate_limit_window
)
auth = AuthMiddleware(jwt_secret=settings.jwt_secret)
 
@server.list_tools()
async def handle_list_tools():
    """Возвращаем все доступные инструменты"""
    return [
        # PostgreSQL инструменты
        Tool(
            name="query_postgres",
            description="Выполняет SQL запрос к PostgreSQL",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "limit": {"type": "integer", "default": 100}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="list_pg_tables",
            description="Показывает доступные таблицы в PostgreSQL",
            inputSchema={"type": "object", "properties": {}}
        ),
        
        # MongoDB инструменты
        Tool(
            name="find_documents",
            description="Ищет документы в MongoDB коллекции",
            inputSchema={
                "type": "object",
                "properties": {
                    "collection": {"type": "string"},
                    "filter": {"type": "object"},
                    "limit": {"type": "integer", "default": 100}
                },
                "required": ["collection", "filter"]
            }
        ),
        Tool(
            name="list_collections",
            description="Показывает коллекции MongoDB",
            inputSchema={"type": "object", "properties": {}}
        ),
        
        # Файловая система
        Tool(
            name="read_file",
            description="Читает содержимое файла",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string"}
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="list_files",
            description="Перечисляет файлы в директории",
            inputSchema={
                "type": "object",
                "properties": {
                    "directory": {"type": "string", "default": ""},
                    "pattern": {"type": "string", "default": "*"}
                }
            }
        ),
        Tool(
            name="search_in_files",
            description="Ищет текст в файлах",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "directory": {"type": "string", "default": ""}
                },
                "required": ["query"]
            }
        ),
        
        # Внешний API
        Tool(
            name="api_get",
            description="GET запрос к внешнему API",
            inputSchema={
                "type": "object",
                "properties": {
                    "endpoint": {"type": "string"},
                    "params": {"type": "object", "default": {}}
                },
                "required": ["endpoint"]
            }
        )
    ]
 
@server.call_tool()
@auth.require_token
@rate_limiter.limit
@request_logger.log
async def handle_call_tool(name: str, arguments: dict, **kwargs):
    """Обрабатываем вызовы всех инструментов"""
    
    try:
        # PostgreSQL
        if name == "query_postgres":
            results = await db_adapter.execute_query(
                arguments["query"],
                limit=arguments.get("limit", 100)
            )
            return [TextContent(
                type="text",
                text=json.dumps(results, ensure_ascii=False, indent=2)
            )]
        
        elif name == "list_pg_tables":
            tables = await db_adapter.get_available_tables()
            return [TextContent(
                type="text",
                text=f"Доступные таблицы: {', '.join(tables)}"
            )]
        
        # MongoDB
        elif name == "find_documents":
            docs = await mongo_adapter.find_documents(
                collection=arguments["collection"],
                filter_query=arguments["filter"],
                limit=arguments.get("limit", 100)
            )
            return [TextContent(
                type="text",
                text=json.dumps(docs, ensure_ascii=False, indent=2)
            )]
        
        elif name == "list_collections":
            collections = await mongo_adapter.get_collections()
            return [TextContent(
                type="text",
                text=f"Коллекции: {', '.join(collections)}"
            )]
        
        # Файловая система
        elif name == "read_file":
            content = await fs_adapter.read_file(arguments["path"])
            return [TextContent(type="text", text=content)]
        
        elif name == "list_files":
            files = await fs_adapter.list_files(
                directory=arguments.get("directory", ""),
                pattern=arguments.get("pattern", "*")
            )
            return [TextContent(
                type="text",
                text="
".join(files) if files else "Файлы не найдены"
            )]
        
        elif name == "search_in_files":
            results = await fs_adapter.search_in_files(
                query=arguments["query"],
                directory=arguments.get("directory", "")
            )
            return [TextContent(
                type="text",
                text=json.dumps(results, ensure_ascii=False, indent=2)
            )]
        
        # Внешний API
        elif name == "api_get":
            response = await api_client.get(
                endpoint=arguments["endpoint"],
                params=arguments.get("params", {})
            )
            return [TextContent(
                type="text",
                text=json.dumps(response, ensure_ascii=False, indent=2)
            )]
        
        else:
            raise ValueError(f"Неизвестный инструмент: {name}")
    
    except Exception as e:
        logger.error(f"Ошибка выполнения {name}: {str(e)}", exc_info=True)
        return [TextContent(
            type="text",
            text=f"Ошибка: {str(e)}"
        )]
 
async def initialize():
    """Инициализация адаптеров при старте"""
    logger.info("Инициализация адаптеров...")
    await db_adapter.initialize()
    await mongo_adapter.connect()
    logger.info("Сервер готов к работе")
 
async def cleanup():
    """Очистка ресурсов при остановке"""
    logger.info("Завершение работы...")
    await db_adapter.close()
    logger.info("Сервер остановлен")
 
async def main():
    """Точка входа"""
    await initialize()
    
    try:
        async with stdio_server() as (read_stream, write_stream):
            await server.run(
                read_stream,
                write_stream,
                server.create_initialization_options()
            )
    finally:
        await cleanup()
 
if __name__ == "__main__":
    asyncio.run(main())

# adapters/postgres.py
import asyncpg
from typing import List, Dict, Any, Optional
from contextlib import asynccontextmanager
import asyncio
 
class PostgresAdapter:
    def __init__(
        self,
        host: str,
        port: int,
        database: str,
        user: str,
        password: str,
        min_pool_size: int = 5,
        max_pool_size: int = 20
    ):
        self.connection_params = {
            "host": host,
            "port": port,
            "database": database,
            "user": user,
            "password": password
        }
        self.min_pool_size = min_pool_size
        self.max_pool_size = max_pool_size
        self.pool: Optional[asyncpg.Pool] = None
    
    async def initialize(self):
        """Создаём connection pool"""
        self.pool = await asyncpg.create_pool(
            **self.connection_params,
            min_size=self.min_pool_size,
            max_size=self.max_pool_size,
            command_timeout=60
        )
    
    async def close(self):
        """Закрываем pool"""
        if self.pool:
            await self.pool.close()
    
    @asynccontextmanager
    async def acquire(self):
        """Context manager для получения connection"""
        async with self.pool.acquire() as connection:
            yield connection
    
    async def execute_query(
        self, 
        query: str,
        limit: int = 100
    ) -> List[Dict[str, Any]]:
        """Выполняет SELECT с лимитом"""
        if "limit" not in query.lower():
            query = f"{query.rstrip(';')} LIMIT {limit}"
        
        async with self.acquire() as conn:
            rows = await conn.fetch(query)
            return [dict(row) for row in rows]
    
    async def get_available_tables(self) -> List[str]:
        """Возвращает список таблиц"""
        query = """
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
            AND table_type = 'BASE TABLE'
            ORDER BY table_name
        """
        async with self.acquire() as conn:
            rows = await conn.fetch(query)
            return [row["table_name"] for row in rows]

# middleware/logging.py
import structlog
import time
import uuid
from functools import wraps
from typing import Callable
 
def setup_logging(level: str):
    """Настраивает structlog"""
    structlog.configure(
        processors=[
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
    )
 
class RequestLogger:
    def __init__(self):
        self.logger = structlog.get_logger()
    
    def log(self, func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            request_id = str(uuid.uuid4())[:8]
            start_time = time.time()
            
            log = self.logger.bind(
                request_id=request_id,
                tool_name=kwargs.get('name', 'unknown')
            )
            
            log.info("request_started")
            
            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start_time
                
                log.info(
                    "request_completed",
                    duration_ms=int(duration * 1000),
                    success=True
                )
                
                return result
                
            except Exception as e:
                duration = time.time() - start_time
                
                log.error(
                    "request_failed",
                    duration_ms=int(duration * 1000),
                    error_type=type(e).__name__,
                    error_message=str(e)
                )
                raise
        
        return wrapper

# tests/test_integration.py
import pytest
import asyncio
from server import server
from adapters.postgres import PostgresAdapter
 
@pytest.fixture
async def test_db():
    """Тестовая база данных"""
    adapter = PostgresAdapter(
        host="localhost",
        port=5432,
        database="mcp_test",
        user="test_user",
        password="test_pass"
    )
    
    await adapter.initialize()
    
    # Создаём тестовые данные
    async with adapter.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS test_users (
                id SERIAL PRIMARY KEY,
                name TEXT NOT NULL
            );
            TRUNCATE test_users;
            INSERT INTO test_users (name) VALUES ('Alice'), ('Bob');
        """)
    
    yield adapter
    
    await adapter.close()
 
@pytest.mark.asyncio
async def test_postgres_query(test_db):
    """Проверка запроса к PostgreSQL"""
    results = await test_db.execute_query(
        "SELECT * FROM test_users"
    )
    
    assert len(results) == 2
    assert results[0]["name"] in ["Alice", "Bob"]