Пишем MCP-сервер на Python: подключение БД, API и файлов как инструментов

types.Tool(
    name="read_file",
    description=(
        "Читает содержимое файла по указанному пути. "
        "Используй когда нужно получить содержимое конкретного файла. "
        "Не подходит для поиска по файлам — используй search_files. "
        "Возвращает текст файла в кодировке UTF-8."
    ),
    inputSchema={
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "Абсолютный или относительный путь к файлу"
            },
            "encoding": {
                "type": "string",
                "description": "Кодировка файла, по умолчанию utf-8",
                "default": "utf-8",
                "enum": ["utf-8", "latin-1", "cp1251"]
            }
        },
        "required": ["path"]
    }
)

"""
Файловые инструменты для MCP-сервера.
pip install mcp
"""
import fnmatch
import logging
from pathlib import Path
from mcp import types

logger = logging.getLogger(__name__)


class FileToolset:
    """Группа файловых инструментов с sandbox-ограничением."""

    def __init__(self, allowed_paths: list[str]):
        # Резолвим пути сразу, чтобы не делать это при каждом вызове
        self.allowed = [Path(p).resolve() for p in allowed_paths]

    # ── Внутренние методы ──────────────────────────────────────────────

    def _check_path(self, path: str) -> Path:
        """Проверяет, что путь находится в разрешённой зоне."""
        resolved = Path(path).resolve()
        for allowed in self.allowed:
            try:
                resolved.relative_to(allowed)
                return resolved          # путь разрешён
            except ValueError:
                continue
        raise PermissionError(
            f"Путь '{path}' находится вне разрешённых директорий: "
            + ", ".join(str(a) for a in self.allowed)
        )

    def _ok(self, text: str) -> list[types.TextContent]:
        return [types.TextContent(type="text", text=text)]

    def _err(self, msg: str) -> list[types.TextContent]:
        logger.warning("FileToolset error: %s", msg)
        return [types.TextContent(type="text", text=f"[ERROR] {msg}")]

    # ── Инструменты ───────────────────────────────────────────────────

    async def read_file(self, path: str, encoding: str = "utf-8") -> list[types.TextContent]:
        try:
            p = self._check_path(path)
            if not p.is_file():
                return self._err(f"'{path}' не является файлом")
            content = p.read_text(encoding=encoding)
            logger.info("read_file: %s (%d bytes)", path, len(content))
            return self._ok(content)
        except PermissionError as e:
            return self._err(str(e))
        except Exception as e:
            return self._err(f"Ошибка чтения: {e}")

    async def write_file(self, path: str, content: str, encoding: str = "utf-8") -> list[types.TextContent]:
        try:
            p = self._check_path(path)
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_text(content, encoding=encoding)
            return self._ok(f"Записано {len(content)} символов → {path}")
        except PermissionError as e:
            return self._err(str(e))
        except Exception as e:
            return self._err(f"Ошибка записи: {e}")

    async def list_directory(self, path: str) -> list[types.TextContent]:
        try:
            p = self._check_path(path)
            if not p.is_dir():
                return self._err(f"'{path}' не является директорией")
            entries = []
            for item in sorted(p.iterdir()):
                prefix = "📁 " if item.is_dir() else "📄 "
                size = f"  ({item.stat().st_size} bytes)" if item.is_file() else ""
                entries.append(f"{prefix}{item.name}{size}")
            return self._ok("\n".join(entries) if entries else "(пустая директория)")
        except PermissionError as e:
            return self._err(str(e))

    async def search_files(self, path: str, pattern: str) -> list[types.TextContent]:
        """Рекурсивный поиск файлов по glob-паттерну."""
        try:
            base = self._check_path(path)
            if not base.is_dir():
                return self._err(f"'{path}' не является директорией")
            matches = [
                str(f.relative_to(base))
                for f in base.rglob("*")
                if f.is_file() and fnmatch.fnmatch(f.name, pattern)
            ]
            if not matches:
                return self._ok(f"Файлы по паттерну '{pattern}' не найдены")
            return self._ok(f"Найдено {len(matches)} файлов:\n" + "\n".join(matches))
        except PermissionError as e:
            return self._err(str(e))

    # ── Описания инструментов ─────────────────────────────────────────

    def get_tools(self) -> list[types.Tool]:
        return [
            types.Tool(
                name="read_file",
                description="Читает содержимое файла по пути. Возвращает текст в UTF-8.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Путь к файлу"},
                        "encoding": {"type": "string", "description": "Кодировка", "default": "utf-8"}
                    },
                    "required": ["path"]
                }
            ),
            types.Tool(
                name="write_file",
                description="Записывает текст в файл. Создаёт директории при необходимости.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Путь к файлу"},
                        "content": {"type": "string", "description": "Текст для записи"}
                    },
                    "required": ["path", "content"]
                }
            ),
            types.Tool(
                name="list_directory",
                description="Показывает содержимое директории (не рекурсивно).",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Путь к директории"}
                    },
                    "required": ["path"]
                }
            ),
            types.Tool(
                name="search_files",
                description="Рекурсивно ищет файлы по glob-паттерну (например '*.py', '*.log').",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Базовая директория для поиска"},
                        "pattern": {"type": "string", "description": "Glob-паттерн, например '*.py'"}
                    },
                    "required": ["path", "pattern"]
                }
            ),
        ]

