Python SDK (FastMCP)

Полное руководство по созданию MCP серверов на Python с использованием FastMCP

FastMCP — официальный высокоуровневый SDK для создания MCP серверов на Python. Использует type hints и docstrings для автоматической генерации схем.

Установка

uv add mcp

pip install mcp

poetry add mcp

С CLI утилитами:

uv add "mcp[cli]"

pip install "mcp[cli]"

poetry add "mcp[cli]"

Требования:

Python 3.10+
asyncio поддержка

Базовый сервер

from mcp.server.fastmcp import FastMCP

# Создание сервера
mcp = FastMCP(
    name="my-server",
    instructions="Сервер для работы с данными. Используй tool1 для X, tool2 для Y."
)

# Запуск
if __name__ == "__main__":
    mcp.run()

Транспорты

FastMCP поддерживает несколько транспортов для связи с клиентами:

Stdio (по умолчанию)

# Стандартный запуск через stdio
mcp.run()

# Или явно
mcp.run(transport="stdio")

SSE (Server-Sent Events)

# Запуск HTTP сервера с SSE
mcp.run(transport="sse", host="0.0.0.0", port=8080)

StreamableHTTP

# Современный HTTP транспорт с поддержкой стриминга
mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

Инструменты (Tools)

Инструменты — функции, которые AI может вызывать.

Базовый инструмент

@mcp.tool()
def add(a: int, b: int) -> int:
    """
    Сложение двух чисел.

    Args:
        a: Первое число
        b: Второе число

    Returns:
        Сумма чисел
    """
    return a + b

Асинхронный инструмент

import aiohttp

@mcp.tool()
async def fetch_url(url: str) -> str:
    """
    Загрузка содержимого URL.

    Args:
        url: URL для загрузки

    Returns:
        Содержимое страницы
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Инструмент с опциональными параметрами

from typing import Optional

@mcp.tool()
def search(
    query: str,
    limit: int = 10,
    offset: int = 0,
    category: Optional[str] = None
) -> list[dict]:
    """
    Поиск в базе данных.

    Args:
        query: Поисковый запрос
        limit: Максимум результатов (по умолчанию 10)
        offset: Смещение для пагинации
        category: Опциональная категория фильтра
    """
    results = database.search(query)
    if category:
        results = [r for r in results if r["category"] == category]
    return results[offset:offset + limit]

Инструмент с Enum

from enum import Enum

class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@mcp.tool()
def create_task(title: str, priority: Priority) -> dict:
    """
    Создание задачи с приоритетом.

    Args:
        title: Название задачи
        priority: Приоритет (low, medium, high)
    """
    return {
        "id": generate_id(),
        "title": title,
        "priority": priority.value
    }

Инструмент с Pydantic моделью

from pydantic import BaseModel, Field

class CreateUserInput(BaseModel):
    name: str = Field(..., description="Имя пользователя")
    email: str = Field(..., description="Email адрес")
    age: int = Field(ge=0, le=150, description="Возраст")

@mcp.tool()
def create_user(data: CreateUserInput) -> dict:
    """Создание нового пользователя"""
    user = User(
        name=data.name,
        email=data.email,
        age=data.age
    )
    db.save(user)
    return user.to_dict()

Ресурсы (Resources)

Ресурсы — данные, которые AI может прочитать.

Статический ресурс

@mcp.resource("config://app")
def get_app_config() -> str:
    """Конфигурация приложения"""
    return json.dumps({
        "version": "1.0.0",
        "debug": False,
        "database": "postgresql://localhost/myapp"
    })

Динамический ресурс с параметрами

@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """
    Чтение файла по пути.

    Args:
        path: Путь к файлу
    """
    with open(path, "r") as f:
        return f.read()

Ресурс с бинарными данными

import base64

@mcp.resource("image://{name}")
def get_image(name: str) -> tuple[str, str]:
    """
    Получение изображения.

    Returns:
        Кортеж (base64_data, mime_type)
    """
    with open(f"images/{name}", "rb") as f:
        data = base64.b64encode(f.read()).decode()
    return data, "image/png"

Ресурс из базы данных

@mcp.resource("db://users/{user_id}")
async def get_user(user_id: int) -> str:
    """Получение пользователя из БД"""
    async with db.acquire() as conn:
        user = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1",
            user_id
        )
        return json.dumps(dict(user))

Промпты (Prompts)

Промпты — шаблоны для структурированных запросов.

Базовый промпт

@mcp.prompt()
def code_review(language: str, code: str) -> str:
    """
    Промпт для код-ревью.

    Args:
        language: Язык программирования
        code: Код для ревью
    """
    return f"""
    Проведи детальное код-ревью следующего {language} кода:

    ```{language}
    {code}

Проверь:

Потенциальные баги и ошибки
Производительность и оптимизации
Соответствие best practices
Читаемость и поддерживаемость
Безопасность

Предложи конкретные улучшения с примерами кода. """

### Промпт с несколькими сообщениями

```python
from mcp.server.fastmcp import Message

