Создание инструментов для вызова функций. Валидация параметров и обработка результатов.
Tools — это действия MCP. В этой теме научимся создавать инструменты для вызова функций, валидации параметров и обработки результатов.
Tools (Инструменты) — функции, которые MCP сервер предоставляет для вызова через LLM. В отличие от Resources (только чтение), Tools могут выполнять действия и изменять данные.
| Характеристика | 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) → Запуск аналитики
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!"
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
}@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 resultsimport 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}"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}")@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 docimport 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")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}%
"""@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_outputimport 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()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 outputimport 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}"# ❌ ПЛОХО: Инструмент делает слишком много
@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# ❌ ПЛОХО: Бесполезное описание
@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# ❌ ПЛОХО: 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,))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"# Идемпотентный инструмент (безопасно вызыватьмного раз)
@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"#!/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.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.