"""
Инструменты для работы с SQLite через MCP.
pip install mcp aiosqlite
"""
import json
import logging
import aiosqlite
from mcp import types

logger = logging.getLogger(__name__)


class SQLiteToolset:
    """
    Инструменты для SQLite с разделением read/write.
    allow_writes=False — сервер в режиме только для чтения.
    """

    def __init__(self, db_path: str, allow_writes: bool = False):
        self.db_path = db_path
        self.allow_writes = allow_writes

    async def query_db(self, sql: str, params: list | None = None) -> list[types.TextContent]:
        """Выполняет SELECT-запрос и возвращает результат в JSON."""
        sql_upper = sql.strip().upper()
        # Разрешаем только SELECT и WITH (CTE)
        if not (sql_upper.startswith("SELECT") or sql_upper.startswith("WITH")):
            return [types.TextContent(type="text", text="[ERROR] query_db принимает только SELECT/WITH. Для изменений используй execute_db.")]

        try:
            async with aiosqlite.connect(self.db_path) as db:
                db.row_factory = aiosqlite.Row
                cursor = await db.execute(sql, params or [])
                rows = await cursor.fetchmany(500)  # Лимит на количество строк
                if not rows:
                    return [types.TextContent(type="text", text="(запрос вернул 0 строк)")]

                data = [dict(row) for row in rows]
                result = json.dumps(data, ensure_ascii=False, indent=2, default=str)
                note = f"\n\n(показано {len(data)} строк)" if len(data) == 500 else ""
                return [types.TextContent(type="text", text=result + note)]
        except Exception as e:
            logger.error("query_db error: %s | sql: %s", e, sql[:200])
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    async def execute_db(self, sql: str, params: list | None = None) -> list[types.TextContent]:
        """Выполняет INSERT/UPDATE/DELETE. Требует allow_writes=True."""
        if not self.allow_writes:
            return [types.TextContent(type="text", text="[ERROR] Сервер настроен в режиме только для чтения.")]

        sql_upper = sql.strip().upper()
        # Запрещаем DDL и DROP
        for dangerous in ("DROP ", "TRUNCATE ", "CREATE ", "ALTER ", "ATTACH ", "DETACH "):
            if sql_upper.startswith(dangerous):
                return [types.TextContent(type="text", text=f"[ERROR] Операция '{dangerous.strip()}' запрещена через этот инструмент.")]

        try:
            async with aiosqlite.connect(self.db_path) as db:
                cursor = await db.execute(sql, params or [])
                await db.commit()
                return [types.TextContent(type="text", text=f"OK. Затронуто строк: {cursor.rowcount}")]
        except Exception as e:
            logger.error("execute_db error: %s | sql: %s", e, sql[:200])
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    async def list_tables(self) -> list[types.TextContent]:
        """Возвращает список таблиц и их базовую информацию."""
        try:
            async with aiosqlite.connect(self.db_path) as db:
                db.row_factory = aiosqlite.Row
                cursor = await db.execute(
                    "SELECT name, type FROM sqlite_master WHERE type IN ('table','view') ORDER BY type, name"
                )
                rows = await cursor.fetchall()
                if not rows:
                    return [types.TextContent(type="text", text="(база данных пустая)")]
                lines = [f"[{row['type']}] {row['name']}" for row in rows]
                return [types.TextContent(type="text", text="\n".join(lines))]
        except Exception as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    async def describe_table(self, table: str) -> list[types.TextContent]:
        """Возвращает схему таблицы: колонки, типы, nullable, PK."""
        try:
            async with aiosqlite.connect(self.db_path) as db:
                db.row_factory = aiosqlite.Row
                cursor = await db.execute(f"PRAGMA table_info('{table}')")
                cols = await cursor.fetchall()
                if not cols:
                    return [types.TextContent(type="text", text=f"Таблица '{table}' не найдена")]

                lines = [f"Таблица: {table}", "─" * 40]
                for col in cols:
                    pk = " [PK]" if col["pk"] else ""
                    nn = " NOT NULL" if col["notnull"] else ""
                    dflt = f" DEFAULT {col['dflt_value']}" if col["dflt_value"] else ""
                    lines.append(f"  {col['name']}: {col['type']}{pk}{nn}{dflt}")
                return [types.TextContent(type="text", text="\n".join(lines))]
        except Exception as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    def get_tools(self) -> list[types.Tool]:
        tools = [
            types.Tool(
                name="query_db",
                description="Выполняет SELECT-запрос к базе данных. Возвращает JSON-массив строк. Максимум 500 строк.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "SELECT-запрос"},
                        "params": {"type": "array", "description": "Параметры запроса (список значений для ?)", "items": {}}
                    },
                    "required": ["sql"]
                }
            ),
            types.Tool(
                name="list_tables",
                description="Возвращает список таблиц и представлений в базе данных.",
                inputSchema={"type": "object", "properties": {}, "required": []}
            ),
            types.Tool(
                name="describe_table",
                description="Показывает схему таблицы: колонки, типы, ограничения.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "table": {"type": "string", "description": "Название таблицы"}
                    },
                    "required": ["table"]
                }
            ),
        ]
        if self.allow_writes:
            tools.append(types.Tool(
                name="execute_db",
                description="Выполняет INSERT/UPDATE/DELETE запрос. Не поддерживает DDL (CREATE/DROP/ALTER).",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "INSERT/UPDATE/DELETE запрос"},
                        "params": {"type": "array", "description": "Параметры запроса", "items": {}}
                    },
                    "required": ["sql"]
                }
            ))
        return tools

