Tools API

Tools — это действия MCP. В этой теме научимся создавать инструменты для вызова функций, валидации параметров и обработки результатов.

#1. Что такое Tools?

Tools (Инструменты) — функции, которые MCP сервер предоставляет для вызова через LLM. В отличие от Resources (только чтение), Tools могут выполнять действия и изменять данные.

#Отличия Tools от Resources

Характеристика	Resources	Tools
Назначение	Чтение данных	Выполнение действий
Параметры	Через URI	Через JSON Schema
Возврат	Содержимое ресурса	Результат выполнения
Side effects	Нет (только чтение)	Могут изменять данные

#Примеры инструментов

search_database(query, limit)  → Поиск в базе знаний
send_email(to, subject, body)  → Отправка email
create_ticket(title, desc)     → Создание тикета в Jira
run_analysis(data_id)          → Запуск аналитики

#2. Базовый синтаксис

#Простой инструмент

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Tools Demo")


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

Сгенерированная JSON Schema:

{
  "name": "add",
  "description": "Сложение двух чисел.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "a": {
        "type": "integer",
        "description": "Первое число"
      },
      "b": {
        "type": "integer",
        "description": "Второе число"
      }
    },
    "required": ["a", "b"]
  }
}

#Инструмент с необязательными параметрами

@mcp.tool()
def greet(
    name: str,
    greeting: str = "Hello",
    punctuation: str = "!",
    times: int = 1
) -> str:
    """
    Персональное приветствие.
    
    Args:
        name: Имя человека
        greeting: Слово приветствия (по умолчанию "Hello")
        punctuation: Знак препинания (по умолчанию "!")
        times: Количество повторений (по умолчанию 1)
    """
    message = f"{greeting}, {name}{punctuation}"
    if times > 1:
        return "\n".join([message] * times)
    return message

Вызов:

greet(name="Alice")                    → "Hello, Alice!"
greet(name="Alice", greeting="Hi")     → "Hi, Alice!"
greet(name="Bob", times=3)             → "Hello, Bob!\nHello, Bob!\nHello, Bob!"

#3. Типы параметров

#Поддерживаемые типы

FastMCP автоматически генерирует JSON Schema из Python type hints:

Python тип	JSON Schema тип	Пример
`str`	string	`name: str`
`int`	integer	`count: int`
`float`	number	`price: float`
`bool`	boolean	`active: bool`
`list[str]`	array of strings	`tags: list[str]`
`dict`	object	`metadata: dict`
`Optional[str]`	string (nullable)	`comment: Optional[str]`

#Примеры сложных типов

from typing import List, Dict, Optional


@mcp.tool()
def create_user(
    name: str,
    email: str,
    roles: List[str] = None,
    metadata: Dict = None,
    bio: Optional[str] = None
) -> dict:
    """
    Создание пользователя.
    
    Args:
        name: Имя пользователя
        email: Email адрес
        roles: Список ролей (по умолчанию ["user"])
        metadata: Дополнительные метаданные
        bio: Краткая биография (необязательно)
    """
    return {
        "name": name,
        "email": email,
        "roles": roles or ["user"],
        "metadata": metadata or {},
        "bio": bio
    }

#4. Валидация параметров

#Базовая валидация

@mcp.tool()
def search(query: str, limit: int = 10) -> list:
    """
    Поиск в базе знаний.
    
    Args:
        query: Поисковый запрос
        limit: Максимум результатов (1-100)
    """
    # Валидация длины запроса
    if len(query) < 2:
        raise ValueError("Запрос должен быть минимум 2 символа")
    
    if len(query) > 200:
        raise ValueError("Запрос не должен превышать 200 символов")
    
    # Валидация лимита
    if limit < 1 or limit > 100:
        raise ValueError("limit должен быть от 1 до 100")
    
    # Поиск
    results = database.search(query, limit)
    return results

#Валидация с regex

import re


