wehub-resource-sync
24 KiB
24 KiB
Глубокое погружение в функции протокола MCP
Это руководство исследует расширенные функции протокола MCP, выходящие за рамки базовой обработки инструментов и ресурсов. Понимание этих функций поможет вам создавать более надежные, удобные для пользователей и готовые к производству серверы MCP.
Рассмотренные функции
- Уведомления о прогрессе — Отчет о ходе длительных операций
- Отмена запросов — Позволяет клиентам отменять текущие запросы
- Шаблоны ресурсов — Динамические URI ресурсов с параметрами
- События жизненного цикла сервера — Корректная инициализация и завершение работы
- Управление логированием — Конфигурация логирования на стороне сервера
- Шаблоны обработки ошибок — Последовательные ответы об ошибках
1. Уведомления о прогрессе
Для операций, которые требуют времени (обработка данных, загрузка файлов, вызовы API), уведомления о прогрессе информируют пользователей.
Как это работает
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (долгая операция)
Server-->>Client: уведомление: прогресс 10%
Server-->>Client: уведомление: прогресс 50%
Server-->>Client: уведомление: прогресс 90%
Server->>Client: результат (завершено)
Реализация на Python
from mcp.server import Server, NotificationOptions
from mcp.types import ProgressNotification
import asyncio
app = Server("progress-server")
@app.tool()
async def process_large_file(file_path: str, ctx) -> str:
"""Process a large file with progress updates."""
# Получить размер файла для вычисления прогресса
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Обработать часть
await process_chunk(chunk)
processed += len(chunk)
# Отправить уведомление о прогрессе
progress = (processed / file_size) * 100
await ctx.send_notification(
ProgressNotification(
progressToken=ctx.request_id,
progress=progress,
total=100,
message=f"Processing: {progress:.1f}%"
)
)
return f"Processed {file_size} bytes"
@app.tool()
async def batch_operation(items: list[str], ctx) -> str:
"""Process multiple items with progress."""
results = []
total = len(items)
for i, item in enumerate(items):
result = await process_item(item)
results.append(result)
# Сообщить о прогрессе после каждого элемента
await ctx.send_notification(
ProgressNotification(
progressToken=ctx.request_id,
progress=i + 1,
total=total,
message=f"Processed {i + 1}/{total}: {item}"
)
)
return f"Completed {total} items"
Реализация на TypeScript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
server.setRequestHandler(CallToolSchema, async (request, extra) => {
const { name, arguments: args } = request.params;
if (name === "process_data") {
const items = args.items as string[];
const results = [];
for (let i = 0; i < items.length; i++) {
const result = await processItem(items[i]);
results.push(result);
// Отправить уведомление о прогрессе
await extra.sendNotification({
method: "notifications/progress",
params: {
progressToken: request.id,
progress: i + 1,
total: items.length,
message: `Processing item ${i + 1}/${items.length}`
}
});
}
return { content: [{ type: "text", text: JSON.stringify(results) }] };
}
});
Обработка на клиенте (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Зарегистрировать обработчик
session.on_notification("notifications/progress", handle_progress)
# Вызвать инструмент (обновления прогресса будут поступать через обработчик)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Отмена запросов
Позволяет клиентам отменять запросы, которые больше не нужны или занимают слишком много времени.
Реализация на Python
from mcp.server import Server
from mcp.types import CancelledError
import asyncio
app = Server("cancellable-server")
@app.tool()
async def long_running_search(query: str, ctx) -> str:
"""Search that can be cancelled."""
results = []
try:
for page in range(100): # Поиск по многим страницам
# Проверить, был ли запрошен отказ
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Симулировать поиск по странице
page_results = await search_page(query, page)
results.extend(page_results)
# Небольшая задержка позволяет проверять отмену
await asyncio.sleep(0.1)
except CancelledError:
# Вернуть частичные результаты
return f"Cancelled. Found {len(results)} results before cancellation."
return f"Found {len(results)} total results"
@app.tool()
async def download_file(url: str, ctx) -> str:
"""Download with cancellation support."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
chunks = []
async for chunk in response.content.iter_chunked(8192):
if ctx.is_cancelled:
return f"Download cancelled at {downloaded}/{total_size} bytes"
chunks.append(chunk)
downloaded += len(chunk)
return f"Downloaded {downloaded} bytes"
Реализация контекста отмены
class CancellableContext:
"""Context object that tracks cancellation state."""
def __init__(self, request_id: str):
self.request_id = request_id
self._cancelled = asyncio.Event()
self._cancel_reason = None
@property
def is_cancelled(self) -> bool:
return self._cancelled.is_set()
def cancel(self, reason: str = "Cancelled"):
self._cancel_reason = reason
self._cancelled.set()
async def check_cancelled(self):
"""Raise if cancelled, otherwise continue."""
if self.is_cancelled:
raise CancelledError(self._cancel_reason)
async def sleep_or_cancel(self, seconds: float):
"""Sleep that can be interrupted by cancellation."""
try:
await asyncio.wait_for(
self._cancelled.wait(),
timeout=seconds
)
raise CancelledError(self._cancel_reason)
except asyncio.TimeoutError:
pass # Обычный тайм-аут, продолжаем
Отмена на стороне клиента
import asyncio
async def search_with_timeout(session, query, timeout=30):
"""Search with automatic cancellation on timeout."""
task = asyncio.create_task(
session.call_tool("long_running_search", {"query": query})
)
try:
result = await asyncio.wait_for(task, timeout=timeout)
return result
except asyncio.TimeoutError:
# Запрос на отмену
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Шаблоны ресурсов
Шаблоны ресурсов позволяют динамически формировать URI с параметрами, что полезно для API и баз данных.
Определение шаблонов
from mcp.server import Server
from mcp.types import ResourceTemplate
app = Server("template-server")
@app.list_resource_templates()
async def list_templates() -> list[ResourceTemplate]:
"""Return available resource templates."""
return [
ResourceTemplate(
uriTemplate="db://users/{user_id}",
name="User Profile",
description="Fetch user profile by ID",
mimeType="application/json"
),
ResourceTemplate(
uriTemplate="api://weather/{city}/{date}",
name="Weather Data",
description="Historical weather for city and date",
mimeType="application/json"
),
ResourceTemplate(
uriTemplate="file://{path}",
name="File Content",
description="Read file at given path",
mimeType="text/plain"
)
]
@app.read_resource()
async def read_resource(uri: str) -> str:
"""Read resource, expanding template parameters."""
# Разобрать URI для извлечения параметров
if uri.startswith("db://users/"):
user_id = uri.split("/")[-1]
return await fetch_user(user_id)
elif uri.startswith("api://weather/"):
parts = uri.replace("api://weather/", "").split("/")
city, date = parts[0], parts[1]
return await fetch_weather(city, date)
elif uri.startswith("file://"):
path = uri.replace("file://", "")
return await read_file(path)
raise ValueError(f"Unknown resource URI: {uri}")
Реализация на TypeScript
server.setRequestHandler(ListResourceTemplatesSchema, async () => {
return {
resourceTemplates: [
{
uriTemplate: "github://repos/{owner}/{repo}/issues/{issue_number}",
name: "GitHub Issue",
description: "Fetch a specific GitHub issue",
mimeType: "application/json"
},
{
uriTemplate: "db://tables/{table}/rows/{id}",
name: "Database Row",
description: "Fetch a row from a database table",
mimeType: "application/json"
}
]
};
});
server.setRequestHandler(ReadResourceSchema, async (request) => {
const uri = request.params.uri;
// Разобрать URI задачи GitHub
const githubMatch = uri.match(/^github:\/\/repos\/([^/]+)\/([^/]+)\/issues\/(\d+)$/);
if (githubMatch) {
const [_, owner, repo, issueNumber] = githubMatch;
const issue = await fetchGitHubIssue(owner, repo, parseInt(issueNumber));
return {
contents: [{
uri,
mimeType: "application/json",
text: JSON.stringify(issue, null, 2)
}]
};
}
throw new Error(`Unknown resource URI: ${uri}`);
});
4. События жизненного цикла сервера
Корректная обработка инициализации и завершения работы обеспечивает правильное управление ресурсами.
Управление жизненным циклом на Python
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Общие данные
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Запуск
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Здесь работает сервер
# Завершение работы
print("🛑 Server shutting down...")
await db_connection.close()
await cache.close()
print("✅ Resources cleaned up")
app = Server("lifecycle-server", lifespan=lifespan)
@app.tool()
async def query_database(sql: str) -> str:
"""Use the shared database connection."""
result = await db_connection.execute(sql)
return str(result)
Жизненный цикл на TypeScript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
class ManagedServer {
private server: Server;
private dbConnection: DatabaseConnection | null = null;
constructor() {
this.server = new Server({
name: "lifecycle-server",
version: "1.0.0"
});
this.setupHandlers();
}
async start() {
// Инициализировать ресурсы
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Запустить сервер
await this.server.connect(transport);
}
async stop() {
// Очистить ресурсы
console.log("🛑 Server shutting down...");
if (this.dbConnection) {
await this.dbConnection.close();
}
await this.server.close();
console.log("✅ Cleanup complete");
}
private setupHandlers() {
this.server.setRequestHandler(CallToolSchema, async (request) => {
// Безопасно использовать this.dbConnection
// ...
});
}
}
// Использование с корректным завершением работы
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Управление логированием
MCP поддерживает уровни логирования на стороне сервера, которыми могут управлять клиенты.
Реализация уровней логирования
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Преобразовать уровни MCP в уровни логирования Python
LEVEL_MAP = {
LoggingLevel.DEBUG: logging.DEBUG,
LoggingLevel.INFO: logging.INFO,
LoggingLevel.WARNING: logging.WARNING,
LoggingLevel.ERROR: logging.ERROR,
}
logger = logging.getLogger("mcp-server")
@app.set_logging_level()
async def set_logging_level(level: LoggingLevel) -> None:
"""Handle client request to change logging level."""
python_level = LEVEL_MAP.get(level, logging.INFO)
logger.setLevel(python_level)
logger.info(f"Logging level set to {level}")
@app.tool()
async def debug_operation(data: str) -> str:
"""Tool with various logging levels."""
logger.debug(f"Processing data: {data}")
try:
result = process(data)
logger.info(f"Successfully processed: {result}")
return result
except Exception as e:
logger.error(f"Processing failed: {e}")
raise
Отправка лог-сообщений клиенту
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Отправить уведомление о журнале клиенту
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Выполняется работа...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Шаблоны обработки ошибок
Единообразная обработка ошибок улучшает отладку и опыт пользователя.
Коды ошибок MCP
from mcp.types import McpError, ErrorCode
class ToolError(McpError):
"""Base class for tool errors."""
pass
class ValidationError(ToolError):
"""Invalid input parameters."""
def __init__(self, message: str):
super().__init__(ErrorCode.INVALID_PARAMS, message)
class NotFoundError(ToolError):
"""Requested resource not found."""
def __init__(self, resource: str):
super().__init__(ErrorCode.INVALID_REQUEST, f"Not found: {resource}")
class PermissionError(ToolError):
"""Access denied."""
def __init__(self, action: str):
super().__init__(ErrorCode.INVALID_REQUEST, f"Permission denied: {action}")
class InternalError(ToolError):
"""Internal server error."""
def __init__(self, message: str):
super().__init__(ErrorCode.INTERNAL_ERROR, message)
Структурированные ответы об ошибках
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Проверить ввод
if not input:
raise ValidationError("Input cannot be empty")
if len(input) > 10000:
raise ValidationError(f"Input too large: {len(input)} chars (max 10000)")
try:
# Проверить разрешения
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Выполнить операцию
result = await perform_operation(input)
if result is None:
raise NotFoundError(input)
return result
except ConnectionError as e:
raise InternalError(f"Database connection failed: {e}")
except TimeoutError as e:
raise InternalError(f"Operation timed out: {e}")
except Exception as e:
# Записать непредвиденные ошибки
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Обработка ошибок в TypeScript
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";
function validateInput(data: unknown): asserts data is ValidInput {
if (typeof data !== "object" || data === null) {
throw new McpError(
ErrorCode.InvalidParams,
"Input must be an object"
);
}
// Больше проверок...
}
server.setRequestHandler(CallToolSchema, async (request) => {
try {
validateInput(request.params.arguments);
const result = await performOperation(request.params.arguments);
return {
content: [{ type: "text", text: JSON.stringify(result) }]
};
} catch (error) {
if (error instanceof McpError) {
throw error; // Уже ошибка MCP
}
// Преобразовать другие ошибки
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Неизвестная ошибка
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Экспериментальные функции (MCP 2025-11-25)
Эти функции отмечены как экспериментальные в спецификации:
Задачи (длительные операции)
# Задачи позволяют отслеживать длительно выполняющиеся операции со состоянием
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Сообщить о начале задачи
await ctx.report_status("running", "Initializing training...")
# Цикл обучения
for epoch in range(100):
await train_epoch(model_id, data_path, epoch)
await ctx.report_status(
"running",
f"Training epoch {epoch + 1}/100",
progress=epoch + 1,
total=100
)
await ctx.report_status("completed", "Training finished")
return f"Model {model_id} trained successfully"
Аннотации инструментов
# Аннотации предоставляют метаданные о поведении инструмента
@app.tool(
annotations={
"destructive": False, # Не изменяет данные
"idempotent": True, # Безопасно повторять попытку
"timeout_seconds": 30, # Ожидаемая максимальная продолжительность
"requires_approval": False # Не требуется одобрение пользователя
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Что дальше
Дополнительные ресурсы
Отказ от ответственности:
Этот документ был переведен с помощью сервиса автоматического перевода Co-op Translator. Несмотря на наши усилия обеспечить точность, имейте в виду, что автоматический перевод может содержать ошибки или неточности. Оригинальный документ на его исходном языке следует считать авторитетным источником. Для критически важной информации рекомендуется использовать профессиональный человеческий перевод. Мы не несем ответственности за любые недоразумения или неверные толкования, возникшие в результате использования этого перевода.