"""
HTTP-инструменты для MCP-сервера.
pip install mcp httpx
"""
import json
import logging
from urllib.parse import urlparse
import httpx
from mcp import types

logger = logging.getLogger(__name__)

DEFAULT_HEADERS = {
    "User-Agent": "MCP-Agent/1.0",
    "Accept": "application/json, text/plain, */*",
}


class APIToolset:
    """
    HTTP-инструменты с whitelist доменов и таймаутами.
    allowed_domains=None означает запрет всех запросов.
    allowed_domains=["*"] — разрешить всё (только для dev).
    """

    def __init__(
        self,
        allowed_domains: list[str] | None = None,
        timeout: float = 15.0,
        extra_headers: dict | None = None,
    ):
        self.allowed_domains = allowed_domains or []
        self.timeout = httpx.Timeout(timeout)
        self.headers = {**DEFAULT_HEADERS, **(extra_headers or {})}

    def _check_url(self, url: str) -> str:
        """Проверяет домен по whitelist."""
        if "*" in self.allowed_domains:
            return url
        parsed = urlparse(url)
        domain = parsed.netloc.lower()
        for allowed in self.allowed_domains:
            if domain == allowed or domain.endswith(f".{allowed}"):
                return url
        raise PermissionError(
            f"Домен '{domain}' не разрешён. Разрешены: {self.allowed_domains}"
        )

    def _truncate(self, text: str, limit: int = 8000) -> str:
        if len(text) <= limit:
            return text
        return text[:limit] + f"\n\n[...обрезано, всего {len(text)} символов]"

    async def http_get(self, url: str, headers: dict | None = None) -> list[types.TextContent]:
        try:
            self._check_url(url)
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                resp = await client.get(
                    url,
                    headers={**self.headers, **(headers or {})},
                    follow_redirects=True,
                )
                resp.raise_for_status()
                return [types.TextContent(type="text", text=self._truncate(resp.text))]
        except PermissionError as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]
        except httpx.TimeoutException:
            return [types.TextContent(type="text", text=f"[ERROR] Таймаут запроса к {url}")]
        except httpx.HTTPStatusError as e:
            return [types.TextContent(type="text", text=f"[ERROR] HTTP {e.response.status_code}: {url}")]
        except Exception as e:
            logger.error("http_get error: %s | url: %s", e, url)
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    async def fetch_json(self, url: str, headers: dict | None = None) -> list[types.TextContent]:
        """Получает JSON и форматирует его читаемо."""
        try:
            self._check_url(url)
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                resp = await client.get(
                    url,
                    headers={**self.headers, **(headers or {})},
                    follow_redirects=True,
                )
                resp.raise_for_status()
                data = resp.json()
                formatted = json.dumps(data, ensure_ascii=False, indent=2)
                return [types.TextContent(type="text", text=self._truncate(formatted))]
        except PermissionError as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    async def http_post(
        self,
        url: str,
        body: dict | None = None,
        headers: dict | None = None,
    ) -> list[types.TextContent]:
        try:
            self._check_url(url)
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                resp = await client.post(
                    url,
                    json=body,
                    headers={**self.headers, **(headers or {})},
                )
                resp.raise_for_status()
                return [types.TextContent(type="text", text=self._truncate(resp.text))]
        except PermissionError as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"[ERROR] {e}")]

    def get_tools(self) -> list[types.Tool]:
        return [
            types.Tool(
                name="http_get",
                description="Выполняет GET-запрос и возвращает тело ответа. Для JSON лучше использовать fetch_json.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "url":     {"type": "string", "description": "URL запроса"},
                        "headers": {"type": "object", "description": "Дополнительные заголовки"}
                    },
                    "required": ["url"]
                }
            ),
            types.Tool(
                name="fetch_json",
                description="Получает JSON по URL и форматирует его для удобного чтения.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "url":     {"type": "string", "description": "URL JSON-эндпоинта"},
                        "headers": {"type": "object", "description": "Дополнительные заголовки"}
                    },
                    "required": ["url"]
                }
            ),
            types.Tool(
                name="http_post",
                description="Выполняет POST-запрос с JSON-телом.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "url":     {"type": "string", "description": "URL эндпоинта"},
                        "body":    {"type": "object", "description": "JSON-тело запроса"},
                        "headers": {"type": "object", "description": "Дополнительные заголовки"}
                    },
                    "required": ["url"]
                }
            ),
        ]

