Создатель MCP
Руководство по разработке серверов MCP
В навыке 9 файлов. Их можно скопировать или скачать по отдельности, сохранив указанные имена и папки.
SKILL.md
---
name: mcp-builder
description: Руководство по созданию качественных серверов MCP (Model Context Protocol), позволяющих большим языковым моделям взаимодействовать с внешними сервисами с помощью хорошо спроектированных инструментов. Используй при создании серверов MCP для интеграции внешних API или сервисов как на Python (FastMCP), так и на Node/TypeScript (MCP SDK).
license: Полные условия в LICENSE.txt
---
# Руководство по разработке серверов MCP
## Обзор
Создавай серверы MCP (Model Context Protocol), позволяющие большим языковым моделям взаимодействовать с внешними сервисами с помощью хорошо спроектированных инструментов. Качество сервера MCP определяется тем, насколько хорошо он позволяет большим языковым моделям выполнять реальные задачи.
---
# Процесс
## 🚀 Рабочий процесс в общих чертах
Создание качественного сервера MCP включает четыре основных этапа:
### Этап 1: глубокое исследование и планирование
#### 1.1 Разберись в современном проектировании MCP
**Охват API и инструменты рабочих процессов:**
Уравновешивай всесторонний охват конечных точек API со специализированными инструментами рабочих процессов. Инструменты рабочих процессов могут быть удобнее для конкретных задач, тогда как всесторонний охват даёт агентам гибкость при составлении операций. Эффективность различается в зависимости от клиента: некоторым клиентам полезно выполнение кода, объединяющего базовые инструменты, а другие лучше работают с рабочими процессами более высокого уровня. При неопределённости отдавай приоритет всестороннему охвату API.
**Именование и обнаруживаемость инструментов:**
Ясные, описательные имена инструментов помогают агентам быстро находить нужные инструменты. Используй единообразные префиксы (например, `github_create_issue`, `github_list_repos`) и именование, ориентированное на действия.
**Управление контекстом:**
Агентам полезны краткие описания инструментов и возможность фильтровать результаты и получать их постранично. Проектируй инструменты, возвращающие целевые, релевантные данные. Некоторые клиенты поддерживают выполнение кода, что может помочь агентам эффективно фильтровать и обрабатывать данные.
**Сообщения об ошибках, подсказывающие действия:**
Сообщения об ошибках должны направлять агентов к решениям с помощью конкретных предложений и дальнейших шагов.
#### 1.2 Изучи документацию протокола MCP
**Ориентируйся в спецификации MCP:**
Начни с карты сайта, чтобы найти нужные страницы: `https://modelcontextprotocol.io/sitemap.xml`
Затем загружай конкретные страницы с суффиксом `.md` для формата Markdown (например, `https://modelcontextprotocol.io/specification/draft.md`).
Ключевые страницы для изучения:
- Обзор спецификации и архитектуры
- Транспортные механизмы (потоковый HTTP, stdio)
- Определения инструментов, ресурсов и промптов
#### 1.3 Изучи документацию фреймворков
**Рекомендуемый стек:**
- **Язык**: TypeScript (высококачественная поддержка SDK и хорошая совместимость со многими средами выполнения, например MCPB. Кроме того, модели ИИ хорошо генерируют код TypeScript благодаря его широкому использованию, статической типизации и хорошим инструментам линтинга)
- **Транспорт**: потоковый HTTP для удалённых серверов с использованием JSON без состояния (проще масштабировать и сопровождать, чем сессии с состоянием и потоковые ответы). stdio для локальных серверов.
**Загрузи документацию фреймворков:**
- **Лучшие практики MCP**: [📋 Посмотреть лучшие практики](./reference/mcp_best_practices.md) — основные рекомендации
**Для TypeScript (рекомендуется):**
- **TypeScript SDK**: используй WebFetch, чтобы загрузить `https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md`
- [⚡ Руководство по TypeScript](./reference/node_mcp_server.md) — паттерны и примеры TypeScript
**Для Python:**
- **Python SDK**: используй WebFetch, чтобы загрузить `https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`
- [🐍 Руководство по Python](./reference/python_mcp_server.md) — паттерны и примеры Python
#### 1.4 Спланируй реализацию
**Разберись в API:**
Изучи документацию API сервиса, чтобы определить ключевые конечные точки, требования к аутентификации и модели данных. При необходимости используй веб-поиск и WebFetch.
**Выбор инструментов:**
Отдавай приоритет всестороннему охвату API. Перечисли конечные точки для реализации, начиная с наиболее распространённых операций.
---
### Этап 2: реализация
#### 2.1 Настрой структуру проекта
Смотри руководства для конкретных языков по настройке проекта:
- [⚡ Руководство по TypeScript](./reference/node_mcp_server.md) — структура проекта, package.json, tsconfig.json
- [🐍 Руководство по Python](./reference/python_mcp_server.md) — организация модулей, зависимости
#### 2.2 Реализуй основную инфраструктуру
Создай общие утилиты:
- API-клиент с аутентификацией
- Вспомогательные функции обработки ошибок
- Форматирование ответов (JSON/Markdown)
- Поддержка пагинации
#### 2.3 Реализуй инструменты
Для каждого инструмента:
**Схема входных данных:**
- Используй Zod (TypeScript) или Pydantic (Python)
- Включай ограничения и ясные описания
- Добавляй примеры в описания полей
**Схема выходных данных:**
- По возможности определяй `outputSchema` для структурированных данных
- Используй `structuredContent` в ответах инструментов (возможность TypeScript SDK)
- Помогает клиентам понимать и обрабатывать результаты инструментов
**Описание инструмента:**
- Краткое описание функциональности
- Описания параметров
- Схема возвращаемого типа
**Реализация:**
- Async/await для операций ввода-вывода
- Корректная обработка ошибок с сообщениями, подсказывающими действия
- Поддержка пагинации, где применимо
- Возвращай и текстовое содержимое, и структурированные данные при использовании современных SDK
**Аннотации:**
- `readOnlyHint`: true/false
- `destructiveHint`: true/false
- `idempotentHint`: true/false
- `openWorldHint`: true/false
---
### Этап 3: проверка и тестирование
#### 3.1 Качество кода
Проверь:
- Отсутствие дублирующегося кода (принцип DRY)
- Единообразную обработку ошибок
- Полное покрытие типами
- Ясные описания инструментов
#### 3.2 Сборка и тестирование
**TypeScript:**
- Запусти `npm run build` для проверки компиляции
- Протестируй с помощью MCP Inspector: `npx @modelcontextprotocol/inspector`
**Python:**
- Проверь синтаксис: `python -m py_compile your_server.py`
- Протестируй с помощью MCP Inspector
Подробные подходы к тестированию и контрольные списки качества смотри в руководствах для конкретных языков.
---
### Этап 4: создание оценочных заданий
После реализации сервера MCP создай всесторонние оценочные задания для проверки его эффективности.
**Загрузи [✅ Руководство по оценке](./reference/evaluation.md), чтобы получить полные рекомендации по оцениванию.**
#### 4.1 Пойми цель оценки
Используй оценочные задания, чтобы проверить, могут ли большие языковые модели эффективно использовать твой сервер MCP для ответов на реалистичные сложные вопросы.
#### 4.2 Создай 10 оценочных вопросов
Для создания эффективных оценочных заданий следуй процессу, описанному в руководстве по оценке:
1. **Изучение инструментов**: перечисли доступные инструменты и разберись в их возможностях
2. **Исследование содержимого**: используй операции ТОЛЬКО ДЛЯ ЧТЕНИЯ, чтобы изучить доступные данные
3. **Генерация вопросов**: создай 10 сложных реалистичных вопросов
4. **Проверка ответов**: самостоятельно реши каждый вопрос, чтобы проверить ответы
#### 4.3 Требования к оценочным заданиям
Убедись, что каждый вопрос:
- **Независимый**: не зависит от других вопросов
- **Только для чтения**: требует лишь неразрушающих операций
- **Сложный**: требует нескольких вызовов инструментов и глубокого исследования
- **Реалистичный**: основан на реальных сценариях использования, важных для людей
- **Проверяемый**: имеет один ясный ответ, который можно проверить сравнением строк
- **Стабильный**: ответ не изменится со временем
#### 4.4 Формат вывода
Создай XML-файл с такой структурой:
```xml
<evaluation>
<qa_pair>
<question>Найди обсуждения запусков моделей ИИ с кодовыми именами животных. Для одной модели нужно было определить специальное обозначение безопасности в формате ASL-X. Какое число X определяли для модели, названной в честь пятнистой дикой кошки?</question>
<answer>3</answer>
</qa_pair>
<!-- More qa_pairs... -->
</evaluation>
```
---
# Справочные файлы
## 📚 Библиотека документации
Загружай эти ресурсы по мере необходимости во время разработки:
### Основная документация MCP (загрузи сначала)
- **Протокол MCP**: начни с карты сайта `https://modelcontextprotocol.io/sitemap.xml`, затем загружай конкретные страницы с суффиксом `.md`
- [📋 Лучшие практики MCP](./reference/mcp_best_practices.md) — универсальные рекомендации по MCP, включая:
- Соглашения об именовании серверов и инструментов
- Рекомендации по формату ответа (JSON и Markdown)
- Лучшие практики пагинации
- Выбор транспорта (потоковый HTTP или stdio)
- Стандарты безопасности и обработки ошибок
### Документация SDK (загружай на этапах 1/2)
- **Python SDK**: загрузи из `https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`
- **TypeScript SDK**: загрузи из `https://raw.githubusercontent.com/modelcontextprotocol/typescript-sdk/main/README.md`
### Руководства по реализации для конкретных языков (загружай на этапе 2)
- [🐍 Руководство по реализации на Python](./reference/python_mcp_server.md) — полное руководство по Python/FastMCP с:
- Паттернами инициализации сервера
- Примерами моделей Pydantic
- Регистрацией инструментов с помощью `@mcp.tool`
- Полными работающими примерами
- Контрольным списком качества
- [⚡ Руководство по реализации на TypeScript](./reference/node_mcp_server.md) — полное руководство по TypeScript с:
- Структурой проекта
- Паттернами схем Zod
- Регистрацией инструментов с помощью `server.registerTool`
- Полными работающими примерами
- Контрольным списком качества
### Руководство по оценке (загружай на этапе 4)
- [✅ Руководство по оценке](./reference/evaluation.md) — полное руководство по созданию оценочных заданий с:
- Рекомендациями по созданию вопросов
- Стратегиями проверки ответов
- Спецификациями формата XML
- Примерами вопросов и ответов
- Запуском оценки с помощью предоставленных скриптов
reference/mcp_best_practices.md
# Лучшие практики серверов MCP
## Краткий справочник
### Именование серверов
- **Python**: `{service}_mcp` (например, `slack_mcp`)
- **Node/TypeScript**: `{service}-mcp-server` (например, `slack-mcp-server`)
### Именование инструментов
- Используй snake_case с префиксом сервиса
- Формат: `{service}_{action}_{resource}`
- Пример: `slack_send_message`, `github_create_issue`
### Форматы ответов
- Поддерживай и JSON, и Markdown
- JSON для программной обработки
- Markdown для удобства чтения человеком
### Пагинация
- Всегда учитывай параметр `limit`
- Возвращай `has_more`, `next_offset`, `total_count`
- По умолчанию — 20–50 элементов
### Транспорт
- **Потоковый HTTP**: для удалённых серверов и сценариев с несколькими клиентами
- **stdio**: для локальных интеграций и инструментов командной строки
- Избегай SSE (устарел в пользу потокового HTTP)
---
## Соглашения об именовании серверов
Следуй этим стандартизированным шаблонам именования:
**Python**: используй формат `{service}_mcp` (нижний регистр с подчёркиваниями)
- Примеры: `slack_mcp`, `github_mcp`, `jira_mcp`
**Node/TypeScript**: используй формат `{service}-mcp-server` (нижний регистр с дефисами)
- Примеры: `slack-mcp-server`, `github-mcp-server`, `jira-mcp-server`
Имя должно быть общим, описывать интегрируемый сервис, легко выводиться из описания задачи и не содержать номеров версий.
---
## Именование и проектирование инструментов
### Именование инструментов
1. **Используй snake_case**: `search_users`, `create_project`, `get_channel_info`
2. **Включай префикс сервиса**: учитывай, что твой сервер MCP может использоваться наряду с другими серверами MCP
- Используй `slack_send_message` вместо просто `send_message`
- Используй `github_create_issue` вместо просто `create_issue`
3. **Ориентируйся на действия**: начинай с глаголов (get, list, search, create и т. д.)
4. **Будь конкретен**: избегай общих имён, которые могут конфликтовать с другими серверами
### Проектирование инструментов
- Описания инструментов должны узко и однозначно описывать функциональность
- Описания должны точно соответствовать реальной функциональности
- Предоставляй аннотации инструментов (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
- Делай операции инструментов сфокусированными и атомарными
---
## Форматы ответов
Все инструменты, возвращающие данные, должны поддерживать несколько форматов:
### Формат JSON (`response_format="json"`)
- Машиночитаемые структурированные данные
- Включай все доступные поля и метаданные
- Единообразные имена и типы полей
- Используй для программной обработки
### Формат Markdown (`response_format="markdown"`, обычно по умолчанию)
- Форматированный текст, удобный для чтения человеком
- Используй заголовки, списки и форматирование для ясности
- Преобразовывай временные метки в удобочитаемый формат
- Показывай отображаемые имена с идентификаторами в скобках
- Опускай многословные метаданные
---
## Пагинация
Для инструментов, перечисляющих ресурсы:
- **Всегда учитывай параметр `limit`**
- **Реализуй пагинацию**: используй `offset` или пагинацию на основе курсора
- **Возвращай метаданные пагинации**: включай `has_more`, `next_offset`/`next_cursor`, `total_count`
- **Никогда не загружай все результаты в память**: особенно важно для больших наборов данных
- **По умолчанию используй разумные лимиты**: обычно 20–50 элементов
Пример ответа с пагинацией:
```json
{
"total": 150,
"count": 20,
"offset": 0,
"items": [...],
"has_more": true,
"next_offset": 20
}
```
---
## Варианты транспорта
### Потоковый HTTP
**Лучше всего подходит для**: удалённых серверов, веб-сервисов, сценариев с несколькими клиентами
**Характеристики**:
- Двунаправленная связь по HTTP
- Поддерживает несколько одновременных клиентов
- Может развёртываться как веб-сервис
- Позволяет отправлять уведомления от сервера клиенту
**Используй, когда**:
- Обслуживаешь нескольких клиентов одновременно
- Развёртываешь как облачный сервис
- Интегрируешь с веб-приложениями
### stdio
**Лучше всего подходит для**: локальных интеграций, инструментов командной строки
**Характеристики**:
- Связь через стандартные потоки ввода-вывода
- Простая настройка, конфигурация сети не требуется
- Запускается как подпроцесс клиента
**Используй, когда**:
- Создаёшь инструменты для локальных сред разработки
- Интегрируешь с настольными приложениями
- Работаешь со сценариями одного пользователя и одной сессии
**Примечание**: серверы stdio НЕ должны писать журналы в stdout (используй stderr для журналирования)
### Выбор транспорта
| Критерий | stdio | Потоковый HTTP |
|-----------|-------|-----------------|
| **Развёртывание** | Локальное | Удалённое |
| **Клиенты** | Один | Несколько |
| **Сложность** | Низкая | Средняя |
| **Реальное время** | Нет | Да |
---
## Лучшие практики безопасности
### Аутентификация и авторизация
**OAuth 2.1**:
- Используй безопасный OAuth 2.1 с сертификатами от признанных центров сертификации
- Проверяй токены доступа до обработки запросов
- Принимай только токены, специально предназначенные для твоего сервера
**Ключи API**:
- Храни ключи API в переменных окружения, никогда не в коде
- Проверяй ключи при запуске сервера
- Предоставляй ясные сообщения об ошибках при сбое аутентификации
### Проверка входных данных
- Очищай пути к файлам для предотвращения обхода каталогов
- Проверяй URL и внешние идентификаторы
- Проверяй размеры и диапазоны параметров
- Предотвращай внедрение команд в системные вызовы
- Используй проверку по схемам (Pydantic/Zod) для всех входных данных
### Обработка ошибок
- Не раскрывай клиентам внутренние ошибки
- Записывай связанные с безопасностью ошибки в журналы на стороне сервера
- Предоставляй полезные сообщения об ошибках, не раскрывающие лишнего
- Освобождай ресурсы после ошибок
### Защита от DNS rebinding
Для серверов потокового HTTP, работающих локально:
- Включи защиту от DNS rebinding
- Проверяй заголовок `Origin` во всех входящих соединениях
- Привязывайся к `127.0.0.1`, а не к `0.0.0.0`
---
## Аннотации инструментов
Предоставляй аннотации, помогающие клиентам понять поведение инструмента:
| Аннотация | Тип | По умолчанию | Описание |
|-----------|------|---------|-------------|
| `readOnlyHint` | boolean | false | Инструмент не изменяет свою среду |
| `destructiveHint` | boolean | true | Инструмент может выполнять разрушающие обновления |
| `idempotentHint` | boolean | false | Повторные вызовы с теми же аргументами не дают дополнительного эффекта |
| `openWorldHint` | boolean | true | Инструмент взаимодействует с внешними сущностями |
**Важно**: аннотации — это подсказки, а не гарантии безопасности. Клиенты не должны принимать критически важные для безопасности решения только на основе аннотаций.
---
## Обработка ошибок
- Используй стандартные коды ошибок JSON-RPC
- Сообщай об ошибках инструментов внутри объектов результата (а не как об ошибках уровня протокола)
- Предоставляй полезные, конкретные сообщения об ошибках с предложением дальнейших действий
- Не раскрывай внутренние детали реализации
- Корректно освобождай ресурсы при ошибках
Пример обработки ошибок:
```typescript
try {
const result = performOperation();
return { content: [{ type: "text", text: result }] };
} catch (error) {
return {
isError: true,
content: [{
type: "text",
text: `Error: ${error.message}. Try using filter='active_only' to reduce results.`
}]
};
}
```
---
## Требования к тестированию
Всестороннее тестирование должно охватывать:
- **Функциональное тестирование**: проверяй корректное выполнение с допустимыми/недопустимыми входными данными
- **Интеграционное тестирование**: проверяй взаимодействие с внешними системами
- **Тестирование безопасности**: проверяй аутентификацию, очистку входных данных, ограничение частоты запросов
- **Тестирование производительности**: проверяй поведение под нагрузкой, тайм-ауты
- **Обработку ошибок**: обеспечивай корректное сообщение об ошибках и освобождение ресурсов
---
## Требования к документации
- Предоставляй ясную документацию по всем инструментам и возможностям
- Включай работающие примеры (не менее 3 на каждую основную функцию)
- Документируй аспекты безопасности
- Указывай необходимые разрешения и уровни доступа
- Документируй ограничения частоты запросов и характеристики производительности
reference/evaluation.md
# Руководство по оценке серверов MCP
## Обзор
Этот документ содержит рекомендации по созданию всесторонних оценочных заданий для серверов MCP. Оценочные задания проверяют, могут ли большие языковые модели эффективно использовать твой сервер MCP, чтобы отвечать на реалистичные сложные вопросы с помощью только предоставленных инструментов.
---
## Краткий справочник
### Требования к оценочным заданиям
- Создай 10 понятных человеку вопросов
- Вопросы должны быть ТОЛЬКО ДЛЯ ЧТЕНИЯ, НЕЗАВИСИМЫМИ, НЕРАЗРУШАЮЩИМИ
- Каждый вопрос требует нескольких вызовов инструментов (потенциально десятков)
- Ответы должны быть одиночными проверяемыми значениями
- Ответы должны быть СТАБИЛЬНЫМИ (не меняться со временем)
### Формат вывода
```xml
<evaluation>
<qa_pair>
<question>Твой вопрос здесь</question>
<answer>Один проверяемый ответ</answer>
</qa_pair>
</evaluation>
```
---
## Цель оценочных заданий
Качество сервера MCP определяется НЕ тем, насколько хорошо или полно сервер реализует инструменты, а тем, насколько хорошо эти реализации (схемы входных/выходных данных, строки документации/описания, функциональность) позволяют большим языковым моделям, не имеющим другого контекста и располагающим доступом ТОЛЬКО к серверам MCP, отвечать на реалистичные и сложные вопросы.
## Обзор оценки
Создай 10 понятных человеку вопросов, для ответа на которые требуются ТОЛЬКО операции ЧТЕНИЯ, НЕЗАВИСИМЫЕ, НЕРАЗРУШАЮЩИЕ и ИДЕМПОТЕНТНЫЕ. Каждый вопрос должен быть:
- Реалистичным
- Ясным и кратким
- Однозначным
- Сложным, потенциально требующим десятков вызовов инструментов или шагов
- Допускающим ответ в виде одного проверяемого значения, которое ты определяешь заранее
## Рекомендации по вопросам
### Основные требования
1. **Вопросы ДОЛЖНЫ быть независимыми**
- Каждый вопрос НЕ должен зависеть от ответа на любой другой вопрос
- Не должен предполагать предыдущих операций записи при обработке другого вопроса
2. **Вопросы ДОЛЖНЫ требовать ТОЛЬКО НЕРАЗРУШАЮЩЕГО И ИДЕМПОТЕНТНОГО использования инструментов**
- Не должны предписывать или требовать изменения состояния для получения правильного ответа
3. **Вопросы должны быть РЕАЛИСТИЧНЫМИ, ЯСНЫМИ, КРАТКИМИ и СЛОЖНЫМИ**
- Должны требовать от другой большой языковой модели использования нескольких (потенциально десятков) инструментов или шагов для ответа
### Сложность и глубина
4. **Вопросы должны требовать глубокого исследования**
- Рассматривай многошаговые вопросы, требующие нескольких подвопросов и последовательных вызовов инструментов
- Каждый шаг должен использовать пользу информации, найденной в предыдущих вопросах
5. **Вопросы могут требовать обширного постраничного просмотра**
- Может понадобиться просмотр нескольких страниц результатов
- Может потребоваться обращение к старым данным (устаревшим на 1–2 года), чтобы найти узкоспециализированную информацию
- Вопросы должны быть ТРУДНЫМИ
6. **Вопросы должны требовать глубокого понимания**
- А не поверхностных знаний
- Можно формулировать сложные идеи как вопросы «True/False», требующие доказательств
- Можно использовать формат выбора ответа, в котором большая языковая модель должна исследовать разные гипотезы
7. **Вопросы не должны решаться простым поиском по ключевым словам**
- Не включай конкретные ключевые слова из искомого содержимого
- Используй синонимы, связанные понятия или перефразирование
- Требуй нескольких поисков, анализа нескольких связанных элементов, извлечения контекста, а затем вывода ответа
### Тестирование инструментов
8. **Вопросы должны подвергать возвращаемые инструментами значения стресс-тестированию**
- Могут вызывать возврат инструментами больших JSON-объектов или списков, перегружая большую языковую модель
- Должны требовать понимания нескольких видов данных:
- Идентификаторы и имена
- Временные метки и даты со временем (месяцы, дни, годы, секунды)
- Идентификаторы файлов, имена, расширения и MIME-типы
- URL, GID и т. д.
- Должны проверять способность инструмента возвращать все полезные формы данных
9. **Вопросы должны В ОСНОВНОМ отражать реальные сценарии использования людьми**
- Те виды задач поиска информации, которые важны ЛЮДЯМ, работающим с помощью большой языковой модели
10. **Вопросы могут требовать десятков вызовов инструментов**
- Это создаёт трудности для больших языковых моделей с ограниченным контекстом
- Побуждает инструменты сервера MCP сокращать возвращаемую информацию
11. **Включай неоднозначные вопросы**
- Они могут быть неоднозначными ИЛИ требовать трудных решений о том, какие инструменты вызывать
- Заставляй большую языковую модель потенциально ошибаться или неверно интерпретировать
- Обеспечь, чтобы, несмотря на НЕОДНОЗНАЧНОСТЬ, ПО-ПРЕЖНЕМУ СУЩЕСТВОВАЛ ЕДИНСТВЕННЫЙ ПРОВЕРЯЕМЫЙ ОТВЕТ
### Стабильность
12. **Вопросы должны быть спроектированы так, чтобы ответ НЕ МЕНЯЛСЯ**
- Не задавай вопросы, основанные на «текущем состоянии», которое динамично
- Например, не подсчитывай:
- Количество реакций на публикацию
- Количество ответов в ветке
- Количество участников канала
13. **НЕ позволяй серверу MCP ОГРАНИЧИВАТЬ виды создаваемых вопросов**
- Создавай трудные и сложные вопросы
- Некоторые из них могут быть неразрешимы доступными инструментами сервера MCP
- Вопросы могут требовать конкретных форматов вывода (дата и время или время эпохи, JSON или MARKDOWN)
- Для решения вопросов могут потребоваться десятки вызовов инструментов
## Рекомендации по ответам
### Проверка
1. **Ответы должны быть ПРОВЕРЯЕМЫМИ прямым сравнением строк**
- Если ответ можно записать в разных форматах, ясно укажи формат вывода в ВОПРОСЕ
- Примеры: «Используй YYYY/MM/DD», «Ответь True или False», «Ответь A, B, C или D и ничего больше».
- Ответ должен быть одним ПРОВЕРЯЕМЫМ значением, например:
- Идентификатор пользователя, имя пользователя, отображаемое имя, личное имя, фамилия
- Идентификатор канала, имя канала
- Идентификатор сообщения, строка
- URL, заголовок
- Числовая величина
- Временная метка, дата и время
- Логическое значение (для вопросов True/False)
- Адрес электронной почты, номер телефона
- Идентификатор файла, имя файла, расширение файла
- Ответ с выбором варианта
- Ответы не должны требовать специального форматирования или сложного структурированного вывода
- Ответ будет проверяться ПРЯМЫМ СРАВНЕНИЕМ СТРОК
### Удобочитаемость
2. **В ответах обычно следует предпочитать УДОБОЧИТАЕМЫЕ ДЛЯ ЧЕЛОВЕКА форматы**
- Примеры: имена, личное имя, фамилия, дата и время, имя файла, строка сообщения, URL, yes/no, true/false, a/b/c/d
- Вместо непрозрачных идентификаторов (хотя идентификаторы допустимы)
- ПОДАВЛЯЮЩЕЕ БОЛЬШИНСТВО ответов должно быть удобно для чтения человеком
### Стабильность
3. **Ответы должны быть СТАБИЛЬНЫМИ/НЕИЗМЕННЫМИ**
- Изучай старое содержимое (например, завершившиеся разговоры, запущенные проекты, вопросы с полученными ответами)
- Создавай ВОПРОСЫ на основе «закрытых» понятий, для которых ответ всегда будет одним и тем же
- Вопросы могут требовать рассмотрения фиксированного временного окна, чтобы исключить изменчивость ответов
- Опирайся на контекст, который ВРЯД ЛИ изменится
- Пример: если нужно найти название научной работы, будь ДОСТАТОЧНО КОНКРЕТЕН, чтобы ответ нельзя было спутать с работами, опубликованными позже
4. **Ответы должны быть ЯСНЫМИ и ОДНОЗНАЧНЫМИ**
- Вопросы должны быть спроектированы так, чтобы существовал единственный ясный ответ
- Ответ можно получить с помощью инструментов сервера MCP
### Разнообразие
5. **Ответы должны быть РАЗНООБРАЗНЫМИ**
- Ответ должен быть одним ПРОВЕРЯЕМЫМ значением разных видов и форматов
- Понятие пользователя: идентификатор пользователя, имя пользователя, отображаемое имя, личное имя, фамилия, адрес электронной почты, номер телефона
- Понятие канала: идентификатор канала, имя канала, тема канала
- Понятие сообщения: идентификатор сообщения, строка сообщения, временная метка, месяц, день, год
6. **Ответы НЕ должны быть сложными структурами**
- Не список значений
- Не сложный объект
- Не список идентификаторов или строк
- Не текст на естественном языке
- ЕСЛИ ТОЛЬКО ответ нельзя однозначно проверить ПРЯМЫМ СРАВНЕНИЕМ СТРОК
- И реалистично воспроизвести
- Должно быть маловероятно, что большая языковая модель вернёт тот же список в другом порядке или формате
## Процесс оценки
### Шаг 1: изучение документации
Прочитай документацию целевого API, чтобы понять:
- Доступные конечные точки и функциональность
- При наличии неоднозначности получи дополнительную информацию из интернета
- Распараллель этот шаг НАСКОЛЬКО ВОЗМОЖНО
- Убедись, что каждый субагент ТОЛЬКО изучает документацию из файловой системы или интернета
### Шаг 2: изучение инструментов
Перечисли инструменты, доступные на сервере MCP:
- Изучи сервер MCP напрямую
- Разберись в схемах входных/выходных данных, строках документации и описаниях
- БЕЗ вызова самих инструментов на этом этапе
### Шаг 3: углубление понимания
Повторяй шаги 1 и 2, пока не получишь хорошее понимание:
- Выполни несколько итераций
- Подумай о видах задач, которые хочешь создать
- Уточняй своё понимание
- НИ НА КАКОМ этапе не следует ЧИТАТЬ код самой реализации сервера MCP
- Используй интуицию и понимание, чтобы создавать разумные, реалистичные, но ОЧЕНЬ трудные задачи
### Шаг 4: изучение содержимого только для чтения
Разобравшись в API и инструментах, ИСПОЛЬЗУЙ инструменты сервера MCP:
- Изучай содержимое ТОЛЬКО с помощью операций ЧТЕНИЯ и НЕРАЗРУШАЮЩИХ операций
- Цель: выявить конкретное содержимое (например, пользователей, каналы, сообщения, проекты, задачи) для создания реалистичных вопросов
- НЕ следует вызывать инструменты, изменяющие состояние
- НЕ читать код самой реализации сервера MCP
- Распараллель этот шаг, направляя отдельных субагентов на независимые исследования
- Убедись, что каждый субагент выполняет только операции ЧТЕНИЯ, НЕРАЗРУШАЮЩИЕ и ИДЕМПОТЕНТНЫЕ операции
- БУДЬ ОСТОРОЖЕН: НЕКОТОРЫЕ ИНСТРУМЕНТЫ могут возвращать МНОГО ДАННЫХ, из-за чего закончится КОНТЕКСТ
- Для исследования делай ПОСТЕПЕННЫЕ, НЕБОЛЬШИЕ И ЦЕЛЕВЫЕ вызовы инструментов
- Во всех запросах вызова инструментов используй параметр `limit` для ограничения результатов (<10)
- Используй пагинацию
### Шаг 5: генерация задач
После изучения содержимого создай 10 понятных человеку вопросов:
- Большая языковая модель должна иметь возможность ответить на них с помощью сервера MCP
- Следуй всем приведённым выше рекомендациям по вопросам и ответам
## Формат вывода
Каждая пара QA состоит из вопроса и ответа. Результат должен быть XML-файлом со следующей структурой:
```xml
<evaluation>
<qa_pair>
<question>Найди созданный во втором квартале 2024 года проект с наибольшим числом завершённых задач. Как называется проект?</question>
<answer>Website Redesign</answer>
</qa_pair>
<qa_pair>
<question>Найди задачи с меткой "bug", закрытые в марте 2024 года. Какой пользователь закрыл больше всего задач? Укажи его имя пользователя.</question>
<answer>sarah_dev</answer>
</qa_pair>
<qa_pair>
<question>Найди запросы на включение изменений, которые изменяли файлы в каталоге /api и были объединены в период с 1 по 31 января 2024 года. Сколько разных участников работало над этими PR?</question>
<answer>7</answer>
</qa_pair>
<qa_pair>
<question>Найди репозиторий с наибольшим числом звёзд, созданный до 2023 года. Как называется репозиторий?</question>
<answer>data-pipeline</answer>
</qa_pair>
</evaluation>
```
## Примеры оценочных заданий
### Хорошие вопросы
**Пример 1: многошаговый вопрос, требующий глубокого исследования (GitHub MCP)**
```xml
<qa_pair>
<question>Найди репозиторий, который был архивирован в третьем квартале 2023 года и до этого был проектом с наибольшим числом форков в организации. Какой основной язык программирования использовался в этом репозитории?</question>
<answer>Python</answer>
</qa_pair>
```
Этот вопрос хорош, потому что:
- Требует нескольких поисков для обнаружения архивированных репозиториев
- Нужно определить, у какого из них было больше всего форков до архивирования
- Требует изучения сведений о репозитории для определения языка
- Ответ — простое проверяемое значение
- Основан на исторических («закрытых») данных, которые не изменятся
**Пример 2: требует понимания контекста без совпадения ключевых слов (MCP для управления проектами)**
```xml
<qa_pair>
<question>Найди инициативу, направленную на улучшение первоначального знакомства клиентов с продуктом и завершённую в конце 2023 года. После завершения руководитель проекта создал ретроспективный документ. Как называлась должность руководителя на тот момент?</question>
<answer>Product Manager</answer>
</qa_pair>
```
Этот вопрос хорош, потому что:
- Не использует конкретное название проекта («инициатива, направленная на улучшение первоначального знакомства клиентов с продуктом»)
- Требует найти завершённые проекты за определённый период
- Нужно определить руководителя проекта и его должность
- Требует понимания контекста из ретроспективных документов
- Ответ удобочитаем для человека и стабилен
- Основан на завершённой работе (не изменится)
**Пример 3: сложное агрегирование, требующее нескольких шагов (MCP для трекера задач)**
```xml
<qa_pair>
<question>Среди всех ошибок, о которых сообщили в январе 2024 года и которым назначили критический приоритет, какой исполнитель устранил наибольшую долю назначенных ему ошибок в течение 48 часов? Укажи имя пользователя исполнителя.</question>
<answer>alex_eng</answer>
</qa_pair>
```
Этот вопрос хорош, потому что:
- Требует фильтрации ошибок по дате, приоритету и статусу
- Нужно сгруппировать по исполнителю и рассчитать доли устранённых ошибок
- Требует понимания временных меток для определения 48-часовых окон
- Проверяет пагинацию (потенциально нужно обработать много ошибок)
- Ответ — одно имя пользователя
- Основан на исторических данных за определённый период
**Пример 4: требует обобщения нескольких типов данных (CRM MCP)**
```xml
<qa_pair>
<question>Найди клиентский аккаунт, который перешёл с тарифа Starter на Enterprise в четвёртом квартале 2023 года и имел наибольшую годовую стоимость контракта. В какой отрасли работает этот клиент?</question>
<answer>Healthcare</answer>
</qa_pair>
```
Этот вопрос хорош, потому что:
- Требует понимания изменений уровня подписки
- Нужно выявить события перехода на более высокий тариф за определённый период
- Требует сравнения стоимости контрактов
- Необходимо получить сведения об отрасли клиента
- Ответ прост и проверяем
- Основан на завершённых исторических операциях
### Плохие вопросы
**Пример 1: ответ меняется со временем**
```xml
<qa_pair>
<question>Сколько открытых задач сейчас назначено команде разработки?</question>
<answer>47</answer>
</qa_pair>
```
Этот вопрос плох, потому что:
- Ответ будет меняться по мере создания, закрытия или переназначения задач
- Не основан на стабильных/неизменных данных
- Зависит от «текущего состояния», которое динамично
**Пример 2: слишком легко решается поиском по ключевым словам**
```xml
<qa_pair>
<question>Найди запрос на включение изменений с заголовком "Add authentication feature" и скажи, кто его создал.</question>
<answer>developer123</answer>
</qa_pair>
```
Этот вопрос плох, потому что:
- Может решаться простым поиском по ключевым словам точного заголовка
- Не требует глубокого исследования или понимания
- Не требует обобщения или анализа
**Пример 3: неоднозначный формат ответа**
```xml
<qa_pair>
<question>Перечисли все репозитории, в которых Python является основным языком.</question>
<answer>repo1, repo2, repo3, data-pipeline, ml-tools</answer>
</qa_pair>
```
Этот вопрос плох, потому что:
- Ответ — список, который можно вернуть в любом порядке
- Его трудно проверить прямым сравнением строк
- Большая языковая модель может оформить его иначе (массив JSON, разделение запятыми, разделение переносами строк)
- Лучше запросить конкретное агрегированное значение (количество) или максимум (больше всего звёзд)
## Процесс проверки
После создания оценочных заданий:
1. **Изучи XML-файл**, чтобы понять схему
2. **Загрузи инструкцию каждой задачи** и параллельно с помощью сервера MCP и инструментов определи правильный ответ, пытаясь решить задачу САМОСТОЯТЕЛЬНО
3. **Отметь любые операции**, требующие ЗАПИСИ или РАЗРУШАЮЩИХ действий
4. **Собери все ПРАВИЛЬНЫЕ ответы** и замени все неправильные ответы в документе
5. **Удали любые `<qa_pair>`**, требующие ЗАПИСИ или РАЗРУШАЮЩИХ действий
Не забудь распараллелить решение задач, чтобы не исчерпать контекст, затем собери все ответы и в конце внеси изменения в файл.
## Советы по созданию качественных оценочных заданий
1. **Тщательно обдумай и заранее спланируй** перед генерацией задач
2. **Распараллеливай при возможности**, чтобы ускорить процесс и управлять контекстом
3. **Сосредоточься на реалистичных сценариях использования**, которые люди действительно хотели бы выполнить
4. **Создавай трудные вопросы**, проверяющие пределы возможностей сервера MCP
5. **Обеспечивай стабильность**, используя исторические данные и закрытые понятия
6. **Проверяй ответы**, самостоятельно решая вопросы с помощью инструментов сервера MCP
7. **Повторяй и уточняй** на основе того, что узнаёшь в процессе
---
# Запуск оценки
После создания файла оценочных заданий можно использовать предоставленную тестовую обвязку для проверки сервера MCP.
## Настройка
1. **Установи зависимости**
```bash
pip install -r scripts/requirements.txt
```
Или установи вручную:
```bash
pip install anthropic mcp
```
2. **Задай ключ API**
```bash
export ANTHROPIC_API_KEY=your_api_key_here
```
## Формат файла оценочных заданий
Файлы оценочных заданий используют формат XML с элементами `<qa_pair>`:
```xml
<evaluation>
<qa_pair>
<question>Найди созданный во втором квартале 2024 года проект с наибольшим числом завершённых задач. Как называется проект?</question>
<answer>Website Redesign</answer>
</qa_pair>
<qa_pair>
<question>Найди задачи с меткой "bug", закрытые в марте 2024 года. Какой пользователь закрыл больше всего задач? Укажи его имя пользователя.</question>
<answer>sarah_dev</answer>
</qa_pair>
</evaluation>
```
## Запуск оценки
Скрипт оценки (`scripts/evaluation.py`) поддерживает три типа транспорта:
**Важно:**
- **Транспорт stdio**: скрипт оценки автоматически запускает процесс сервера MCP и управляет им за тебя. Не запускай сервер вручную.
- **Транспорты sse/http**: перед запуском оценки необходимо отдельно запустить сервер MCP. Скрипт подключается к уже работающему серверу по указанному URL.
### 1. Локальный сервер STDIO
Для локально работающих серверов MCP (скрипт запускает сервер автоматически):
```bash
python scripts/evaluation.py \
-t stdio \
-c python \
-a my_mcp_server.py \
evaluation.xml
```
С переменными окружения:
```bash
python scripts/evaluation.py \
-t stdio \
-c python \
-a my_mcp_server.py \
-e API_KEY=abc123 \
-e DEBUG=true \
evaluation.xml
```
### 2. Server-Sent Events (SSE)
Для серверов MCP на основе SSE (сначала необходимо запустить сервер):
```bash
python scripts/evaluation.py \
-t sse \
-u https://example.com/mcp \
-H "Authorization: Bearer token123" \
-H "X-Custom-Header: value" \
evaluation.xml
```
### 3. HTTP (потоковый HTTP)
Для серверов MCP на основе HTTP (сначала необходимо запустить сервер):
```bash
python scripts/evaluation.py \
-t http \
-u https://example.com/mcp \
-H "Authorization: Bearer token123" \
evaluation.xml
```
## Параметры командной строки
```
usage: evaluation.py [-h] [-t {stdio,sse,http}] [-m MODEL] [-c COMMAND]
[-a ARGS [ARGS ...]] [-e ENV [ENV ...]] [-u URL]
[-H HEADERS [HEADERS ...]] [-o OUTPUT]
eval_file
positional arguments:
eval_file Path to evaluation XML file
optional arguments:
-h, --help Show help message
-t, --transport Transport type: stdio, sse, or http (default: stdio)
-m, --model Claude model to use (default: claude-3-7-sonnet-20250219)
-o, --output Output file for report (default: print to stdout)
stdio options:
-c, --command Command to run MCP server (e.g., python, node)
-a, --args Arguments for the command (e.g., server.py)
-e, --env Environment variables in KEY=VALUE format
sse/http options:
-u, --url MCP server URL
-H, --header HTTP headers in 'Key: Value' format
```
## Вывод
Скрипт оценки создаёт подробный отчёт, включающий:
- **Сводную статистику**:
- Точность (правильных/всего)
- Среднюю продолжительность задачи
- Среднее число вызовов инструментов на задачу
- Общее число вызовов инструментов
- **Результаты по каждой задаче**:
- Промпт и ожидаемый ответ
- Фактический ответ агента
- Был ли ответ правильным (✅/❌)
- Продолжительность и сведения о вызовах инструментов
- Краткое изложение агентом своего подхода
- Отзыв агента об инструментах
### Сохрани отчёт в файл
```bash
python scripts/evaluation.py \
-t stdio \
-c python \
-a my_server.py \
-o evaluation_report.md \
evaluation.xml
```
## Полный пример рабочего процесса
Вот полный пример создания и запуска оценки:
1. **Создай свой файл оценочных заданий** (`my_evaluation.xml`):
```xml
<evaluation>
<qa_pair>
<question>Найди пользователя, создавшего больше всего задач в январе 2024 года. Каково его имя пользователя?</question>
<answer>alice_developer</answer>
</qa_pair>
<qa_pair>
<question>Среди всех запросов на включение изменений, объединённых в первом квартале 2024 года, в каком репозитории их было больше всего? Укажи имя репозитория.</question>
<answer>backend-api</answer>
</qa_pair>
<qa_pair>
<question>Найди проект, который был завершён в декабре 2023 года и имел наибольшую продолжительность от начала до конца. Сколько дней он занял?</question>
<answer>127</answer>
</qa_pair>
</evaluation>
```
2. **Установи зависимости**:
```bash
pip install -r scripts/requirements.txt
export ANTHROPIC_API_KEY=your_api_key
```
3. **Запусти оценку**:
```bash
python scripts/evaluation.py \
-t stdio \
-c python \
-a github_mcp_server.py \
-e GITHUB_TOKEN=ghp_xxx \
-o github_eval_report.md \
my_evaluation.xml
```
4. **Изучи отчёт** в `github_eval_report.md`, чтобы:
- Увидеть, какие вопросы пройдены/не пройдены
- Прочитать отзыв агента о твоих инструментах
- Выявить области для улучшения
- Доработать проектирование сервера MCP
## Устранение неполадок
### Ошибки соединения
При ошибках соединения:
- **STDIO**: проверь правильность команды и аргументов
- **SSE/HTTP**: проверь доступность URL и правильность заголовков
- Убедись, что необходимые ключи API заданы в переменных окружения или заголовках
### Низкая точность
Если многие оценочные задания не проходят:
- Изучи отзыв агента по каждой задаче
- Проверь, ясны ли и полны ли описания инструментов
- Убедись, что входные параметры хорошо документированы
- Подумай, не возвращают ли инструменты слишком много или слишком мало данных
- Убедись, что сообщения об ошибках подсказывают конкретные действия
### Проблемы с тайм-аутами
Если время выполнения задач истекает:
- Используй более способную модель (например, `claude-3-7-sonnet-20250219`)
- Проверь, не возвращают ли инструменты слишком много данных
- Убедись, что пагинация работает корректно
- Рассмотри упрощение сложных вопросов
reference/node_mcp_server.md
# Руководство по реализации сервера MCP на Node/TypeScript
## Обзор
Этот документ содержит лучшие практики и примеры для Node/TypeScript по реализации серверов MCP с помощью MCP TypeScript SDK. Он охватывает структуру проекта, настройку сервера, паттерны регистрации инструментов, проверку входных данных с помощью Zod, обработку ошибок и полные работающие примеры.
---
## Краткий справочник
### Основные импорты
```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import express from "express";
import { z } from "zod";
```
### Инициализация сервера
```typescript
const server = new McpServer({
name: "service-mcp-server",
version: "1.0.0"
});
```
### Паттерн регистрации инструментов
```typescript
server.registerTool(
"tool_name",
{
title: "Tool Display Name",
description: "What the tool does",
inputSchema: { param: z.string() },
outputSchema: { result: z.string() }
},
async ({ param }) => {
const output = { result: `Processed: ${param}` };
return {
content: [{ type: "text", text: JSON.stringify(output) }],
structuredContent: output // Modern pattern for structured data
};
}
);
```
---
## MCP TypeScript SDK
Официальный MCP TypeScript SDK предоставляет:
- Класс `McpServer` для инициализации сервера
- Метод `registerTool` для регистрации инструментов
- Интеграцию схем Zod для проверки входных данных во время выполнения
- Типобезопасные реализации обработчиков инструментов
**ВАЖНО — используй только современные API:**
- **ИСПОЛЬЗУЙ**: `server.registerTool()`, `server.registerResource()`, `server.registerPrompt()`
- **НЕ ИСПОЛЬЗУЙ**: старые устаревшие API, такие как `server.tool()`, `server.setRequestHandler(ListToolsRequestSchema, ...)`, или ручную регистрацию обработчиков
- Методы `register*` обеспечивают лучшую типобезопасность, автоматическую обработку схем и являются рекомендуемым подходом
Полные сведения смотри в документации MCP SDK в справочных материалах.
## Соглашение об именовании серверов
Серверы MCP на Node/TypeScript должны следовать такому шаблону именования:
- **Формат**: `{service}-mcp-server` (нижний регистр с дефисами)
- **Примеры**: `github-mcp-server`, `jira-mcp-server`, `stripe-mcp-server`
Имя должно быть:
- Общим (не привязанным к конкретным функциям)
- Описывающим интегрируемый сервис/API
- Легко выводимым из описания задачи
- Без номеров версий или дат
## Структура проекта
Создай следующую структуру для серверов MCP на Node/TypeScript:
```
{service}-mcp-server/
├── package.json
├── tsconfig.json
├── README.md
├── src/
│ ├── index.ts # Основная точка входа с инициализацией McpServer
│ ├── types.ts # Определения типов и интерфейсы TypeScript
│ ├── tools/ # Реализации инструментов (один файл на предметную область)
│ ├── services/ # API-клиенты и общие утилиты
│ ├── schemas/ # Схемы проверки Zod
│ └── constants.ts # Общие константы (API_URL, CHARACTER_LIMIT и т. д.)
└── dist/ # Собранные JavaScript-файлы (точка входа: dist/index.js)
```
## Реализация инструментов
### Именование инструментов
Используй snake_case для имён инструментов (например, "search_users", "create_project", "get_channel_info") с ясными именами, ориентированными на действия.
**Избегай конфликтов имён**: включай контекст сервиса, чтобы предотвратить совпадения:
- Используй "slack_send_message" вместо просто "send_message"
- Используй "github_create_issue" вместо просто "create_issue"
- Используй "asana_list_tasks" вместо просто "list_tasks"
### Структура инструмента
Инструменты регистрируются методом `registerTool` со следующими требованиями:
- Используй схемы Zod для проверки входных данных во время выполнения и типобезопасности
- Поле `description` должно быть задано явно — комментарии JSDoc НЕ извлекаются автоматически
- Явно задавай `title`, `description`, `inputSchema` и `annotations`
- `inputSchema` должна быть объектом схемы Zod (а не JSON-схемой)
- Явно типизируй все параметры и возвращаемые значения
```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
const server = new McpServer({
name: "example-mcp",
version: "1.0.0"
});
// Zod schema for input validation
const UserSearchInputSchema = z.object({
query: z.string()
.min(2, "Query must be at least 2 characters")
.max(200, "Query must not exceed 200 characters")
.describe("Search string to match against names/emails"),
limit: z.number()
.int()
.min(1)
.max(100)
.default(20)
.describe("Maximum results to return"),
offset: z.number()
.int()
.min(0)
.default(0)
.describe("Number of results to skip for pagination"),
response_format: z.nativeEnum(ResponseFormat)
.default(ResponseFormat.MARKDOWN)
.describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();
// Type definition from Zod schema
type UserSearchInput = z.infer<typeof UserSearchInputSchema>;
server.registerTool(
"example_search_users",
{
title: "Search Example Users",
description: `Search for users in the Example system by name, email, or team.
This tool searches across all user profiles in the Example platform, supporting partial matches and various search filters. It does NOT create or modify users, only searches existing ones.
Args:
- query (string): Search string to match against names/emails
- limit (number): Maximum results to return, between 1-100 (default: 20)
- offset (number): Number of results to skip for pagination (default: 0)
- response_format ('markdown' | 'json'): Output format (default: 'markdown')
Returns:
For JSON format: Structured data with schema:
{
"total": number, // Total number of matches found
"count": number, // Number of results in this response
"offset": number, // Current pagination offset
"users": [
{
"id": string, // User ID (e.g., "U123456789")
"name": string, // Full name (e.g., "John Doe")
"email": string, // Email address
"team": string, // Team name (optional)
"active": boolean // Whether user is active
}
],
"has_more": boolean, // Whether more results are available
"next_offset": number // Offset for next page (if has_more is true)
}
Examples:
- Use when: "Find all marketing team members" -> params with query="team:marketing"
- Use when: "Search for John's account" -> params with query="john"
- Don't use when: You need to create a user (use example_create_user instead)
Error Handling:
- Returns "Error: Rate limit exceeded" if too many requests (429 status)
- Returns "No users found matching '<query>'" if search returns empty`,
inputSchema: UserSearchInputSchema,
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: true
}
},
async (params: UserSearchInput) => {
try {
// Input validation is handled by Zod schema
// Make API request using validated parameters
const data = await makeApiRequest<any>(
"users/search",
"GET",
undefined,
{
q: params.query,
limit: params.limit,
offset: params.offset
}
);
const users = data.users || [];
const total = data.total || 0;
if (!users.length) {
return {
content: [{
type: "text",
text: `No users found matching '${params.query}'`
}]
};
}
// Prepare structured output
const output = {
total,
count: users.length,
offset: params.offset,
users: users.map((user: any) => ({
id: user.id,
name: user.name,
email: user.email,
...(user.team ? { team: user.team } : {}),
active: user.active ?? true
})),
has_more: total > params.offset + users.length,
...(total > params.offset + users.length ? {
next_offset: params.offset + users.length
} : {})
};
// Format text representation based on requested format
let textContent: string;
if (params.response_format === ResponseFormat.MARKDOWN) {
const lines = [`# User Search Results: '${params.query}'`, "",
`Found ${total} users (showing ${users.length})`, ""];
for (const user of users) {
lines.push(`## ${user.name} (${user.id})`);
lines.push(`- **Email**: ${user.email}`);
if (user.team) lines.push(`- **Team**: ${user.team}`);
lines.push("");
}
textContent = lines.join("\n");
} else {
textContent = JSON.stringify(output, null, 2);
}
return {
content: [{ type: "text", text: textContent }],
structuredContent: output // Modern pattern for structured data
};
} catch (error) {
return {
content: [{
type: "text",
text: handleApiError(error)
}]
};
}
}
);
```
## Схемы Zod для проверки входных данных
Zod обеспечивает проверку типов во время выполнения:
```typescript
import { z } from "zod";
// Basic schema with validation
const CreateUserSchema = z.object({
name: z.string()
.min(1, "Name is required")
.max(100, "Name must not exceed 100 characters"),
email: z.string()
.email("Invalid email format"),
age: z.number()
.int("Age must be a whole number")
.min(0, "Age cannot be negative")
.max(150, "Age cannot be greater than 150")
}).strict(); // Use .strict() to forbid extra fields
// Enums
enum ResponseFormat {
MARKDOWN = "markdown",
JSON = "json"
}
const SearchSchema = z.object({
response_format: z.nativeEnum(ResponseFormat)
.default(ResponseFormat.MARKDOWN)
.describe("Output format")
});
// Optional fields with defaults
const PaginationSchema = z.object({
limit: z.number()
.int()
.min(1)
.max(100)
.default(20)
.describe("Maximum results to return"),
offset: z.number()
.int()
.min(0)
.default(0)
.describe("Number of results to skip")
});
```
## Варианты формата ответа
Для гибкости поддерживай несколько форматов вывода:
```typescript
enum ResponseFormat {
MARKDOWN = "markdown",
JSON = "json"
}
const inputSchema = z.object({
query: z.string(),
response_format: z.nativeEnum(ResponseFormat)
.default(ResponseFormat.MARKDOWN)
.describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
});
```
**Формат Markdown**:
- Используй заголовки, списки и форматирование для ясности
- Преобразовывай временные метки в удобочитаемый формат
- Показывай отображаемые имена с идентификаторами в скобках
- Опускай многословные метаданные
- Логически группируй связанную информацию
**Формат JSON**:
- Возвращай полные структурированные данные, подходящие для программной обработки
- Включай все доступные поля и метаданные
- Используй единообразные имена и типы полей
## Реализация пагинации
Для инструментов, перечисляющих ресурсы:
```typescript
const ListSchema = z.object({
limit: z.number().int().min(1).max(100).default(20),
offset: z.number().int().min(0).default(0)
});
async function listItems(params: z.infer<typeof ListSchema>) {
const data = await apiRequest(params.limit, params.offset);
const response = {
total: data.total,
count: data.items.length,
offset: params.offset,
items: data.items,
has_more: data.total > params.offset + data.items.length,
next_offset: data.total > params.offset + data.items.length
? params.offset + data.items.length
: undefined
};
return JSON.stringify(response, null, 2);
}
```
## Ограничения количества символов и усечение
Добавь константу CHARACTER_LIMIT, чтобы предотвратить чрезмерно большие ответы:
```typescript
// At module level in constants.ts
export const CHARACTER_LIMIT = 25000; // Maximum response size in characters
async function searchTool(params: SearchInput) {
let result = generateResponse(data);
// Check character limit and truncate if needed
if (result.length > CHARACTER_LIMIT) {
const truncatedData = data.slice(0, Math.max(1, data.length / 2));
response.data = truncatedData;
response.truncated = true;
response.truncation_message =
`Response truncated from ${data.length} to ${truncatedData.length} items. ` +
`Use 'offset' parameter or add filters to see more results.`;
result = JSON.stringify(response, null, 2);
}
return result;
}
```
## Обработка ошибок
Предоставляй ясные сообщения об ошибках, подсказывающие действия:
```typescript
import axios, { AxiosError } from "axios";
function handleApiError(error: unknown): string {
if (error instanceof AxiosError) {
if (error.response) {
switch (error.response.status) {
case 404:
return "Error: Resource not found. Please check the ID is correct.";
case 403:
return "Error: Permission denied. You don't have access to this resource.";
case 429:
return "Error: Rate limit exceeded. Please wait before making more requests.";
default:
return `Error: API request failed with status ${error.response.status}`;
}
} else if (error.code === "ECONNABORTED") {
return "Error: Request timed out. Please try again.";
}
}
return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}
```
## Общие утилиты
Выноси общую функциональность в переиспользуемые функции:
```typescript
// Shared API request function
async function makeApiRequest<T>(
endpoint: string,
method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
data?: any,
params?: any
): Promise<T> {
try {
const response = await axios({
method,
url: `${API_BASE_URL}/${endpoint}`,
data,
params,
timeout: 30000,
headers: {
"Content-Type": "application/json",
"Accept": "application/json"
}
});
return response.data;
} catch (error) {
throw error;
}
}
```
## Лучшие практики Async/Await
Всегда используй async/await для сетевых запросов и операций ввода-вывода:
```typescript
// Good: Async network request
async function fetchData(resourceId: string): Promise<ResourceData> {
const response = await axios.get(`${API_URL}/resource/${resourceId}`);
return response.data;
}
// Bad: Promise chains
function fetchData(resourceId: string): Promise<ResourceData> {
return axios.get(`${API_URL}/resource/${resourceId}`)
.then(response => response.data); // Harder to read and maintain
}
```
## Лучшие практики TypeScript
1. **Используй строгий TypeScript**: включи строгий режим в tsconfig.json
2. **Определяй интерфейсы**: создавай ясные определения интерфейсов для всех структур данных
3. **Избегай `any`**: используй корректные типы или `unknown` вместо `any`
4. **Zod для проверки во время выполнения**: используй схемы Zod для проверки внешних данных
5. **Защитники типов**: создавай функции проверки типов для сложных проверок
6. **Обработка ошибок**: всегда используй try-catch с корректной проверкой типа ошибки
7. **Безопасность относительно null**: используй optional chaining (`?.`) и nullish coalescing (`??`)
```typescript
// Good: Type-safe with Zod and interfaces
interface UserResponse {
id: string;
name: string;
email: string;
team?: string;
active: boolean;
}
const UserSchema = z.object({
id: z.string(),
name: z.string(),
email: z.string().email(),
team: z.string().optional(),
active: z.boolean()
});
type User = z.infer<typeof UserSchema>;
async function getUser(id: string): Promise<User> {
const data = await apiCall(`/users/${id}`);
return UserSchema.parse(data); // Runtime validation
}
// Bad: Using any
async function getUser(id: string): Promise<any> {
return await apiCall(`/users/${id}`); // No type safety
}
```
## Конфигурация пакета
### package.json
```json
{
"name": "{service}-mcp-server",
"version": "1.0.0",
"description": "MCP server for {Service} API integration",
"type": "module",
"main": "dist/index.js",
"scripts": {
"start": "node dist/index.js",
"dev": "tsx watch src/index.ts",
"build": "tsc",
"clean": "rm -rf dist"
},
"engines": {
"node": ">=18"
},
"dependencies": {
"@modelcontextprotocol/sdk": "^1.6.1",
"axios": "^1.7.9",
"zod": "^3.23.8"
},
"devDependencies": {
"@types/node": "^22.10.0",
"tsx": "^4.19.2",
"typescript": "^5.7.2"
}
}
```
### tsconfig.json
```json
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"lib": ["ES2022"],
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true,
"allowSyntheticDefaultImports": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}
```
## Полный пример
```typescript
#!/usr/bin/env node
/**
* MCP Server for Example Service.
*
* This server provides tools to interact with Example API, including user search,
* project management, and data export capabilities.
*/
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import axios, { AxiosError } from "axios";
// Constants
const API_BASE_URL = "https://api.example.com/v1";
const CHARACTER_LIMIT = 25000;
// Enums
enum ResponseFormat {
MARKDOWN = "markdown",
JSON = "json"
}
// Zod schemas
const UserSearchInputSchema = z.object({
query: z.string()
.min(2, "Query must be at least 2 characters")
.max(200, "Query must not exceed 200 characters")
.describe("Search string to match against names/emails"),
limit: z.number()
.int()
.min(1)
.max(100)
.default(20)
.describe("Maximum results to return"),
offset: z.number()
.int()
.min(0)
.default(0)
.describe("Number of results to skip for pagination"),
response_format: z.nativeEnum(ResponseFormat)
.default(ResponseFormat.MARKDOWN)
.describe("Output format: 'markdown' for human-readable or 'json' for machine-readable")
}).strict();
type UserSearchInput = z.infer<typeof UserSearchInputSchema>;
// Shared utility functions
async function makeApiRequest<T>(
endpoint: string,
method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
data?: any,
params?: any
): Promise<T> {
try {
const response = await axios({
method,
url: `${API_BASE_URL}/${endpoint}`,
data,
params,
timeout: 30000,
headers: {
"Content-Type": "application/json",
"Accept": "application/json"
}
});
return response.data;
} catch (error) {
throw error;
}
}
function handleApiError(error: unknown): string {
if (error instanceof AxiosError) {
if (error.response) {
switch (error.response.status) {
case 404:
return "Error: Resource not found. Please check the ID is correct.";
case 403:
return "Error: Permission denied. You don't have access to this resource.";
case 429:
return "Error: Rate limit exceeded. Please wait before making more requests.";
default:
return `Error: API request failed with status ${error.response.status}`;
}
} else if (error.code === "ECONNABORTED") {
return "Error: Request timed out. Please try again.";
}
}
return `Error: Unexpected error occurred: ${error instanceof Error ? error.message : String(error)}`;
}
// Create MCP server instance
const server = new McpServer({
name: "example-mcp",
version: "1.0.0"
});
// Register tools
server.registerTool(
"example_search_users",
{
title: "Search Example Users",
description: `[Full description as shown above]`,
inputSchema: UserSearchInputSchema,
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: true
}
},
async (params: UserSearchInput) => {
// Implementation as shown above
}
);
// Main function
// For stdio (local):
async function runStdio() {
if (!process.env.EXAMPLE_API_KEY) {
console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
process.exit(1);
}
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP server running via stdio");
}
// For streamable HTTP (remote):
async function runHTTP() {
if (!process.env.EXAMPLE_API_KEY) {
console.error("ERROR: EXAMPLE_API_KEY environment variable is required");
process.exit(1);
}
const app = express();
app.use(express.json());
app.post('/mcp', async (req, res) => {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
enableJsonResponse: true
});
res.on('close', () => transport.close());
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
});
const port = parseInt(process.env.PORT || '3000');
app.listen(port, () => {
console.error(`MCP server running on http://localhost:${port}/mcp`);
});
}
// Choose transport based on environment
const transport = process.env.TRANSPORT || 'stdio';
if (transport === 'http') {
runHTTP().catch(error => {
console.error("Server error:", error);
process.exit(1);
});
} else {
runStdio().catch(error => {
console.error("Server error:", error);
process.exit(1);
});
}
```
---
## Расширенные возможности MCP
### Регистрация ресурсов
Предоставляй данные как ресурсы для эффективного доступа на основе URI:
```typescript
import { ResourceTemplate } from "@modelcontextprotocol/sdk/types.js";
// Register a resource with URI template
server.registerResource(
{
uri: "file://documents/{name}",
name: "Document Resource",
description: "Access documents by name",
mimeType: "text/plain"
},
async (uri: string) => {
// Extract parameter from URI
const match = uri.match(/^file:\/\/documents\/(.+)$/);
if (!match) {
throw new Error("Invalid URI format");
}
const documentName = match[1];
const content = await loadDocument(documentName);
return {
contents: [{
uri,
mimeType: "text/plain",
text: content
}]
};
}
);
// List available resources dynamically
server.registerResourceList(async () => {
const documents = await getAvailableDocuments();
return {
resources: documents.map(doc => ({
uri: `file://documents/${doc.name}`,
name: doc.name,
mimeType: "text/plain",
description: doc.description
}))
};
});
```
**Когда использовать ресурсы, а когда инструменты:**
- **Ресурсы**: для доступа к данным с простыми параметрами на основе URI
- **Инструменты**: для сложных операций, требующих проверки и бизнес-логики
- **Ресурсы**: когда данные относительно статичны или основаны на шаблонах
- **Инструменты**: когда у операций есть побочные эффекты или сложные рабочие процессы
### Варианты транспорта
TypeScript SDK поддерживает два основных транспортных механизма:
#### Потоковый HTTP (рекомендуется для удалённых серверов)
```typescript
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";
const app = express();
app.use(express.json());
app.post('/mcp', async (req, res) => {
// Create new transport for each request (stateless, prevents request ID collisions)
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
enableJsonResponse: true
});
res.on('close', () => transport.close());
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
});
app.listen(3000);
```
#### stdio (для локальных интеграций)
```typescript
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const transport = new StdioServerTransport();
await server.connect(transport);
```
**Выбор транспорта:**
- **Потоковый HTTP**: веб-сервисы, удалённый доступ, несколько клиентов
- **stdio**: инструменты командной строки, локальная разработка, интеграция подпроцессов
### Поддержка уведомлений
Уведомляй клиентов об изменениях состояния сервера:
```typescript
// Notify when tools list changes
server.notification({
method: "notifications/tools/list_changed"
});
// Notify when resources change
server.notification({
method: "notifications/resources/list_changed"
});
```
Используй уведомления умеренно — только когда возможности сервера действительно меняются.
---
## Лучшие практики кода
### Компонуемость и переиспользуемость кода
В твоей реализации ОБЯЗАТЕЛЬНО должны быть приоритетными компонуемость и повторное использование кода:
1. **Выноси общую функциональность**:
- Создавай переиспользуемые вспомогательные функции для операций, используемых несколькими инструментами
- Создавай общие API-клиенты для HTTP-запросов вместо дублирования кода
- Централизуй логику обработки ошибок в служебных функциях
- Выноси бизнес-логику в отдельные функции, которые можно компоновать
- Выноси общую функциональность выбора и форматирования полей Markdown или JSON
2. **Избегай дублирования**:
- НИКОГДА не копируй сходный код между инструментами
- Если пишешь сходную логику второй раз, вынеси её в функцию
- Общие операции, такие как пагинация, фильтрация, выбор полей и форматирование, должны переиспользоваться
- Логика аутентификации/авторизации должна быть централизованной
## Сборка и запуск
Всегда собирай код TypeScript перед запуском:
```bash
# Build the project
npm run build
# Run the server
npm start
# Development with auto-reload
npm run dev
```
Всегда убеждайся, что `npm run build` завершается успешно, прежде чем считать реализацию завершённой.
## Контрольный список качества
Перед завершением реализации сервера MCP на Node/TypeScript убедись в следующем:
### Стратегическое проектирование
- [ ] Инструменты обеспечивают полные рабочие процессы, а не только обёртки над конечными точками API
- [ ] Имена инструментов отражают естественное деление задач
- [ ] Форматы ответов оптимизированы для эффективного использования контекста агента
- [ ] Где уместно, используются удобочитаемые идентификаторы
- [ ] Сообщения об ошибках направляют агентов к корректному использованию
### Качество реализации
- [ ] СФОКУСИРОВАННАЯ РЕАЛИЗАЦИЯ: реализованы наиболее важные и ценные инструменты
- [ ] Все инструменты зарегистрированы через `registerTool` с полной конфигурацией
- [ ] Все инструменты включают `title`, `description`, `inputSchema` и `annotations`
- [ ] Аннотации заданы корректно (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
- [ ] Все инструменты используют схемы Zod для проверки входных данных во время выполнения с обязательным `.strict()`
- [ ] Во всех схемах Zod есть надлежащие ограничения и описательные сообщения об ошибках
- [ ] У всех инструментов есть исчерпывающие описания с явными типами входных/выходных данных
- [ ] Описания включают примеры возвращаемых значений и полную документацию схем
- [ ] Сообщения об ошибках ясны, подсказывают действия и обучают
### Качество TypeScript
- [ ] Интерфейсы TypeScript определены для всех структур данных
- [ ] Строгий TypeScript включён в tsconfig.json
- [ ] Тип `any` не используется — вместо него используются `unknown` или корректные типы
- [ ] У всех асинхронных функций явно заданы возвращаемые типы Promise<T>
- [ ] Обработка ошибок использует надлежащие защитники типов (например, `axios.isAxiosError`, `z.ZodError`)
### Расширенные возможности (где применимо)
- [ ] Ресурсы зарегистрированы для подходящих конечных точек данных
- [ ] Настроен подходящий транспорт (stdio или потоковый HTTP)
- [ ] Реализованы уведомления для динамических возможностей сервера
- [ ] Обеспечена типобезопасность с интерфейсами SDK
### Конфигурация проекта
- [ ] Package.json включает все необходимые зависимости
- [ ] Скрипт сборки создаёт работающий JavaScript в каталоге dist/
- [ ] Основная точка входа корректно настроена как dist/index.js
- [ ] Имя сервера соответствует формату: `{service}-mcp-server`
- [ ] tsconfig.json корректно настроен со строгим режимом
### Качество кода
- [ ] Пагинация корректно реализована, где применимо
- [ ] Большие ответы проверяются по константе CHARACTER_LIMIT и усекаются с ясными сообщениями
- [ ] Для потенциально больших наборов результатов предоставлены возможности фильтрации
- [ ] Все сетевые операции корректно обрабатывают тайм-ауты и ошибки соединения
- [ ] Общая функциональность вынесена в переиспользуемые функции
- [ ] Возвращаемые типы единообразны для сходных операций
### Тестирование и сборка
- [ ] `npm run build` успешно завершается без ошибок
- [ ] dist/index.js создан и является исполняемым
- [ ] Сервер запускается: `node dist/index.js --help`
- [ ] Все импорты корректно разрешаются
- [ ] Пробные вызовы инструментов работают как ожидается
reference/python_mcp_server.md
# Руководство по реализации сервера MCP на Python
## Обзор
Этот документ содержит лучшие практики и примеры для Python по реализации серверов MCP с помощью MCP Python SDK. Он охватывает настройку сервера, паттерны регистрации инструментов, проверку входных данных с помощью Pydantic, обработку ошибок и полные работающие примеры.
---
## Краткий справочник
### Основные импорты
```python
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
```
### Инициализация сервера
```python
mcp = FastMCP("service_mcp")
```
### Паттерн регистрации инструментов
```python
@mcp.tool(name="tool_name", annotations={...})
async def tool_function(params: InputModel) -> str:
# Implementation
pass
```
---
## MCP Python SDK и FastMCP
Официальный MCP Python SDK предоставляет FastMCP — высокоуровневый фреймворк для создания серверов MCP. Он обеспечивает:
- Автоматическую генерацию description и inputSchema из сигнатур функций и строк документации
- Интеграцию моделей Pydantic для проверки входных данных
- Регистрацию инструментов на основе декоратора `@mcp.tool`
**Чтобы получить полную документацию SDK, используй WebFetch для загрузки:**
`https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`
## Соглашение об именовании серверов
Серверы MCP на Python должны следовать такому шаблону именования:
- **Формат**: `{service}_mcp` (нижний регистр с подчёркиваниями)
- **Примеры**: `github_mcp`, `jira_mcp`, `stripe_mcp`
Имя должно быть:
- Общим (не привязанным к конкретным функциям)
- Описывающим интегрируемый сервис/API
- Легко выводимым из описания задачи
- Без номеров версий или дат
## Реализация инструментов
### Именование инструментов
Используй snake_case для имён инструментов (например, "search_users", "create_project", "get_channel_info") с ясными именами, ориентированными на действия.
**Избегай конфликтов имён**: включай контекст сервиса, чтобы предотвратить совпадения:
- Используй "slack_send_message" вместо просто "send_message"
- Используй "github_create_issue" вместо просто "create_issue"
- Используй "asana_list_tasks" вместо просто "list_tasks"
### Структура инструмента с FastMCP
Инструменты определяются с помощью декоратора `@mcp.tool` и моделей Pydantic для проверки входных данных:
```python
from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP
# Initialize the MCP server
mcp = FastMCP("example_mcp")
# Define Pydantic model for input validation
class ServiceToolInput(BaseModel):
'''Input model for service tool operation.'''
model_config = ConfigDict(
str_strip_whitespace=True, # Auto-strip whitespace from strings
validate_assignment=True, # Validate on assignment
extra='forbid' # Forbid extra fields
)
param1: str = Field(..., description="First parameter description (e.g., 'user123', 'project-abc')", min_length=1, max_length=100)
param2: Optional[int] = Field(default=None, description="Optional integer parameter with constraints", ge=0, le=1000)
tags: Optional[List[str]] = Field(default_factory=list, description="List of tags to apply", max_items=10)
@mcp.tool(
name="service_tool_name",
annotations={
"title": "Human-Readable Tool Title",
"readOnlyHint": True, # Tool does not modify environment
"destructiveHint": False, # Tool does not perform destructive operations
"idempotentHint": True, # Repeated calls have no additional effect
"openWorldHint": False # Tool does not interact with external entities
}
)
async def service_tool_name(params: ServiceToolInput) -> str:
'''Tool description automatically becomes the 'description' field.
This tool performs a specific operation on the service. It validates all inputs
using the ServiceToolInput Pydantic model before processing.
Args:
params (ServiceToolInput): Validated input parameters containing:
- param1 (str): First parameter description
- param2 (Optional[int]): Optional parameter with default
- tags (Optional[List[str]]): List of tags
Returns:
str: JSON-formatted response containing operation results
'''
# Implementation here
pass
```
## Основные возможности Pydantic v2
- Используй `model_config` вместо вложенного класса `Config`
- Используй `field_validator` вместо устаревшего `validator`
- Используй `model_dump()` вместо устаревшего `dict()`
- Валидаторам требуется декоратор `@classmethod`
- Для методов-валидаторов обязательны аннотации типов
```python
from pydantic import BaseModel, Field, field_validator, ConfigDict
class CreateUserInput(BaseModel):
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True
)
name: str = Field(..., description="User's full name", min_length=1, max_length=100)
email: str = Field(..., description="User's email address", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: int = Field(..., description="User's age", ge=0, le=150)
@field_validator('email')
@classmethod
def validate_email(cls, v: str) -> str:
if not v.strip():
raise ValueError("Email cannot be empty")
return v.lower()
```
## Варианты формата ответа
Для гибкости поддерживай несколько форматов вывода:
```python
from enum import Enum
class ResponseFormat(str, Enum):
'''Output format for tool responses.'''
MARKDOWN = "markdown"
JSON = "json"
class UserSearchInput(BaseModel):
query: str = Field(..., description="Search query")
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
```
**Формат Markdown**:
- Используй заголовки, списки и форматирование для ясности
- Преобразовывай временные метки в удобочитаемый формат (например, "2024-01-15 10:30:00 UTC" вместо времени эпохи)
- Показывай отображаемые имена с идентификаторами в скобках (например, "@john.doe (U123456)")
- Опускай многословные метаданные (например, показывай только один URL изображения профиля, а не все размеры)
- Логически группируй связанную информацию
**Формат JSON**:
- Возвращай полные структурированные данные, подходящие для программной обработки
- Включай все доступные поля и метаданные
- Используй единообразные имена и типы полей
## Реализация пагинации
Для инструментов, перечисляющих ресурсы:
```python
class ListInput(BaseModel):
limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
async def list_items(params: ListInput) -> str:
# Make API request with pagination
data = await api_request(limit=params.limit, offset=params.offset)
# Return pagination info
response = {
"total": data["total"],
"count": len(data["items"]),
"offset": params.offset,
"items": data["items"],
"has_more": data["total"] > params.offset + len(data["items"]),
"next_offset": params.offset + len(data["items"]) if data["total"] > params.offset + len(data["items"]) else None
}
return json.dumps(response, indent=2)
```
## Обработка ошибок
Предоставляй ясные сообщения об ошибках, подсказывающие действия:
```python
def _handle_api_error(e: Exception) -> str:
'''Consistent error formatting across all tools.'''
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 404:
return "Error: Resource not found. Please check the ID is correct."
elif e.response.status_code == 403:
return "Error: Permission denied. You don't have access to this resource."
elif e.response.status_code == 429:
return "Error: Rate limit exceeded. Please wait before making more requests."
return f"Error: API request failed with status {e.response.status_code}"
elif isinstance(e, httpx.TimeoutException):
return "Error: Request timed out. Please try again."
return f"Error: Unexpected error occurred: {type(e).__name__}"
```
## Общие утилиты
Выноси общую функциональность в переиспользуемые функции:
```python
# Shared API request function
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
'''Reusable function for all API calls.'''
async with httpx.AsyncClient() as client:
response = await client.request(
method,
f"{API_BASE_URL}/{endpoint}",
timeout=30.0,
**kwargs
)
response.raise_for_status()
return response.json()
```
## Лучшие практики Async/Await
Всегда используй async/await для сетевых запросов и операций ввода-вывода:
```python
# Good: Async network request
async def fetch_data(resource_id: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(f"{API_URL}/resource/{resource_id}")
response.raise_for_status()
return response.json()
# Bad: Synchronous request
def fetch_data(resource_id: str) -> dict:
response = requests.get(f"{API_URL}/resource/{resource_id}") # Blocks
return response.json()
```
## Аннотации типов
Повсеместно используй аннотации типов:
```python
from typing import Optional, List, Dict, Any
async def get_user(user_id: str) -> Dict[str, Any]:
data = await fetch_user(user_id)
return {"id": data["id"], "name": data["name"]}
```
## Строки документации инструментов
Каждый инструмент должен иметь исчерпывающие строки документации с явными сведениями о типах:
```python
async def search_users(params: UserSearchInput) -> str:
'''
Search for users in the Example system by name, email, or team.
This tool searches across all user profiles in the Example platform,
supporting partial matches and various search filters. It does NOT
create or modify users, only searches existing ones.
Args:
params (UserSearchInput): Validated input parameters containing:
- query (str): Search string to match against names/emails (e.g., "john", "@example.com", "team:marketing")
- limit (Optional[int]): Maximum results to return, between 1-100 (default: 20)
- offset (Optional[int]): Number of results to skip for pagination (default: 0)
Returns:
str: JSON-formatted string containing search results with the following schema:
Success response:
{
"total": int, # Total number of matches found
"count": int, # Number of results in this response
"offset": int, # Current pagination offset
"users": [
{
"id": str, # User ID (e.g., "U123456789")
"name": str, # Full name (e.g., "John Doe")
"email": str, # Email address (e.g., "john@example.com")
"team": str # Team name (e.g., "Marketing") - optional
}
]
}
Error response:
"Error: <error message>" or "No users found matching '<query>'"
Examples:
- Use when: "Find all marketing team members" -> params with query="team:marketing"
- Use when: "Search for John's account" -> params with query="john"
- Don't use when: You need to create a user (use example_create_user instead)
- Don't use when: You have a user ID and need full details (use example_get_user instead)
Error Handling:
- Input validation errors are handled by Pydantic model
- Returns "Error: Rate limit exceeded" if too many requests (429 status)
- Returns "Error: Invalid API authentication" if API key is invalid (401 status)
- Returns formatted list of results or "No users found matching 'query'"
'''
```
## Полный пример
Ниже приведён полный пример MCP-сервера на Python:
```python
#!/usr/bin/env python3
'''
MCP Server for Example Service.
This server provides tools to interact with Example API, including user search,
project management, and data export capabilities.
'''
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP
# Initialize the MCP server
mcp = FastMCP("example_mcp")
# Constants
API_BASE_URL = "https://api.example.com/v1"
# Enums
class ResponseFormat(str, Enum):
'''Output format for tool responses.'''
MARKDOWN = "markdown"
JSON = "json"
# Pydantic Models for Input Validation
class UserSearchInput(BaseModel):
'''Input model for user search operations.'''
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True
)
query: str = Field(..., description="Search string to match against names/emails", min_length=2, max_length=200)
limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")
@field_validator('query')
@classmethod
def validate_query(cls, v: str) -> str:
if not v.strip():
raise ValueError("Query cannot be empty or whitespace only")
return v.strip()
# Shared utility functions
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
'''Reusable function for all API calls.'''
async with httpx.AsyncClient() as client:
response = await client.request(
method,
f"{API_BASE_URL}/{endpoint}",
timeout=30.0,
**kwargs
)
response.raise_for_status()
return response.json()
def _handle_api_error(e: Exception) -> str:
'''Consistent error formatting across all tools.'''
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 404:
return "Error: Resource not found. Please check the ID is correct."
elif e.response.status_code == 403:
return "Error: Permission denied. You don't have access to this resource."
elif e.response.status_code == 429:
return "Error: Rate limit exceeded. Please wait before making more requests."
return f"Error: API request failed with status {e.response.status_code}"
elif isinstance(e, httpx.TimeoutException):
return "Error: Request timed out. Please try again."
return f"Error: Unexpected error occurred: {type(e).__name__}"
# Tool definitions
@mcp.tool(
name="example_search_users",
annotations={
"title": "Search Example Users",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def example_search_users(params: UserSearchInput) -> str:
'''Search for users in the Example system by name, email, or team.
[Full docstring as shown above]
'''
try:
# Make API request using validated parameters
data = await _make_api_request(
"users/search",
params={
"q": params.query,
"limit": params.limit,
"offset": params.offset
}
)
users = data.get("users", [])
total = data.get("total", 0)
if not users:
return f"No users found matching '{params.query}'"
# Format response based on requested format
if params.response_format == ResponseFormat.MARKDOWN:
lines = [f"# User Search Results: '{params.query}'", ""]
lines.append(f"Found {total} users (showing {len(users)})")
lines.append("")
for user in users:
lines.append(f"## {user['name']} ({user['id']})")
lines.append(f"- **Email**: {user['email']}")
if user.get('team'):
lines.append(f"- **Team**: {user['team']}")
lines.append("")
return "\n".join(lines)
else:
# Machine-readable JSON format
import json
response = {
"total": total,
"count": len(users),
"offset": params.offset,
"users": users
}
return json.dumps(response, indent=2)
except Exception as e:
return _handle_api_error(e)
if __name__ == "__main__":
mcp.run()
```
---
## Расширенные возможности FastMCP
### Внедрение параметра Context
FastMCP может автоматически внедрять в инструменты параметр `Context` для расширенных возможностей: журналирования, отчётов о ходе выполнения, чтения ресурсов и взаимодействия с пользователем:
```python
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("example_mcp")
@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
'''Advanced tool with context access for logging and progress.'''
# Report progress for long operations
await ctx.report_progress(0.25, "Starting search...")
# Log information for debugging
await ctx.log_info("Processing query", {"query": query, "timestamp": datetime.now()})
# Perform search
results = await search_api(query)
await ctx.report_progress(0.75, "Formatting results...")
# Access server configuration
server_name = ctx.fastmcp.name
return format_results(results)
@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
'''Tool that can request additional input from users.'''
# Request sensitive information when needed
api_key = await ctx.elicit(
prompt="Please provide your API key:",
input_type="password"
)
# Use the provided key
return await api_call(resource_id, api_key)
```
**Возможности Context:**
- `ctx.report_progress(progress, message)` — сообщать о ходе выполнения длительных операций
- `ctx.log_info(message, data)` / `ctx.log_error()` / `ctx.log_debug()` — журналирование
- `ctx.elicit(prompt, input_type)` — запрашивать ввод у пользователей
- `ctx.fastmcp.name` — получать доступ к конфигурации сервера
- `ctx.read_resource(uri)` — читать ресурсы MCP
### Регистрация ресурсов
Предоставляйте данные в виде ресурсов для эффективного доступа на основе шаблонов:
```python
@mcp.resource("file://documents/{name}")
async def get_document(name: str) -> str:
'''Expose documents as MCP resources.
Resources are useful for static or semi-static data that doesn't
require complex parameters. They use URI templates for flexible access.
'''
document_path = f"./docs/{name}"
with open(document_path, "r") as f:
return f.read()
@mcp.resource("config://settings/{key}")
async def get_setting(key: str, ctx: Context) -> str:
'''Expose configuration as resources with context.'''
settings = await load_settings()
return json.dumps(settings.get(key, {}))
```
**Когда использовать ресурсы, а когда инструменты:**
- **Ресурсы**: для доступа к данным с простыми параметрами (шаблонами URI)
- **Инструменты**: для сложных операций с проверкой данных и бизнес-логикой
### Типы структурированного вывода
FastMCP поддерживает не только строки, но и несколько других типов возвращаемых значений:
```python
from typing import TypedDict
from dataclasses import dataclass
from pydantic import BaseModel
# TypedDict for structured returns
class UserData(TypedDict):
id: str
name: str
email: str
@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
'''Returns structured data - FastMCP handles serialization.'''
return {"id": user_id, "name": "John Doe", "email": "john@example.com"}
# Pydantic models for complex validation
class DetailedUser(BaseModel):
id: str
name: str
email: str
created_at: datetime
metadata: Dict[str, Any]
@mcp.tool()
async def get_user_detailed(user_id: str) -> DetailedUser:
'''Returns Pydantic model - automatically generates schema.'''
user = await fetch_user(user_id)
return DetailedUser(**user)
```
### Управление жизненным циклом
Инициализируйте ресурсы, которые сохраняются между запросами:
```python
from contextlib import asynccontextmanager
@asynccontextmanager
async def app_lifespan():
'''Manage resources that live for the server's lifetime.'''
# Initialize connections, load config, etc.
db = await connect_to_database()
config = load_configuration()
# Make available to all tools
yield {"db": db, "config": config}
# Cleanup on shutdown
await db.close()
mcp = FastMCP("example_mcp", lifespan=app_lifespan)
@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
'''Access lifespan resources through context.'''
db = ctx.request_context.lifespan_state["db"]
results = await db.query(query)
return format_results(results)
```
### Варианты транспорта
FastMCP поддерживает два основных транспортных механизма:
```python
# stdio transport (for local tools) - default
if __name__ == "__main__":
mcp.run()
# Streamable HTTP transport (for remote servers)
if __name__ == "__main__":
mcp.run(transport="streamable_http", port=8000)
```
**Выбор транспорта:**
- **stdio**: инструменты командной строки, локальные интеграции, выполнение подпроцессов
- **Streamable HTTP**: веб-сервисы, удалённый доступ, несколько клиентов
---
## Рекомендации по написанию кода
### Компонуемость и повторное использование кода
Ваша реализация ДОЛЖНА отдавать приоритет компонуемости и повторному использованию кода:
1. **Выделяйте общую функциональность**:
- Создавайте повторно используемые вспомогательные функции для операций, применяемых в нескольких инструментах
- Создавайте общие API-клиенты для HTTP-запросов вместо дублирования кода
- Централизуйте логику обработки ошибок во вспомогательных функциях
- Выделяйте бизнес-логику в отдельные функции, которые можно комбинировать
- Выделяйте общую функциональность выбора и форматирования полей Markdown или JSON
2. **Избегайте дублирования**:
- НИКОГДА не копируйте и не вставляйте похожий код между инструментами
- Если вы пишете похожую логику дважды, выделите её в функцию
- Общие операции, такие как пагинация, фильтрация, выбор полей и форматирование, должны использоваться совместно
- Логика аутентификации/авторизации должна быть централизована
### Рекомендации, специфичные для Python
1. **Используйте аннотации типов**: всегда указывайте аннотации типов параметров функций и возвращаемых значений
2. **Модели Pydantic**: определяйте понятные модели Pydantic для проверки всех входных данных
3. **Избегайте ручной проверки**: поручайте Pydantic проверку входных данных с помощью ограничений
4. **Правильные импорты**: группируйте импорты (стандартная библиотека, сторонние пакеты, локальные модули)
5. **Обработка ошибок**: используйте конкретные типы исключений (httpx.HTTPStatusError, а не общее Exception)
6. **Асинхронные контекстные менеджеры**: используйте `async with` для ресурсов, требующих освобождения
7. **Константы**: определяйте константы уровня модуля в UPPER_CASE
## Контрольный список качества
Прежде чем завершить реализацию MCP-сервера на Python, убедитесь в следующем:
### Стратегический дизайн
- [ ] Инструменты позволяют выполнять полные рабочие процессы, а не просто оборачивают конечные точки API
- [ ] Имена инструментов отражают естественное разделение задач
- [ ] Форматы ответов оптимизированы для эффективного использования контекста агента
- [ ] Где это уместно, используются понятные человеку идентификаторы
- [ ] Сообщения об ошибках направляют агентов к правильному использованию
### Качество реализации
- [ ] СФОКУСИРОВАННАЯ РЕАЛИЗАЦИЯ: реализованы самые важные и ценные инструменты
- [ ] У всех инструментов описательные имена и документация
- [ ] Типы возвращаемых значений согласованы для сходных операций
- [ ] Обработка ошибок реализована для всех внешних вызовов
- [ ] Имя сервера соответствует формату: `{service}_mcp`
- [ ] Все сетевые операции используют async/await
- [ ] Общая функциональность выделена в повторно используемые функции
- [ ] Сообщения об ошибках понятны, подсказывают действия и помогают разобраться
- [ ] Выходные данные корректно проверяются и форматируются
### Конфигурация инструментов
- [ ] Все инструменты реализуют 'name' и 'annotations' в декораторе
- [ ] Аннотации заданы правильно (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
- [ ] Все инструменты используют Pydantic BaseModel для проверки входных данных с определениями Field()
- [ ] Все поля Pydantic имеют явные типы и описания с ограничениями
- [ ] У всех инструментов подробные docstring с явными типами входных и выходных данных
- [ ] Docstring включают полную структуру схемы для возвращаемых значений dict/JSON
- [ ] Модели Pydantic выполняют проверку входных данных (ручная проверка не требуется)
### Расширенные возможности (где применимо)
- [ ] Внедрение Context используется для журналирования, отображения прогресса или запроса ввода
- [ ] Для соответствующих конечных точек данных зарегистрированы ресурсы
- [ ] Для постоянных соединений реализовано управление жизненным циклом
- [ ] Используются структурированные типы вывода (TypedDict, модели Pydantic)
- [ ] Настроен подходящий транспорт (stdio или streamable HTTP)
### Качество кода
- [ ] Файл содержит правильные импорты, включая импорты Pydantic
- [ ] Где применимо, корректно реализована пагинация
- [ ] Для потенциально больших наборов результатов предусмотрены параметры фильтрации
- [ ] Все асинхронные функции правильно определены с помощью `async def`
- [ ] Использование HTTP-клиента следует асинхронным подходам с подходящими контекстными менеджерами
- [ ] Аннотации типов используются во всём коде
- [ ] Константы определены на уровне модуля в UPPER_CASE
### Тестирование
- [ ] Сервер успешно запускается: `python your_server.py --help`
- [ ] Все импорты разрешаются правильно
- [ ] Примерные вызовы инструментов работают ожидаемым образом
- [ ] Ошибочные сценарии корректно обрабатываются
scripts/connections.py
"""Lightweight connection handling for MCP servers."""
from abc import ABC, abstractmethod
from contextlib import AsyncExitStack
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamablehttp_client
class MCPConnection(ABC):
"""Base class for MCP server connections."""
def __init__(self):
self.session = None
self._stack = None
@abstractmethod
def _create_context(self):
"""Create the connection context based on connection type."""
async def __aenter__(self):
"""Initialize MCP server connection."""
self._stack = AsyncExitStack()
await self._stack.__aenter__()
try:
ctx = self._create_context()
result = await self._stack.enter_async_context(ctx)
if len(result) == 2:
read, write = result
elif len(result) == 3:
read, write, _ = result
else:
raise ValueError(f"Unexpected context result: {result}")
session_ctx = ClientSession(read, write)
self.session = await self._stack.enter_async_context(session_ctx)
await self.session.initialize()
return self
except BaseException:
await self._stack.__aexit__(None, None, None)
raise
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Clean up MCP server connection resources."""
if self._stack:
await self._stack.__aexit__(exc_type, exc_val, exc_tb)
self.session = None
self._stack = None
async def list_tools(self) -> list[dict[str, Any]]:
"""Retrieve available tools from the MCP server."""
response = await self.session.list_tools()
return [
{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema,
}
for tool in response.tools
]
async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any:
"""Call a tool on the MCP server with provided arguments."""
result = await self.session.call_tool(tool_name, arguments=arguments)
return result.content
class MCPConnectionStdio(MCPConnection):
"""MCP connection using standard input/output."""
def __init__(self, command: str, args: list[str] = None, env: dict[str, str] = None):
super().__init__()
self.command = command
self.args = args or []
self.env = env
def _create_context(self):
return stdio_client(
StdioServerParameters(command=self.command, args=self.args, env=self.env)
)
class MCPConnectionSSE(MCPConnection):
"""MCP connection using Server-Sent Events."""
def __init__(self, url: str, headers: dict[str, str] = None):
super().__init__()
self.url = url
self.headers = headers or {}
def _create_context(self):
return sse_client(url=self.url, headers=self.headers)
class MCPConnectionHTTP(MCPConnection):
"""MCP connection using Streamable HTTP."""
def __init__(self, url: str, headers: dict[str, str] = None):
super().__init__()
self.url = url
self.headers = headers or {}
def _create_context(self):
return streamablehttp_client(url=self.url, headers=self.headers)
def create_connection(
transport: str,
command: str = None,
args: list[str] = None,
env: dict[str, str] = None,
url: str = None,
headers: dict[str, str] = None,
) -> MCPConnection:
"""Factory function to create the appropriate MCP connection.
Args:
transport: Connection type ("stdio", "sse", or "http")
command: Command to run (stdio only)
args: Command arguments (stdio only)
env: Environment variables (stdio only)
url: Server URL (sse and http only)
headers: HTTP headers (sse and http only)
Returns:
MCPConnection instance
"""
transport = transport.lower()
if transport == "stdio":
if not command:
raise ValueError("Command is required for stdio transport")
return MCPConnectionStdio(command=command, args=args, env=env)
elif transport == "sse":
if not url:
raise ValueError("URL is required for sse transport")
return MCPConnectionSSE(url=url, headers=headers)
elif transport in ["http", "streamable_http", "streamable-http"]:
if not url:
raise ValueError("URL is required for http transport")
return MCPConnectionHTTP(url=url, headers=headers)
else:
raise ValueError(f"Unsupported transport type: {transport}. Use 'stdio', 'sse', or 'http'")
scripts/evaluation.py
"""MCP Server Evaluation Harness
This script evaluates MCP servers by running test questions against them using Claude.
"""
import argparse
import asyncio
import json
import re
import sys
import time
import traceback
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any
from anthropic import Anthropic
from connections import create_connection
EVALUATION_PROMPT = """You are an AI assistant with access to tools.
When given a task, you MUST:
1. Use the available tools to complete the task
2. Provide summary of each step in your approach, wrapped in <summary> tags
3. Provide feedback on the tools provided, wrapped in <feedback> tags
4. Provide your final response, wrapped in <response> tags
Summary Requirements:
- In your <summary> tags, you must explain:
- The steps you took to complete the task
- Which tools you used, in what order, and why
- The inputs you provided to each tool
- The outputs you received from each tool
- A summary for how you arrived at the response
Feedback Requirements:
- In your <feedback> tags, provide constructive feedback on the tools:
- Comment on tool names: Are they clear and descriptive?
- Comment on input parameters: Are they well-documented? Are required vs optional parameters clear?
- Comment on descriptions: Do they accurately describe what the tool does?
- Comment on any errors encountered during tool usage: Did the tool fail to execute? Did the tool return too many tokens?
- Identify specific areas for improvement and explain WHY they would help
- Be specific and actionable in your suggestions
Response Requirements:
- Your response should be concise and directly address what was asked
- Always wrap your final response in <response> tags
- If you cannot solve the task return <response>NOT_FOUND</response>
- For numeric responses, provide just the number
- For IDs, provide just the ID
- For names or text, provide the exact text requested
- Your response should go last"""
def parse_evaluation_file(file_path: Path) -> list[dict[str, Any]]:
"""Parse XML evaluation file with qa_pair elements."""
try:
tree = ET.parse(file_path)
root = tree.getroot()
evaluations = []
for qa_pair in root.findall(".//qa_pair"):
question_elem = qa_pair.find("question")
answer_elem = qa_pair.find("answer")
if question_elem is not None and answer_elem is not None:
evaluations.append({
"question": (question_elem.text or "").strip(),
"answer": (answer_elem.text or "").strip(),
})
return evaluations
except Exception as e:
print(f"Error parsing evaluation file {file_path}: {e}")
return []
def extract_xml_content(text: str, tag: str) -> str | None:
"""Extract content from XML tags."""
pattern = rf"<{tag}>(.*?)</{tag}>"
matches = re.findall(pattern, text, re.DOTALL)
return matches[-1].strip() if matches else None
async def agent_loop(
client: Anthropic,
model: str,
question: str,
tools: list[dict[str, Any]],
connection: Any,
) -> tuple[str, dict[str, Any]]:
"""Run the agent loop with MCP tools."""
messages = [{"role": "user", "content": question}]
response = await asyncio.to_thread(
client.messages.create,
model=model,
max_tokens=4096,
system=EVALUATION_PROMPT,
messages=messages,
tools=tools,
)
messages.append({"role": "assistant", "content": response.content})
tool_metrics = {}
while response.stop_reason == "tool_use":
tool_use = next(block for block in response.content if block.type == "tool_use")
tool_name = tool_use.name
tool_input = tool_use.input
tool_start_ts = time.time()
try:
tool_result = await connection.call_tool(tool_name, tool_input)
tool_response = json.dumps(tool_result) if isinstance(tool_result, (dict, list)) else str(tool_result)
except Exception as e:
tool_response = f"Error executing tool {tool_name}: {str(e)}\n"
tool_response += traceback.format_exc()
tool_duration = time.time() - tool_start_ts
if tool_name not in tool_metrics:
tool_metrics[tool_name] = {"count": 0, "durations": []}
tool_metrics[tool_name]["count"] += 1
tool_metrics[tool_name]["durations"].append(tool_duration)
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": tool_response,
}]
})
response = await asyncio.to_thread(
client.messages.create,
model=model,
max_tokens=4096,
system=EVALUATION_PROMPT,
messages=messages,
tools=tools,
)
messages.append({"role": "assistant", "content": response.content})
response_text = next(
(block.text for block in response.content if hasattr(block, "text")),
None,
)
return response_text, tool_metrics
async def evaluate_single_task(
client: Anthropic,
model: str,
qa_pair: dict[str, Any],
tools: list[dict[str, Any]],
connection: Any,
task_index: int,
) -> dict[str, Any]:
"""Evaluate a single QA pair with the given tools."""
start_time = time.time()
print(f"Task {task_index + 1}: Running task with question: {qa_pair['question']}")
response, tool_metrics = await agent_loop(client, model, qa_pair["question"], tools, connection)
response_value = extract_xml_content(response, "response")
summary = extract_xml_content(response, "summary")
feedback = extract_xml_content(response, "feedback")
duration_seconds = time.time() - start_time
return {
"question": qa_pair["question"],
"expected": qa_pair["answer"],
"actual": response_value,
"score": int(response_value == qa_pair["answer"]) if response_value else 0,
"total_duration": duration_seconds,
"tool_calls": tool_metrics,
"num_tool_calls": sum(len(metrics["durations"]) for metrics in tool_metrics.values()),
"summary": summary,
"feedback": feedback,
}
REPORT_HEADER = """
# Evaluation Report
## Summary
- **Accuracy**: {correct}/{total} ({accuracy:.1f}%)
- **Average Task Duration**: {average_duration_s:.2f}s
- **Average Tool Calls per Task**: {average_tool_calls:.2f}
- **Total Tool Calls**: {total_tool_calls}
---
"""
TASK_TEMPLATE = """
### Task {task_num}
**Question**: {question}
**Ground Truth Answer**: `{expected_answer}`
**Actual Answer**: `{actual_answer}`
**Correct**: {correct_indicator}
**Duration**: {total_duration:.2f}s
**Tool Calls**: {tool_calls}
**Summary**
{summary}
**Feedback**
{feedback}
---
"""
async def run_evaluation(
eval_path: Path,
connection: Any,
model: str = "claude-3-7-sonnet-20250219",
) -> str:
"""Run evaluation with MCP server tools."""
print("🚀 Starting Evaluation")
client = Anthropic()
tools = await connection.list_tools()
print(f"📋 Loaded {len(tools)} tools from MCP server")
qa_pairs = parse_evaluation_file(eval_path)
print(f"📋 Loaded {len(qa_pairs)} evaluation tasks")
results = []
for i, qa_pair in enumerate(qa_pairs):
print(f"Processing task {i + 1}/{len(qa_pairs)}")
result = await evaluate_single_task(client, model, qa_pair, tools, connection, i)
results.append(result)
correct = sum(r["score"] for r in results)
accuracy = (correct / len(results)) * 100 if results else 0
average_duration_s = sum(r["total_duration"] for r in results) / len(results) if results else 0
average_tool_calls = sum(r["num_tool_calls"] for r in results) / len(results) if results else 0
total_tool_calls = sum(r["num_tool_calls"] for r in results)
report = REPORT_HEADER.format(
correct=correct,
total=len(results),
accuracy=accuracy,
average_duration_s=average_duration_s,
average_tool_calls=average_tool_calls,
total_tool_calls=total_tool_calls,
)
report += "".join([
TASK_TEMPLATE.format(
task_num=i + 1,
question=qa_pair["question"],
expected_answer=qa_pair["answer"],
actual_answer=result["actual"] or "N/A",
correct_indicator="✅" if result["score"] else "❌",
total_duration=result["total_duration"],
tool_calls=json.dumps(result["tool_calls"], indent=2),
summary=result["summary"] or "N/A",
feedback=result["feedback"] or "N/A",
)
for i, (qa_pair, result) in enumerate(zip(qa_pairs, results))
])
return report
def parse_headers(header_list: list[str]) -> dict[str, str]:
"""Parse header strings in format 'Key: Value' into a dictionary."""
headers = {}
if not header_list:
return headers
for header in header_list:
if ":" in header:
key, value = header.split(":", 1)
headers[key.strip()] = value.strip()
else:
print(f"Warning: Ignoring malformed header: {header}")
return headers
def parse_env_vars(env_list: list[str]) -> dict[str, str]:
"""Parse environment variable strings in format 'KEY=VALUE' into a dictionary."""
env = {}
if not env_list:
return env
for env_var in env_list:
if "=" in env_var:
key, value = env_var.split("=", 1)
env[key.strip()] = value.strip()
else:
print(f"Warning: Ignoring malformed environment variable: {env_var}")
return env
async def main():
parser = argparse.ArgumentParser(
description="Evaluate MCP servers using test questions",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Evaluate a local stdio MCP server
python evaluation.py -t stdio -c python -a my_server.py eval.xml
# Evaluate an SSE MCP server
python evaluation.py -t sse -u https://example.com/mcp -H "Authorization: Bearer token" eval.xml
# Evaluate an HTTP MCP server with custom model
python evaluation.py -t http -u https://example.com/mcp -m claude-3-5-sonnet-20241022 eval.xml
""",
)
parser.add_argument("eval_file", type=Path, help="Path to evaluation XML file")
parser.add_argument("-t", "--transport", choices=["stdio", "sse", "http"], default="stdio", help="Transport type (default: stdio)")
parser.add_argument("-m", "--model", default="claude-3-7-sonnet-20250219", help="Claude model to use (default: claude-3-7-sonnet-20250219)")
stdio_group = parser.add_argument_group("stdio options")
stdio_group.add_argument("-c", "--command", help="Command to run MCP server (stdio only)")
stdio_group.add_argument("-a", "--args", nargs="+", help="Arguments for the command (stdio only)")
stdio_group.add_argument("-e", "--env", nargs="+", help="Environment variables in KEY=VALUE format (stdio only)")
remote_group = parser.add_argument_group("sse/http options")
remote_group.add_argument("-u", "--url", help="MCP server URL (sse/http only)")
remote_group.add_argument("-H", "--header", nargs="+", dest="headers", help="HTTP headers in 'Key: Value' format (sse/http only)")
parser.add_argument("-o", "--output", type=Path, help="Output file for evaluation report (default: stdout)")
args = parser.parse_args()
if not args.eval_file.exists():
print(f"Error: Evaluation file not found: {args.eval_file}")
sys.exit(1)
headers = parse_headers(args.headers) if args.headers else None
env_vars = parse_env_vars(args.env) if args.env else None
try:
connection = create_connection(
transport=args.transport,
command=args.command,
args=args.args,
env=env_vars,
url=args.url,
headers=headers,
)
except ValueError as e:
print(f"Error: {e}")
sys.exit(1)
print(f"🔗 Connecting to MCP server via {args.transport}...")
async with connection:
print("✅ Connected successfully")
report = await run_evaluation(args.eval_file, connection, args.model)
if args.output:
args.output.write_text(report)
print(f"\n✅ Report saved to {args.output}")
else:
print("\n" + report)
if __name__ == "__main__":
asyncio.run(main())
scripts/example_evaluation.xml
<evaluation>
<qa_pair>
<question>Рассчитайте сложные проценты для $10,000, вложенных под 5% годовых с ежемесячным начислением процентов на 3 года. Какова итоговая сумма в долларах (с округлением до 2 знаков после запятой)?</question>
<answer>11614.72</answer>
</qa_pair>
<qa_pair>
<question>Тело брошено под углом 45 градусов с начальной скоростью 50 м/с. Рассчитайте общее расстояние (в метрах), пройденное им от точки запуска за 2 секунды, считая g=9.8 м/с². Округлите до 2 знаков после запятой.</question>
<answer>87.25</answer>
</qa_pair>
<qa_pair>
<question>Объём сферы составляет 500 кубических метров. Рассчитайте площадь её поверхности в квадратных метрах. Округлите до 2 знаков после запятой.</question>
<answer>304.65</answer>
</qa_pair>
<qa_pair>
<question>Рассчитайте стандартное отклонение генеральной совокупности для этого набора данных: [12, 15, 18, 22, 25, 30, 35]. Округлите до 2 знаков после запятой.</question>
<answer>7.61</answer>
</qa_pair>
<qa_pair>
<question>Рассчитайте pH раствора с концентрацией ионов водорода 3.5 × 10^-5 M. Округлите до 2 знаков после запятой.</question>
<answer>4.46</answer>
</qa_pair>
</evaluation>
Текст доступен бесплатно по CC0 1.0. Источники и лицензии.
Как использовать навык
Прочитайте инструкцию и проверьте, какие файлы, инструменты и подключения ей нужны. Перенесите навык в совместимое приложение для AI-агентов или используйте подходящие шаги в чате. Если навык состоит из нескольких файлов, сохраните их структуру.