@mcp.tool()
def send_email(to: str, subject: str, body: str) -> str:
    """
    Отправка email.
    
    Args:
        to: Email получателя
        subject: Тема письма
        body: Текст письма
    """
    # Валидация email
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    if not re.match(email_pattern, to):
        raise ValueError(f"Невалидный email адрес: {to}")
    
    # Валидация темы
    if len(subject) > 100:
        raise ValueError("Тема не должна превышать 100 символов")
    
    # Отправка
    email_service.send(to=to, subject=subject, body=body)
    return f"Email sent to {to}"

#Валидация через Pydantic

from pydantic import BaseModel, EmailStr, Field, validator


class CreateUserInput(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    roles: list[str] = Field(default_factory=lambda: ["user"])
    
    @validator('name')
    def validate_name(cls, v):
        if not v.isalpha() and ' ' not in v:
            raise ValueError('Имя должно содержать буквы и пробелы')
        return v.title()


@mcp.tool()
def create_user_validated(input_data: dict) -> dict:
    """
    Создание пользователя с валидацией Pydantic.
    
    Args:
        input_data: Данные пользователя в формате JSON
    """
    try:
        user = CreateUserInput(**input_data)
        return {"status": "success", "user": user.dict()}
    except ValueError as e:
        raise ValueError(f"Validation error: {e}")

#5. Обработка ошибок

#Возврат ошибок

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

Ответ MCP при ошибке:

{
  "content": [
    {
      "type": "text",
      "text": "Error: Деление на ноль невозможно"
    }
  ],
  "isError": true
}

#Разные типы ошибок

class NotFoundError(Exception):
    """Ресурс не найден."""
    pass


class PermissionError(Exception):
    """Недостаточно прав."""
    pass


@mcp.tool()
def get_document(doc_id: str, user_id: str) -> dict:
    """Получение документа."""
    # Проверка существования
    doc = database.get_document(doc_id)
    if doc is None:
        raise NotFoundError(f"Document {doc_id} not found")
    
    # Проверка прав
    if doc.owner_id != user_id and not user_is_admin(user_id):
        raise PermissionError(f"Access denied to document {doc_id}")
    
    return doc

#Логирование ошибок

import logging

logger = logging.getLogger(__name__)


@mcp.tool()
def process_payment(amount: float, card_number: str) -> str:
    """Обработка платежа."""
    try:
        # Валидация
        if amount <= 0:
            raise ValueError("Amount must be positive")
        
        # Обработка (скрываем детали от LLM)
        result = payment_gateway.charge(card_number, amount)
        
        logger.info(f"Payment processed: {amount}$ for card ending {card_number[-4:]}")
        return f"Payment of {amount}$ successful"
        
    except PaymentGatewayError as e:
        # Логируем полную ошибку
        logger.exception(f"Payment gateway error: {e}")
        # LLM показываем упрощённую версию
        raise ValueError("Payment service temporarily unavailable")

#6. Сложные возвращаемые значения

#Возврат структурированных данных

from dataclasses import dataclass
from typing import List


@dataclass
class SearchResult:
    title: str
    url: str
    score: float


@mcp.tool()
def search_articles(query: str, limit: int = 10) -> List[dict]:
    """
    Поиск статей.
    
    Args:
        query: Поисковый запрос
        limit: Максимум результатов
    """
    results = database.search(query, limit)
    
    # Преобразование в словари для JSON сериализации
    return [
        {
            "title": r.title,
            "url": r.url,
            "score": r.score
        }
        for r in results
    ]

#Форматированный вывод

@mcp.tool()
def get_report(start_date: str, end_date: str) -> str:
    """
    Отчёт за период.
    
    Args:
        start_date: Начальная дата (YYYY-MM-DD)
        end_date: Конечная дата (YYYY-MM-DD)
    """
    data = analytics.get_report(start_date, end_date)
    
    # Форматированный вывод
    return f"""
Отчёт за период: {start_date} — {end_date}
{'=' * 50}

Продажи:
  • Всего заказов: {data['total_orders']:,}
  • Общая выручка: ${data['revenue']:,.2f}
  • Средний чек: ${data['avg_check']:,.2f}

Топ товаров:
{chr(10).join(f'  • {item["name"]}: {item["count"]} шт.' for item in data['top_items'])}

Конверсия: {data['conversion_rate']:.2f}%
"""

#Markdown вывод для LLM

@mcp.tool()
def analyze_code(code: str) -> str:
    """Анализ кода."""
    issues = lint_code(code)
    
    # Форматирование в Markdown для лучшей читаемости
    md_output = "## Анализ кода\n\n"
    
    if not issues:
        md_output += "✅ **Проблем не найдено**\n"
    else:
        md_output += f"### Найдено проблем: {len(issues)}\n\n"
        
        for i, issue in enumerate(issues, 1):
            severity_icon = {"error": "🔴", "warning": "🟡", "info": "🔵"}[issue.severity]
            md_output += f"{severity_icon} **{i}. {issue.type}** (строка {issue.line})\n"
            md_output += f"   ```\n   {issue.message}\n   ```\n\n"
    
    return md_output

#7. Асинхронные инструменты

#Async/await в инструментах

import aiohttp


@mcp.tool()
async def fetch_url(url: str) -> str:
    """
    Получение содержимого URL.
    
    Args:
        url: URL для запроса
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=10) as response:
            response.raise_for_status()
            return await response.text()

#Параллельное выполнение

import asyncio


@mcp.tool()
async def fetch_multiple_urls(urls: list[str]) -> str:
    """
    Параллельное получение нескольких URL.
    
    Args:
        urls: Список URL для запроса
    """
    async def fetch(session, url):
        try:
            async with session.get(url, timeout=10) as response:
                return f"{url}: {response.status} OK"
        except Exception as e:
            return f"{url}: Error - {e}"
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    return "\n".join(results)

#Асинхронная работа с БД

import asyncpg


@mcp.tool()
async def query_database(sql: str, params: dict = None) -> str:
    """
    Выполнение SQL запроса (только SELECT).
    
    Args:
        sql: SQL запрос (только SELECT)
        params: Параметры запроса
    """
    # Валидация: только SELECT
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Разрешены только SELECT запросы")
    
    conn = await asyncpg.connect(
        host="localhost",
        user="postgres",
        password="secret",
        database="mydb"
    )
    
    try:
        rows = await conn.fetch(sql, **(params or {}))
        
        if not rows:
            return "Нет результатов"
        
        # Форматирование результатов
        columns = list(rows[0].keys())
        output = " | ".join(columns) + "\n"
        output += "-" * len(output) + "\n"
        
        for row in rows[:100]:  # Ограничение 100 строк
            output += " | ".join(str(v) for v in row.values()) + "\n"
        
        return output
    finally:
        await conn.close()

#8. Инструменты с состоянием

#Использование классов

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Stateful Tools")


class ShoppingCart:
    def __init__(self):
        self.items = {}
    
    def add(self, product_id: str, quantity: int, price: float):
        if product_id in self.items:
            self.items[product_id]['quantity'] += quantity
        else:
            self.items[product_id] = {
                'quantity': quantity,
                'price': price
            }
    
    def remove(self, product_id: str):
        if product_id in self.items:
            del self.items[product_id]
    
    def total(self) -> float:
        return sum(
            item['quantity'] * item['price']
            for item in self.items.values()
        )
    
    def list_items(self) -> list:
        return [
            {
                "product_id": pid,
                "quantity": data['quantity'],
                "price": data['price'],
                "subtotal": data['quantity'] * data['price']
            }
            for pid, data in self.items.items()
        ]


# Глобальное состояние (для демонстрации)
carts = {}


@mcp.tool()
def cart_add(user_id: str, product_id: str, quantity: int, price: float) -> str:
    """Добавить товар в корзину."""
    if user_id not in carts:
        carts[user_id] = ShoppingCart()
    
    carts[user_id].add(product_id, quantity, price)
    return f"Added {quantity} x {product_id} to cart"


@mcp.tool()
def cart_remove(user_id: str, product_id: str) -> str:
    """Удалить товар из корзины."""
    if user_id not in carts:
        raise ValueError("Cart not found")
    
    carts[user_id].remove(product_id)
    return f"Removed {product_id} from cart"


@mcp.tool()
def cart_total(user_id: str) -> str:
    """Получить общую сумму корзины."""
    if user_id not in carts:
        return "Cart is empty"
    
    total = carts[user_id].total()
    return f"Cart total: ${total:.2f}"


@mcp.tool()
def cart_list(user_id: str) -> str:
    """Показать содержимое корзины."""
    if user_id not in carts:
        return "Cart is empty"
    
    items = carts[user_id].list_items()
    if not items:
        return "Cart is empty"
    
    output = "Shopping Cart:\n"
    for item in items:
        output += f"  • {item['product_id']}: {item['quantity']} x ${item['price']} = ${item['subtotal']:.2f}\n"
    output += f"\nTotal: ${carts[user_id].total():.2f}"
    return output

#9. Инструменты для работы с внешними API

#REST API клиент

import aiohttp
from typing import Optional


@mcp.tool()
async def github_get_repo(owner: str, repo: str) -> dict:
    """
    Информация о репозитории GitHub.
    
    Args:
        owner: Владелец репозитория
        repo: Название репозитория
    """
    url = f"https://api.github.com/repos/{owner}/{repo}"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 404:
                raise ValueError(f"Repository {owner}/{repo} not found")
            response.raise_for_status()
            data = await response.json()
            
            return {
                "name": data["full_name"],
                "description": data["description"],
                "stars": data["stargazers_count"],
                "forks": data["forks_count"],
                "language": data["language"],
                "url": data["html_url"]
            }


@mcp.tool()
async def github_search_repos(query: str, limit: int = 10) -> list:
    """
    Поиск репозиториев на GitHub.
    
    Args:
        query: Поисковый запрос
        limit: Максимум результатов
    """
    url = "https://api.github.com/search/repositories"
    params = {"q": query, "per_page": limit}
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url, params=params) as response:
            response.raise_for_status()
            data = await response.json()
            
            return [
                {
                    "name": repo["full_name"],
                    "description": repo["description"],
                    "stars": repo["stargazers_count"],
                    "url": repo["html_url"]
                }
                for repo in data["items"]
            ]

#Инструменты с авторизацией

import os


@mcp.tool()
async def slack_send_message(channel: str, message: str) -> str:
    """
    Отправка сообщения в Slack.
    
    Args:
        channel: Название канала (#general) или ID
        message: Текст сообщения
    """
    token = os.getenv("SLACK_BOT_TOKEN")
    if not token:
        raise ValueError("SLACK_BOT_TOKEN not configured")
    
    url = "https://slack.com/api/chat.postMessage"
    headers = {"Authorization": f"Bearer {token}"}
    data = {"channel": channel, "text": message}
    
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=data) as response:
            result = await response.json()
            
            if not result["ok"]:
                raise ValueError(f"Slack API error: {result['error']}")
            
            return f"Message sent to {channel}"

#10. Best Practices для Tools

#1. Одно действие — один инструмент

# ❌ ПЛОХО: Инструмент делает слишком много
@mcp.tool()
def process_order(user_id: str, items: list, payment: dict, shipping: dict) -> dict:
    # Валидация, создание заказа, оплата, доставка, уведомление...
    pass

# ✅ ХОРОШО: Разделение на атомарные инструменты
@mcp.tool()
def create_order(user_id: str, items: list) -> dict:
    """Создание заказа."""
    pass

@mcp.tool()
def process_payment(order_id: str, payment: dict) -> str:
    """Оплата заказа."""
    pass

@mcp.tool()
def schedule_shipping(order_id: str, shipping: dict) -> str:
    """Планирование доставки."""
    pass

#2. Понятные описания

# ❌ ПЛОХО: Бесполезное описание
@mcp.tool()
def proc(data: dict) -> dict:
    """Обработка данных."""
    pass

# ✅ ХОРОШО: Подробное описание
@mcp.tool()
def process_customer_feedback(
    feedback_text: str,
    rating: int,
    customer_id: str
) -> dict:
    """
    Обработка отзыва клиента.
    
    Создаёт тикет в системе поддержки, отправляет
    уведомление менеджеру и обновляет рейтинг клиента.
    
    Args:
        feedback_text: Текст отзыва
        rating: Оценка от 1 до 5
        customer_id: ID клиента
    """
    pass

#3. Безопасная обработка данных

# ❌ ПЛОХО: SQL-инъекция
@mcp.tool()
def get_user(username: str) -> dict:
    return db.query(f"SELECT * FROM users WHERE name = '{username}'")

# ✅ ХОРОШО: Параметризированный запрос
@mcp.tool()
def get_user(username: str) -> dict:
    return db.query("SELECT * FROM users WHERE name = %s", (username,))

#4. Ограничение ресурсов

import time
from functools import wraps


def timeout(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = [None]
            exception = [None]
            
            def target():
                try:
                    result[0] = func(*args, **kwargs)
                except Exception as e:
                    exception[0] = e
            
            thread = threading.Thread(target=target)
            thread.daemon = True
            thread.start()
            thread.join(timeout=seconds)
            
            if thread.is_alive():
                raise TimeoutError(f"Operation timed out after {seconds}s")
            if exception[0]:
                raise exception[0]
            return result[0]
        return wrapper
    return decorator


@mcp.tool()
@timeout(30)
def long_running_operation(data: str) -> str:
    """Операция с таймаутом 30 секунд."""
    time.sleep(10)  # Имитация работы
    return "Done"

#5. Идемпотентность

# Идемпотентный инструмент (безопасно вызыватьмного раз)
@mcp.tool()
def set_user_status(user_id: str, status: str) -> str:
    """Установка статуса пользователя."""
    database.update_user_status(user_id, status)
    return f"User {user_id} status set to {status}"

# Неидемпотентный инструмент (создаёт дубликаты)
@mcp.tool()
def create_notification(user_id: str, message: str) -> str:
    """Создание уведомления (может создать дубликаты)."""
    database.create_notification(user_id, message)
    return "Notification created"

#11. Полный пример: Project Management MCP

#!/usr/bin/env python3
"""MCP сервер для управления проектами."""

from mcp.server.fastmcp import FastMCP
from typing import List, Optional
from datetime import datetime
import uuid

mcp = FastMCP("Project Management MCP")


# === Имитация базы данных ===
projects = {}
tasks = {}


# === PROJECTS ===

@mcp.tool()
def create_project(name: str, description: str = "") -> dict:
    """
    Создание нового проекта.
    
    Args:
        name: Название проекта
        description: Описание проекта
    """
    project_id = str(uuid.uuid4())
    projects[project_id] = {
        "id": project_id,
        "name": name,
        "description": description,
        "created_at": datetime.now().isoformat(),
        "status": "active"
    }
    return projects[project_id]


@mcp.tool()
def list_projects() -> List[dict]:
    """Получить список всех проектов."""
    return list(projects.values())


@mcp.tool()
def get_project(project_id: str) -> dict:
    """
    Получить информацию о проекте.
    
    Args:
        project_id: ID проекта
    """
    if project_id not in projects:
        raise ValueError(f"Project {project_id} not found")
    return projects[project_id]


@mcp.tool()
def delete_project(project_id: str) -> str:
    """
    Удалить проект.
    
    Args:
        project_id: ID проекта
    """
    if project_id not in projects:
        raise ValueError(f"Project {project_id} not found")
    del projects[project_id]
    return f"Project {project_id} deleted"


# === TASKS ===

@mcp.tool()
def create_task(
    project_id: str,
    title: str,
    description: str = "",
    assignee: str = "",
    priority: str = "medium"
) -> dict:
    """
    Создание задачи в проекте.
    
    Args:
        project_id: ID проекта
        title: Заголовок задачи
        description: Описание задачи
        assignee: Исполнитель
        priority: Приоритет (low, medium, high)
    """
    if project_id not in projects:
        raise ValueError(f"Project {project_id} not found")
    
    if priority not in ["low", "medium", "high"]:
        raise ValueError("Priority must be low, medium, or high")
    
    task_id = str(uuid.uuid4())
    tasks[task_id] = {
        "id": task_id,
        "project_id": project_id,
        "title": title,
        "description": description,
        "assignee": assignee,
        "priority": priority,
        "status": "todo",
        "created_at": datetime.now().isoformat()
    }
    return tasks[task_id]


@mcp.tool()
def list_tasks(project_id: str, status: str = None) -> List[dict]:
    """
    Получить список задач проекта.
    
    Args:
        project_id: ID проекта
        status: Фильтр по статусу (todo, in_progress, done)
    """
    project_tasks = [t for t in tasks.values() if t["project_id"] == project_id]
    
    if status:
        project_tasks = [t for t in project_tasks if t["status"] == status]
    
    return project_tasks


@mcp.tool()
def update_task_status(task_id: str, status: str) -> dict:
    """
    Обновить статус задачи.
    
    Args:
        task_id: ID задачи
        status: Новый статус (todo, in_progress, done)
    """
    if task_id not in tasks:
        raise ValueError(f"Task {task_id} not found")
    
    if status not in ["todo", "in_progress", "done"]:
        raise ValueError("Status must be todo, in_progress, or done")
    
    tasks[task_id]["status"] = status
    return tasks[task_id]


@mcp.tool()
def get_task(task_id: str) -> dict:
    """
    Получить информацию о задаче.
    
    Args:
        task_id: ID задачи
    """
    if task_id not in tasks:
        raise ValueError(f"Task {task_id} not found")
    return tasks[task_id]


# === REPORTS ===

@mcp.tool()
def get_project_summary(project_id: str) -> str:
    """
    Получить сводку по проекту.
    
    Args:
        project_id: ID проекта
    """
    if project_id not in projects:
        raise ValueError(f"Project {project_id} not found")
    
    project = projects[project_id]
    project_tasks = [t for t in tasks.values() if t["project_id"] == project_id]
    
    todo_count = len([t for t in project_tasks if t["status"] == "todo"])
    in_progress_count = len([t for t in project_tasks if t["status"] == "in_progress"])
    done_count = len([t for t in project_tasks if t["status"] == "done"])
    
    return f"""
## Project: {project['name']}

**Status:** {project['status']}
**Created:** {project['created_at']}

### Tasks Summary
| Status | Count |
|--------|-------|
| Todo | {todo_count} |
| In Progress | {in_progress_count} |
| Done | {done_count} |

**Total:** {len(project_tasks)} tasks
**Progress:** {done_count / len(project_tasks) * 100 if project_tasks else 0:.1f}%
"""


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

#Ключевые выводы

Концепт	Описание
@mcp.tool()	Декоратор для регистрации инструментов
JSON Schema	Автоматически генерируется из type hints
Валидация	Проверяйте входные параметры явно
Обработка ошибок	Выбрасывайте ValueError с понятными сообщениями
Async инструменты	Используйте async/await для I/O операций
Best practices	Одно действие, понятные описания, безопасность

Следующая тема: Prompts API — шаблоны промптов для стандартизации взаимодействия с LLM.

Характеристика

Resources

Tools

Назначение

Чтение данных

Выполнение действий

Параметры

Через URI

Через JSON Schema

Возврат

Содержимое ресурса

Результат выполнения

Side effects

Нет (только чтение)

Могут изменять данные

search_database(query, limit) → Поиск в базе знаний send_email(to, subject, body) → Отправка email create_ticket(title, desc) → Создание тикета в Jira run_analysis(data_id) → Запуск аналитики

Python тип

JSON Schema тип

Пример

str

string

name: str

int

integer

count: int

float

number

price: float

bool

boolean

active: bool

list[str]

array of strings

tags: list[str]

dict

object

metadata: dict

Optional[str]

string (nullable)

comment: Optional[str]

Концепт

Описание

@mcp.tool()

Декоратор для регистрации инструментов

JSON Schema

Автоматически генерируется из type hints

Валидация

Проверяйте входные параметры явно

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

Выбрасывайте ValueError с понятными сообщениями

Async инструменты

Используйте async/await для I/O операций

Best practices

Одно действие, понятные описания, безопасность