from mcp.server import Server
from mcp import types

server = Server("my-server")


@server.list_resources()
async def list_resources() -> list[types.Resource]:
    """Объявляем доступные ресурсы."""
    return [
        types.Resource(
            uri="db://schema",
            name="Database Schema",
            description="Схема всех таблиц базы данных в текстовом формате",
            mimeType="text/plain",
        ),
        types.Resource(
            uri="file:///project/README.md",
            name="Project README",
            description="Документация проекта",
            mimeType="text/markdown",
        ),
    ]


@server.read_resource()
async def read_resource(uri: str) -> str:
    """Возвращает содержимое ресурса по URI."""
    if uri == "db://schema":
        # Например, читаем схему из SQLite
        import aiosqlite
        async with aiosqlite.connect("app.db") as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            )
            tables = [row["name"] for row in await cursor.fetchall()]
            lines = []
            for table in tables:
                c = await db.execute(f"PRAGMA table_info('{table}')")
                cols = await c.fetchall()
                col_strs = ", ".join(f"{col['name']} {col['type']}" for col in cols)
                lines.append(f"{table}({col_strs})")
            return "\n".join(lines)

    if uri.startswith("file://"):
        path = uri[7:]
        return open(path, encoding="utf-8").read()

    raise ValueError(f"Unknown resource URI: {uri}")