@mcp.prompt()
def debug_assistant(error: str, context: str) -> list[Message]:
    """Промпт для помощи с отладкой"""
    return [
        Message(
            role="system",
            content="Ты опытный отладчик Python. Анализируй ошибки и предлагай решения."
        ),
        Message(
            role="user",
            content=f"""
            Ошибка: {error}

            Контекст:
            {context}

            Помоги найти и исправить проблему.
            """
        )
    ]

Контекст и зависимости

Доступ к контексту запроса

from mcp.server.fastmcp import Context

@mcp.tool()
async def get_request_info(ctx: Context) -> dict:
    """Информация о текущем запросе"""
    return {
        "request_id": ctx.request_id,
        "client_id": ctx.client_id,
    }

Логирование через контекст

@mcp.tool()
async def process_data(ctx: Context, data: str) -> str:
    """Обработка данных с логированием"""
    ctx.info(f"Начало обработки: {len(data)} символов")
    ctx.debug("Детали обработки...")

    result = transform(data)

    ctx.info("Обработка завершена")
    return result

Отслеживание прогресса

@mcp.tool()
async def long_task(ctx: Context, items: list[str]) -> str:
    """Длительная задача с прогрессом"""
    total = len(items)

    for i, item in enumerate(items):
        await ctx.report_progress(
            progress=i,
            total=total,
            message=f"Обработка {item}"
        )
        await process_item(item)

    return f"Обработано {total} элементов"

Dependency Injection с Lifespan

from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppContext:
    db: DatabaseConnection
    cache: CacheClient

@asynccontextmanager
async def lifespan(server: FastMCP):
    """Управление жизненным циклом зависимостей"""
    db = await DatabaseConnection.connect("postgresql://localhost/app")
    cache = await CacheClient.connect("redis://localhost")

    try:
        yield AppContext(db=db, cache=cache)
    finally:
        await db.disconnect()
        await cache.disconnect()

mcp = FastMCP("my-server", lifespan=lifespan)

@mcp.tool()
async def query_with_cache(ctx: Context, sql: str) -> list[dict]:
    """Запрос с кешированием"""
    app: AppContext = ctx.request_context.lifespan_context

    # Проверка кеша
    cached = await app.cache.get(sql)
    if cached:
        return cached

    # Запрос к БД
    result = await app.db.query(sql)
    await app.cache.set(sql, result, ttl=300)
    return result

Возврат изображений

from mcp.server.fastmcp import Image

@mcp.tool()
def create_chart(data: list[float]) -> Image:
    """Создание графика из данных"""
    import matplotlib.pyplot as plt
    import io

    plt.figure(figsize=(10, 6))
    plt.plot(data)
    plt.title("График данных")

    buf = io.BytesIO()
    plt.savefig(buf, format='png')
    buf.seek(0)

    return Image(data=buf.read(), format="png")

@mcp.tool()
def get_screenshot(url: str) -> Image:
    """Скриншот веб-страницы"""
    from playwright.sync_api import sync_playwright

    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto(url)
        screenshot = page.screenshot()
        browser.close()

    return Image(data=screenshot, format="png")