"""
Production MCP-сервер: файлы + БД + API.
pip install mcp aiosqlite httpx

Запуск: python server.py
"""
import asyncio
import logging
import os
from dataclasses import dataclass, field
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# Импорт наших toolset'ов (предполагаем, что они в отдельных файлах)
from file_tools import FileToolset
from db_tools   import SQLiteToolset
from api_tools  import APIToolset

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
    handlers=[logging.FileHandler("mcp_server.log")]
    # ВАЖНО: не добавляем StreamHandler — stdout занят stdio-транспортом
)
logger = logging.getLogger("mcp_server")


@dataclass
class ServerConfig:
    # Файловая система
    allowed_paths: list[str] = field(default_factory=lambda: ["/tmp/agent-workspace"])
    # База данных
    db_path: str = "app.db"
    allow_db_writes: bool = False
    # API
    allowed_domains: list[str] = field(default_factory=lambda: ["api.github.com", "httpbin.org"])
    api_timeout: float = 15.0
    extra_api_headers: dict = field(default_factory=dict)


def build_server(config: ServerConfig) -> Server:
    server = Server("multi-capability-server")

    # Инициализируем toolset'ы
    files = FileToolset(allowed_paths=config.allowed_paths)
    db    = SQLiteToolset(db_path=config.db_path, allow_writes=config.allow_db_writes)
    api   = APIToolset(
        allowed_domains=config.allowed_domains,
        timeout=config.api_timeout,
        extra_headers=config.extra_api_headers,
    )

    # Собираем единый реестр инструментов
    all_tools = files.get_tools() + db.get_tools() + api.get_tools()

    # Диспетчер по имени инструмента
    DISPATCH = {
        "read_file":       lambda a: files.read_file(a["path"], a.get("encoding", "utf-8")),
        "write_file":      lambda a: files.write_file(a["path"], a["content"], a.get("encoding", "utf-8")),
        "list_directory":  lambda a: files.list_directory(a["path"]),
        "search_files":    lambda a: files.search_files(a["path"], a["pattern"]),
        "query_db":        lambda a: db.query_db(a["sql"], a.get("params")),
        "execute_db":      lambda a: db.execute_db(a["sql"], a.get("params")),
        "list_tables":     lambda a: db.list_tables(),
        "describe_table":  lambda a: db.describe_table(a["table"]),
        "http_get":        lambda a: api.http_get(a["url"], a.get("headers")),
        "fetch_json":      lambda a: api.fetch_json(a["url"], a.get("headers")),
        "http_post":       lambda a: api.http_post(a["url"], a.get("body"), a.get("headers")),
    }

    @server.list_tools()
    async def list_tools() -> list[types.Tool]:
        return all_tools

    @server.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
        logger.info("tool_call: %s | args_keys: %s", name, list(arguments.keys()))
        handler = DISPATCH.get(name)
        if handler is None:
            return [types.TextContent(type="text", text=f"[ERROR] Инструмент '{name}' не найден")]
        try:
            return await handler(arguments)
        except KeyError as e:
            return [types.TextContent(type="text", text=f"[ERROR] Отсутствует обязательный параметр: {e}")]
        except Exception as e:
            logger.exception("Unexpected error in %s", name)
            return [types.TextContent(type="text", text=f"[ERROR] Внутренняя ошибка сервера: {type(e).__name__}")]

    return server


async def main():
    config = ServerConfig(
        allowed_paths=[
            os.path.expanduser("~/projects"),
            "/tmp/agent-workspace",
        ],
        db_path=os.environ.get("DB_PATH", "app.db"),
        allow_db_writes=os.environ.get("ALLOW_WRITES", "").lower() == "true",
        allowed_domains=os.environ.get("ALLOWED_DOMAINS", "api.github.com").split(","),
    )

    server = build_server(config)
    logger.info("Starting MCP server | tools: %d | paths: %s", 11, config.allowed_paths)

    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "DB_PATH": "/Users/ivan/projects/app.db",
        "ALLOW_WRITES": "false",
        "ALLOWED_DOMAINS": "api.github.com,jsonplaceholder.typicode.com"
      }
    }
  }
}