Композиция серверов

Используйте mcp.mount() для объединения нескольких серверов:

from mcp.server.fastmcp import FastMCP

# Основной сервер
main = FastMCP("main-server")

# Подсерверы
files_server = FastMCP("files")
db_server = FastMCP("database")

@files_server.tool()
def read_file(path: str) -> str:
    """Чтение файла"""
    return open(path).read()

@db_server.tool()
def query(sql: str) -> list[dict]:
    """SQL запрос"""
    return db.execute(sql)

# Монтирование подсерверов с префиксами
main.mount("files", files_server)
main.mount("db", db_server)

# Инструменты будут доступны как:
# - files_read_file
# - db_query

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

Обработка ошибок

from mcp.server.fastmcp import McpError

@mcp.tool()
def divide(a: float, b: float) -> float:
    """Деление чисел"""
    if b == 0:
        raise McpError("INVALID_PARAMS", "Деление на ноль невозможно")
    return a / b

@mcp.tool()
def read_secure_file(path: str) -> str:
    """Чтение файла с проверкой доступа"""
    if not path.startswith("/allowed/"):
        raise McpError("FORBIDDEN", "Доступ к этому файлу запрещён")
    return open(path).read()

Аннотации инструментов

from mcp.server.fastmcp.tools import ToolAnnotations

@mcp.tool(
    annotations=ToolAnnotations(
        title="Запрос к БД",
        read_only_hint=True,
        open_world_hint=False
    )
)
def safe_query(sql: str) -> list[dict]:
    """Безопасный SQL запрос (только SELECT)"""
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Только SELECT запросы")
    return db.execute(sql)

Модульная структура

Для больших проектов рекомендуется разделять код на модули:

my-server/
├── src/
│   ├── __init__.py
│   ├── server.py           # Точка входа
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── calculator.py
│   │   ├── filesystem.py
│   │   └── database.py
│   ├── resources/
│   │   ├── __init__.py
│   │   └── config.py
│   └── prompts/
│       ├── __init__.py
│       └── templates.py
├── tests/
├── pyproject.toml
└── README.md

src/server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

# Импорт модулей (автоматическая регистрация декораторов)
from .tools import calculator, filesystem, database
from .resources import config
from .prompts import templates

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

src/tools/calculator.py

from ..server import mcp

@mcp.tool()
def add(a: float, b: float) -> float:
    """Сложение"""
    return a + b

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Умножение"""
    return a * b

Альтернатива: композиция через mount

from mcp.server.fastmcp import FastMCP

calculator = FastMCP("calculator")

@calculator.tool()
def add(a: float, b: float) -> float:
    """Сложение"""
    return a + b

# src/server.py
from mcp.server.fastmcp import FastMCP
from .tools.calculator import calculator

mcp = FastMCP("my-server")
mcp.mount("calc", calculator)  # Инструменты: calc_add, calc_multiply

Тестирование

Unit тесты

import pytest
from server import mcp

@pytest.mark.asyncio
async def test_add_tool():
    result = await mcp.call_tool("add", {"a": 2, "b": 3})
    assert result == 5

@pytest.mark.asyncio
async def test_divide_by_zero():
    with pytest.raises(McpError) as exc:
        await mcp.call_tool("divide", {"a": 1, "b": 0})
    assert "Деление на ноль" in str(exc.value)

Интеграционные тесты с MCP Inspector

# Запуск инспектора
npx @modelcontextprotocol/inspector python server.py

# Или через uvx
uvx mcp-inspector python server.py

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["-m", "src.server"],
      "cwd": "/path/to/my-server",
      "env": {
        "DATABASE_URL": "postgresql://localhost/myapp"
      }
    }
  }
}

{
  "mcpServers": {
    "my-server": {
      "command": "uv",
      "args": ["run", "python", "-m", "src.server"],
      "cwd": "/path/to/my-server"
    }
  }
}