trello-tools: плагин и маркетплейс gp-tools для Claude Code

Сервер запускается через uv run --script, ключи запрашиваются
при установке плагина (userConfig) вместо .env.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-24 14:14:29 +00:00
commit b4cabbf146
14 changed files with 2100 additions and 0 deletions
+14
View File
@@ -0,0 +1,14 @@
{
"name": "gp-tools",
"owner": {
"name": "maxerton"
},
"description": "Инструменты для работы с Trello и qa-docs-service",
"plugins": [
{
"name": "trello-tools",
"source": "./",
"description": "Карточки Trello, QA-документы и задания для фронта/бэка"
}
]
}
+62
View File
@@ -0,0 +1,62 @@
{
"name": "trello-tools",
"description": "Карточки Trello, QA-документы и задания для фронта/бэка",
"author": {
"name": "maxerton"
},
"mcpServers": {
"trello-tools": {
"command": "uv",
"args": ["run", "--quiet", "--script", "${CLAUDE_PLUGIN_ROOT}/server.py"]
}
},
"userConfig": {
"trello_api_key": {
"type": "string",
"title": "Trello API key",
"description": "Ключ Trello: https://trello.com/power-ups/admin → API key",
"sensitive": true
},
"trello_token": {
"type": "string",
"title": "Trello token",
"description": "Токен Trello, выданный под этот API key",
"sensitive": true
},
"qa_tasks_token": {
"type": "string",
"title": "QA tasks token",
"description": "Токен заданий для фронта/бэка (доступ только к /tasks). Выдаёт владелец сервиса",
"sensitive": true
},
"qa_service_api_key": {
"type": "string",
"title": "QA service API key",
"description": "Ключ владельца для публикации QA-документов. Фронту и бэку не нужен — оставь пустым",
"sensitive": true
},
"qa_service_url": {
"type": "string",
"title": "QA service URL",
"description": "Адрес qa-docs-service",
"default": "https://hpd.maxerton.live/"
},
"trello_board_id": {
"type": "string",
"title": "Trello board",
"description": "Доска, на которой trello_create_card создаёт карточки: id или shortLink из https://trello.com/b/<shortLink>/...",
"default": "3SXxWovk"
},
"trello_create_lists": {
"type": "string",
"title": "Trello lists",
"description": "Столбцы этой доски, куда разрешено создавать карточки, через запятую",
"default": "Next Spring Backend,Next Sprint FrontEnd"
},
"trello_attachments_dir": {
"type": "string",
"title": "Attachments dir",
"description": "Куда trello_get_attachment скачивает вложения. Пусто — <tmp>/trello-attachments/<card_id>/"
}
}
}
+15
View File
@@ -0,0 +1,15 @@
TRELLO_API_KEY=
TRELLO_TOKEN=
QA_SERVICE_URL=https://qa.<домен>/
QA_SERVICE_API_KEY=
# Доска, на которой trello_create_card создаёт карточки: id или shortLink
# из ссылки https://trello.com/b/<shortLink>/...
TRELLO_BOARD_ID=
# Столбцы этой доски, куда разрешено создавать карточки, через запятую.
TRELLO_CREATE_LISTS=
# Только для Claude на фронте/бэке, у которого нет QA_SERVICE_API_KEY:
# токен заданий (= TASKS_TOKEN в qa-docs-service), даёт доступ только к /tasks.
QA_TASKS_TOKEN=
# Куда trello_get_attachment скачивает файлы вложений
# (по умолчанию <tmp>/trello-attachments/<card_id>/).
TRELLO_ATTACHMENTS_DIR=
+5
View File
@@ -0,0 +1,5 @@
.env
__pycache__/
*.pyc
.pytest_cache/
*.zip
+49
View File
@@ -0,0 +1,49 @@
# trello-tools
MCP-сервер для Claude Code: карточки Trello, QA-документы и задания для фронта/бэка
через qa-docs-service. Распространяется как плагин из маркетплейса `gp-tools`.
## Установка
Нужен [uv](https://docs.astral.sh/uv/) — через него плагин ставит Python и зависимости
сам, при первом запуске:
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```
В Claude Code:
```
/plugin marketplace add https://git.maxerton.live/maxerton/trello-mcp.git
/plugin install trello-tools@gp-tools
```
При установке Claude Code спросит ключи. Токены хранятся у тебя локально
(`~/.claude/.credentials.json`), в репозиторий ничего не попадает.
| Поле | Кому нужно |
|---|---|
| Trello API key, Trello token | всем, кто работает с карточками Trello |
| QA tasks token | фронту и бэку — доступ к заданиям `/tasks` (выдаёт владелец) |
| QA service API key | только владельцу, остальные оставляют пустым |
| QA service URL, Trello board, Trello lists | уже заполнены, менять не нужно |
Изменить ключи позже — через `/config`. После установки перезапусти Claude Code.
## Обновление
```
/plugin marketplace update gp-tools
/plugin update trello-tools@gp-tools
```
и перезапусти Claude Code.
## Для владельца
- Опубликовать новую версию — `git push`: каждый коммит считается новой версией.
- Если сервер зарегистрирован ещё и вручную (`claude mcp add trello-tools ...`),
а плагин тоже установлен — инструменты задвоятся. Убрать ручную регистрацию:
`claude mcp remove trello-tools -s local`.
- Локальная разработка по-прежнему через `.env` (см. `.env.example`) и `install.txt`.
+1
View File
@@ -0,0 +1 @@
claude mcp add trello-tools --scope local -- ~/projects/gp_mcp/.venv/bin/python ~/projects/gp_mcp/trello-mcp/server.py
+166
View File
@@ -0,0 +1,166 @@
"""Клиент qa-docs-service."""
from __future__ import annotations
import os
import httpx
class QaServiceConfigError(RuntimeError):
"""Не заданы обязательные переменные окружения для qa-docs-service."""
def _config() -> tuple[str, str]:
url = os.environ.get("QA_SERVICE_URL")
api_key = os.environ.get("QA_SERVICE_API_KEY")
if not url or not api_key:
raise QaServiceConfigError(
"QA_SERVICE_URL / QA_SERVICE_API_KEY не заданы (проверь .env)"
)
return url.rstrip("/"), api_key
def slug_from(slug_or_url: str) -> str:
"""Принимает и slug, и полную ссылку вида .../docs/<slug>."""
return slug_or_url.strip().rstrip("/").rsplit("/", 1)[-1]
async def publish_doc(
title: str,
markdown: str,
card_url: str | None = None,
problem: str | None = None,
) -> dict:
"""Публикует новый QA-документ, возвращает {"slug": ..., "url": ...}."""
base_url, api_key = _config()
async with httpx.AsyncClient() as client:
response = await client.post(
f"{base_url}/docs",
headers={"X-API-Key": api_key},
json={
"title": title,
"markdown": markdown,
"card_url": card_url,
"problem": problem,
},
)
response.raise_for_status()
return response.json()
async def update_doc(
slug: str,
title: str,
markdown: str,
card_url: str | None = None,
problem: str | None = None,
) -> dict:
"""Перезаписывает уже опубликованный документ по slug (та же ссылка).
Возвращает {"slug": ..., "url": ...}; 404 (через raise_for_status), если
под этим slug ничего не опубликовано.
"""
base_url, api_key = _config()
async with httpx.AsyncClient() as client:
response = await client.put(
f"{base_url}/docs/{slug_from(slug)}",
headers={"X-API-Key": api_key},
json={
"title": title,
"markdown": markdown,
"card_url": card_url,
"problem": problem,
},
)
response.raise_for_status()
return response.json()
async def get_doc(slug: str) -> dict:
"""Скачивает опубликованный документ: исходный markdown, метаданные и
отметки тестировщиков по сценариям (поле scenarios).
404 (через raise_for_status), если под этим slug ничего не опубликовано.
"""
base_url, api_key = _config()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{base_url}/docs/{slug_from(slug)}/raw",
headers={"X-API-Key": api_key},
)
response.raise_for_status()
return response.json()
def _tasks_config() -> tuple[str, dict]:
"""URL сервиса и заголовки для /tasks.
Подходит и основной ключ (у тебя), и QA_TASKS_TOKEN (у Claude на
фронте/бэке, которому доступны только задания).
"""
url = os.environ.get("QA_SERVICE_URL")
token = os.environ.get("QA_SERVICE_API_KEY") or os.environ.get("QA_TASKS_TOKEN")
if not url or not token:
raise QaServiceConfigError(
"QA_SERVICE_URL и QA_SERVICE_API_KEY или QA_TASKS_TOKEN не заданы (проверь .env)"
)
return url.rstrip("/"), {"Authorization": f"Bearer {token}"}
async def publish_task(
target: str, title: str, markdown: str, card_url: str | None = None
) -> dict:
"""Публикует задание для фронта/бэка, возвращает {slug, url, target, ...}."""
base_url, headers = _tasks_config()
async with httpx.AsyncClient() as client:
response = await client.post(
f"{base_url}/tasks",
headers=headers,
json={"title": title, "markdown": markdown, "target": target, "card_url": card_url},
)
response.raise_for_status()
return response.json()
async def update_task(
slug: str, target: str, title: str, markdown: str, card_url: str | None = None
) -> dict:
"""Перезаписывает задание по slug (та же ссылка)."""
base_url, headers = _tasks_config()
async with httpx.AsyncClient() as client:
response = await client.put(
f"{base_url}/tasks/{slug_from(slug)}",
headers=headers,
json={"title": title, "markdown": markdown, "target": target, "card_url": card_url},
)
response.raise_for_status()
return response.json()
async def get_task(slug: str) -> dict:
"""Задание целиком: {slug, url, title, target, card_url, markdown, ...}."""
base_url, headers = _tasks_config()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{base_url}/tasks/{slug_from(slug)}/raw", headers=headers
)
response.raise_for_status()
return response.json()
async def list_tasks(card_url: str | None = None, target: str | None = None) -> list[dict]:
"""Задания без текста, новые первыми; фильтры по карточке и адресату."""
base_url, headers = _tasks_config()
params = {key: value for key, value in (("card", card_url), ("target", target)) if value}
async with httpx.AsyncClient() as client:
response = await client.get(f"{base_url}/tasks", headers=headers, params=params)
response.raise_for_status()
return response.json()
+3
View File
@@ -0,0 +1,3 @@
mcp[cli]>=2.0
httpx>=0.28
python-dotenv>=1.0
+570
View File
@@ -0,0 +1,570 @@
# /// script
# requires-python = ">=3.10"
# dependencies = ["mcp>=2.1,<3", "httpx>=0.28", "python-dotenv>=1.0"]
# ///
"""trello-mcp — MCP-сервер (stdio) для связки Trello + qa-docs-service.
Запускается как обычный локальный процесс, Claude Code общается с ним
через stdin/stdout. Регистрация (пример, scope local):
claude mcp add trello-tools --scope local -- \\
~/projects/gp_mcp/trello-mcp/.venv/bin/python \\
~/projects/gp_mcp/trello-mcp/server.py
Либо как плагин trello-tools из маркетплейса gp-tools (см. README.md):
тогда сервер запускается через `uv run --script`, зависимости берутся из
блока `# /// script` выше, а настройки — из userConfig плагина.
Примечание: план изначально предполагал `mcp.server.fastmcp.FastMCP`
(SDK v1). В установленной версии `mcp` (2.x) декораторный API называется
`MCPServer` (`mcp.server.mcpserver.MCPServer`) — по сути тот же
декоратор `.tool()`, только с другим именем класса.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Literal
from dotenv import load_dotenv
from mcp.server.mcpserver import AcceptedElicitation, Context, Image, MCPServer
from mcp.shared.exceptions import MCPError
from pydantic import BaseModel, Field
import qa_client
import trello
load_dotenv(Path(__file__).parent / ".env")
# В плагине .env нет: Claude Code передаёт поля userConfig из
# .claude-plugin/plugin.json как CLAUDE_PLUGIN_OPTION_<KEY>. Уже заданные
# переменные (например, QA_TASKS_TOKEN из ~/.bashrc) не перетираются.
PLUGIN_OPTIONS = (
"TRELLO_API_KEY",
"TRELLO_TOKEN",
"TRELLO_BOARD_ID",
"TRELLO_CREATE_LISTS",
"TRELLO_ATTACHMENTS_DIR",
"QA_SERVICE_URL",
"QA_SERVICE_API_KEY",
"QA_TASKS_TOKEN",
)
for _name in PLUGIN_OPTIONS:
_value = os.environ.get(f"CLAUDE_PLUGIN_OPTION_{_name}")
if _value and not os.environ.get(_name):
os.environ[_name] = _value
mcp = MCPServer("trello-tools")
QA_DOC_TEMPLATE = """\
# <Заголовок задачи> — [ссылка на карточку Trello]
## Что сделано
2-4 предложения по сути изменений.
## Было
Как вело себя раньше, конкретно, с примером.
## Стало
Как работает сейчас.
## Как проверить
Каждый сценарий — отдельный заголовок третьего уровня, начинающийся со
слова «Сценарий» (в украинском документе — «Сценарій»). Под каждым таким
заголовком на страничке появятся галочка «Проверено», выбор «Работает /
Не работает» и поле комментария для тестировщика (что именно не
работает). Шаги — списком под заголовком.
### Сценарий 1: <короткое название>
1. Шаг
2. Шаг
Ожидаемый результат: ...
### Сценарий 2: <короткое название>
...
## Что могло зацепить
Смежные места / риски регрессии.
---
Дата, ветка/коммит — если уместно."""
SCENARIOS_NOTE = (
"Отметки тестировщиков хранятся отдельно от markdown и при обновлении "
"документа сохраняются — но привязаны к тексту заголовка сценария: "
"если переименовать «### Сценарий ...», его отметка сбросится. "
"Не меняй заголовки сценариев без необходимости."
)
@mcp.tool(
description=(
"Вызывай, когда пользователь прислал ссылку на карточку Trello — "
"чтобы подтянуть контекст задачи в разговор, включая обсуждение "
"в комментариях.\n\n"
"Вход: card_url — ссылка вида https://trello.com/c/abc123/45-название.\n"
"Выход: {id, name, desc, url, board_id, list_id, labels, attachments, "
"comments}.\n"
"labels — метки карточки [{name, color}].\n"
"attachments — только СПИСОК вложений, без содержимого: [{id, name, "
"kind, mime, size, date, url}], kind — image (картинка), file (файл), "
"card (другая карточка Trello), link (прочая ссылка). Содержимое "
"тяни через trello_get_attachment, только если оно нужно для задачи "
"(скриншот бага, макет, лог, ТЗ, связанная карточка) — не всё подряд.\n"
"comments — список {author, text, date} в хронологическом порядке "
"(старые → новые), до 20 последних.\n\n"
"Если в описании или комментариях есть ссылка на задание вида "
".../tasks/<slug> — это задание для фронта/бэка от другой сессии "
"Claude: прочитай его через get_dev_task. Все задания по карточке "
"можно найти через list_dev_tasks(card_url)."
)
)
async def trello_get_card(card_url: str) -> dict:
return await trello.get_card(card_url)
# Картинки больше этого не отдаём в контекст — только путь к файлу.
INLINE_IMAGE_MAX_BYTES = 5 * 1024 * 1024
# Текстовые файлы до этого размера отдаём прямо в ответе.
INLINE_TEXT_MAX_BYTES = 200 * 1024
_TEXT_MIMES = ("application/json", "application/xml", "application/x-yaml", "application/csv")
# Форматы, которые модель умеет смотреть как картинку; прочие (svg, heic…) — по path.
_INLINE_IMAGE_MIMES = ("image/png", "image/jpeg", "image/gif", "image/webp")
@mcp.tool(
description=(
"Достаёт одно вложение карточки Trello. Вызывай выборочно — когда "
"вложение из attachments в trello_get_card действительно нужно для "
"задачи.\n\n"
"Вход: card_id — id карточки (поле id из trello_get_card), "
"attachment_id — id вложения из attachments.\n"
"Что вернётся по kind:\n"
"- image — метаданные + сама картинка (png/jpeg/gif/webp до 5 МБ; "
"иначе — только path);\n"
"- file — метаданные + path к скачанному файлу; текстовые файлы до "
"200 КБ — ещё и text. Остальное (PDF, docx, архивы) читай по path "
"своими инструментами;\n"
"- card — связанная карточка целиком, как из trello_get_card;\n"
"- link — только метаданные и url, сам сайт не открывается.\n"
"Файлы > 25 МБ не скачиваются."
)
)
async def trello_get_attachment(card_id: str, attachment_id: str):
attachment = await trello.get_attachment(card_id, attachment_id)
if attachment["kind"] == "card":
return {**attachment, "card": await trello.get_card(attachment["url"])}
if attachment["kind"] == "link":
return attachment
data = attachment.pop("data")
mime = attachment["mime"] or ""
if mime in _INLINE_IMAGE_MIMES and len(data) <= INLINE_IMAGE_MAX_BYTES:
return [attachment, Image(data=data, format=mime.split("/", 1)[1])]
if (mime.startswith("text/") or mime in _TEXT_MIMES) and len(data) <= INLINE_TEXT_MAX_BYTES:
try:
attachment["text"] = data.decode("utf-8")
except UnicodeDecodeError:
pass
return attachment
@mcp.tool(
description=(
"Постит текстовый комментарий в карточку Trello — любой текст, "
"ссылка не обязательна.\n\n"
"Произвольные комментарии — ТОЛЬКО по прямой просьбе пользователя "
"(«напиши в карточку…», «оставь комментарий…»). Текст бери "
"дословно из просьбы; если сформулировать нужно тебе — сначала "
"покажи текст и дождись одобрения. Исключение — комментарий со "
"ссылкой после publish_qa_doc / publish_dev_task: его правила "
"описаны в этих инструментах.\n\n"
"Вход: card_id — полный id карточки (не shortLink; его отдаёт "
"trello_get_card в поле id), text — текст комментария.\n"
"Выход: {id} — id созданного комментария."
)
)
async def trello_add_comment(card_id: str, text: str) -> dict:
return await trello.add_comment(card_id, text)
class CardConfirmation(BaseModel):
# Включена по умолчанию: подтверждение — это нажатие «принять», вторая
# ступень в том же окне только путает (галочку не замечают, и карточка
# молча не создаётся). Снять её — тоже способ отказаться.
confirm: bool = Field(
default=True, description="Создать карточку именно с этим текстом"
)
def _card_preview(
list_name: str,
name: str,
desc: str,
template: dict | None = None,
labels: list[dict] | None = None,
) -> str:
lines = [f"Столбец: {list_name}"]
if template:
lines.append(f"Шаблон: {template['name']}")
if template["members"]:
lines.append(f" участники: {', '.join(template['members'])}")
if template["cover"]:
lines.append(f" обложка: {template['cover']['color']}")
if template["labels"]:
lines.append(f" метки шаблона: {', '.join(l['name'] or l['color'] for l in template['labels'])}")
lines += [f" чек-лист «{c['name']}»" for c in template["checklists"]]
if labels:
lines.append(f"Метки: {', '.join(label['name'] for label in labels)}")
return (
"Создать карточку в Trello?\n\n"
+ "\n".join(lines)
+ f"\nНазвание: {name}\n\n"
f"Описание:\n{desc or '(пусто)'}"
)
async def create_card_confirmed(
ctx: Context,
list_name: str,
name: str,
desc: str,
template_name: str | None = None,
label_names: list[str] | None = None,
) -> dict:
"""Создаёт карточку только после подтверждения пользователем в форме клиента.
Подтверждение запрашивается у пользователя напрямую (MCP elicitation),
в обход модели: карточка создаётся ровно с тем текстом, который он видел.
"""
target = await trello.resolve_create_list(list_name)
template = await trello.resolve_template(template_name) if template_name else None
labels = await trello.resolve_labels(label_names) if label_names else []
try:
answer = await ctx.elicit(
_card_preview(target["name"], name, desc, template, labels), CardConfirmation
)
except MCPError as error:
return {
"created": False,
"reason": (
"Клиент не смог показать пользователю окно подтверждения "
f"({error}). Карточка НЕ создана."
),
}
if not isinstance(answer, AcceptedElicitation):
return {
"created": False,
"reason": "Пользователь отклонил или закрыл окно подтверждения. Карточка НЕ создана.",
}
if not answer.data.confirm:
return {
"created": False,
"reason": (
"Пользователь принял окно, но снял галочку «Создать карточку». "
"Карточка НЕ создана."
),
}
card = await trello.create_card(
target["id"], name, desc, template, [label["id"] for label in labels]
)
return {
"created": True,
"list": target["name"],
"template": template["name"] if template else None,
"labels": [label["name"] for label in labels],
**card,
}
@mcp.tool(
description=(
"Список шаблонов карточек на доске Trello. Шаблон задаёт участников "
"карточки и оформление (обложку, метки, чек-листы) — по названию "
"видно, для какого типа задач он (баг, пожелание клиента, фронт…). "
"Вызывай перед trello_create_card. Только чтение — можно без спроса.\n\n"
"Выход: [{id, name, list, desc, members, cover, labels, checklists}]. "
"Всё это переносится в новую карточку автоматически; desc обычно "
"пуст, если нет — повтори его структуру в описании."
)
)
async def trello_list_card_templates() -> list[dict]:
return await trello.list_templates()
@mcp.tool(
description=(
"Список меток доски Trello [{id, name, color, uses}], самые "
"используемые первыми (uses — на скольких карточках стоит). Вызывай "
"перед trello_create_card, чтобы подобрать метки. Только чтение — "
"можно без спроса."
)
)
async def trello_list_labels() -> list[dict]:
return await trello.list_labels()
_create_targets = trello.create_targets()
@mcp.tool(
description=(
"Создаёт карточку в Trello. ВЫЗЫВАЙ ТОЛЬКО по прямой просьбе "
"пользователя создать карточку («создай карточку», «заведи задачу "
"в Trello» и т.п.). Не создавай карточки по своей инициативе — "
"даже если по ходу работы нашёл баг или это кажется полезным; "
"в таком случае максимум предложи.\n\n"
"Порядок:\n"
"1. Вызови trello_list_card_templates и выбери подходящий шаблон "
"(баг → «баг», пожелание клиента → «Пожелание клиента» и т.п.); если "
"пользователь назвал шаблон — бери его, если неочевидно — спроси. "
"Составь черновик: столбец, шаблон, метки, название, описание (markdown). "
"Если у шаблона есть desc — повтори его структуру (те же заголовки "
"и порядок) и заполни. Метки и чек-листы шаблона перенесутся сами — "
"в описании их не дублируй.\n"
" Подбери метки через trello_list_labels:\n"
" - по смыслу — тематические: раздел/модуль продукта, интеграция, "
"тип задачи (bug…), приоритет (Minor/Major/Critical…); ставь только "
"уверенно подходящие, лучше меньше, чем лишние;\n"
" - ТОЛЬКО по указанию пользователя — метки с именами людей и "
"статусные/процессные (Доробити, READY TO FRONT, NEED_HOTFIX, "
"WAITING…, Перевірити на проді и т.п.);\n"
" - метки, которые назвал пользователь, ставь всегда.\n"
" Метки шаблона добавятся сами, их в labels не повторяй.\n"
"2. Покажи черновик пользователю в чате ДОСЛОВНО и спроси, "
"создавать ли. Правки пользователя вноси и показывай снова.\n"
"3. Вызывай тул только после явного «да» и передавай ровно "
"одобренный текст, ничего не меняя.\n"
"Тул дополнительно сам покажет пользователю окно подтверждения; "
"если он откажет — карточка не создастся (created=false), "
"не пытайся вызвать повторно без новой просьбы.\n\n"
"Разрешённые столбцы (list_name — строго одно из): "
+ (
", ".join(f"«{name}»" for name in _create_targets)
if _create_targets
else "НЕ НАСТРОЕНЫ (TRELLO_CREATE_LISTS пуст) — создание недоступно"
)
+ ". Если пользователь не указал столбец и вариантов несколько — "
"спроси.\n\n"
"Вход: list_name, name — название карточки, desc — описание, "
"template — название шаблона из trello_list_card_templates "
"(опционально, без шаблона — пустая карточка), labels — названия "
"меток из trello_list_labels (опционально).\n"
"Выход: {created, reason} при отказе или {created, list, template, "
"labels, id, name, url}."
)
)
async def trello_create_card(
list_name: str,
name: str,
desc: str,
ctx: Context,
template: str | None = None,
labels: list[str] | None = None,
) -> dict:
return await create_card_confirmed(ctx, list_name, name, desc, template, labels)
@mcp.tool(
description=(
"Публикует НОВЫЙ QA-документ по итогам работы и отдаёт стабильную "
"ссылку. Если для этой же карточки Trello в этом разговоре уже "
"публиковался QA-документ и известен его slug (из url вида "
".../docs/<slug>) — используй вместо этого update_qa_doc, чтобы "
"обновить существующую ссылку, а не плодить новую при каждом "
"повторном прогоне. Если документ не привязан к карточке Trello — "
"используй publish_standalone_qa_doc.\n\n"
"markdown должен следовать шаблону:\n\n"
f"{QA_DOC_TEMPLATE}\n\n"
"После публикации, если известен card_id карточки Trello, вызови "
'trello_add_comment(card_id, f"QA: {url}") с URL из ответа этого тула, '
"чтобы тестировщик увидел ссылку в комментарии к карточке.\n\n"
"Вход: title, markdown (по шаблону выше), card_url — опционально, "
"ссылка на исходную карточку Trello для шапки странички.\n"
"Выход: {slug, url}."
)
)
async def publish_qa_doc(
title: str, markdown: str, card_url: str | None = None
) -> dict:
return await qa_client.publish_doc(title, markdown, card_url)
@mcp.tool(
description=(
"Обновляет УЖЕ ОПУБЛИКОВАННЫЙ QA-документ на месте — тот же slug, "
"та же ссылка, тестировщику не нужно переходить по новой. Используй "
"вместо publish_qa_doc, когда повторно дорабатываешь ту же карточку "
"Trello в рамках этого разговора и знаешь slug из предыдущего "
"publish_qa_doc/update_qa_doc (это последний сегмент url — "
".../docs/<slug>).\n\n"
"markdown — по тому же шаблону, что и в publish_qa_doc:\n\n"
f"{QA_DOC_TEMPLATE}\n\n"
"Если slug не найден на сервере — вернётся ошибка 404; в этом "
"случае используй publish_qa_doc, чтобы опубликовать заново.\n\n"
"Чтобы поправить документ, которого нет в текущем разговоре, сначала "
"скачай его через get_qa_doc. Поля перезаписываются целиком: передай "
"card_url и problem заново (их отдаёт get_qa_doc), иначе они "
f"очистятся. {SCENARIOS_NOTE}\n\n"
"Вход: slug (или полная ссылка), title, markdown (по шаблону выше), "
"card_url и problem — опционально.\n"
"Выход: {slug, url}."
)
)
async def update_qa_doc(
slug: str,
title: str,
markdown: str,
card_url: str | None = None,
problem: str | None = None,
) -> dict:
return await qa_client.update_doc(slug, title, markdown, card_url, problem)
@mcp.tool(
description=(
"Публикует документ проверки, НЕ привязанный к карточке Trello, — "
"например, когда пользователь просит выгрузить проверку по проблеме, "
"для которой карточки нет. Описание проблемы выводится отдельным "
"блоком в шапке странички, поэтому в markdown его не дублируй.\n\n"
"markdown — по тому же шаблону, что и в publish_qa_doc, только без "
"ссылки на карточку в заголовке:\n\n"
f"{QA_DOC_TEMPLATE}\n\n"
"Обновлять такой документ — через update_qa_doc, передавая problem "
"заново.\n\n"
"Вход: title, problem — описание проблемы (что не так, где, как "
"воспроизводится; можно markdown), markdown (по шаблону выше).\n"
"Выход: {slug, url}."
)
)
async def publish_standalone_qa_doc(title: str, problem: str, markdown: str) -> dict:
return await qa_client.publish_doc(title, markdown, card_url=None, problem=problem)
@mcp.tool(
description=(
"Скачивает уже опубликованный QA-документ: исходный markdown, "
"метаданные и отметки тестировщиков по каждому сценарию.\n\n"
"Используй, когда нужно:\n"
"- поправить существующий документ — скачай, отредактируй markdown "
"и отправь через update_qa_doc с тем же slug (ссылка не изменится);\n"
"- узнать результаты проверки — какие сценарии тестировщик проверил, "
"какие работают, какие нет и что он написал в комментариях.\n\n"
"Вход: slug или полная ссылка на документ (.../docs/<slug>).\n"
"Выход: {slug, url, title, markdown, card_url, problem, created_at, "
"updated_at, scenarios}. scenarios — список в порядке документа: "
"{scenario — заголовок, checked — проверен ли, passed — true "
"работает / false не работает / null не проверен, comment — "
"комментарий тестировщика (что именно не работает), updated_at}."
)
)
async def get_qa_doc(slug: str) -> dict:
return await qa_client.get_doc(slug)
TASK_TARGETS = "front | back"
TASK_GUIDE = (
"markdown — в свободной форме, но пишется для ДРУГОЙ нейронки, у которой "
"нет этого разговора и, возможно, нет доступа к этому репозиторию. "
"Поэтому явно: контекст задачи и зачем это нужно; что именно сделать; "
"контракт на стыке (эндпоинты, методы, поля запроса/ответа с типами и "
"примерами, коды ошибок); что уже сделано на нашей стороне и где "
"(ветка/файлы); открытые вопросы. Заголовок в markdown не повторяй — "
"он берётся из title."
)
@mcp.tool(
description=(
"Публикует задание для фронта или бэка — документ для другой "
"нейронки (Claude на той стороне), которая будет его делать.\n\n"
"Когда: по ходу работы над карточкой Trello понял, что нужна "
"работа на другой стороне (например, бэку нужен эндпоинт от фронта "
"или наоборот). ПРЕДЛОЖИ пользователю одной фразой: что за задание "
"и кому, и что оставишь ссылку в карточке. Если согласился (даже "
"коротко: «да», «давай», «ок») — публикуй сразу, дословно текст "
"на одобрение показывать не нужно. Без согласия не публикуй.\n\n"
f"{TASK_GUIDE}\n\n"
"После публикации, если известен card_id карточки, оставь "
"комментарий через trello_add_comment — ровно три строки:\n"
"Задание для <фронта|бэка>: <title>\n<url>\n"
'Получить: curl -H "Authorization: Bearer $QA_TASKS_TOKEN" <url>\n\n'
"Если по этой же карточке задание для той же стороны уже "
"публиковалось (slug известен или находится через list_dev_tasks) — "
"обнови его через update_dev_task, а не плоди новое.\n\n"
f"Вход: target ({TASK_TARGETS}) — кому задание, title, markdown, "
"card_url — ссылка на карточку Trello, если есть.\n"
"Выход: {slug, url, title, target, card_url, created_at, updated_at}."
)
)
async def publish_dev_task(
target: Literal["front", "back"],
title: str,
markdown: str,
card_url: str | None = None,
) -> dict:
return await qa_client.publish_task(target, title, markdown, card_url)
@mcp.tool(
description=(
"Обновляет уже опубликованное задание для фронта/бэка на месте — "
"ссылка не меняется. Правила согласия те же, что у "
"publish_dev_task: предложил — пользователь согласился — обновляешь. "
"Если текущего текста нет в разговоре, сначала скачай его через "
"get_dev_task. Поля перезаписываются целиком.\n\n"
f"{TASK_GUIDE}\n\n"
f"Вход: slug (или ссылка), target ({TASK_TARGETS}), title, markdown, "
"card_url.\nВыход: как у publish_dev_task."
)
)
async def update_dev_task(
slug: str,
target: Literal["front", "back"],
title: str,
markdown: str,
card_url: str | None = None,
) -> dict:
return await qa_client.update_task(slug, target, title, markdown, card_url)
@mcp.tool(
description=(
"Читает задание для фронта/бэка по slug или ссылке вида "
".../tasks/<slug> — например, из комментария в карточке Trello. "
"Вызывай без спроса, это только чтение.\n\n"
"Выход: {slug, url, title, target, card_url, markdown, created_at, "
"updated_at}."
)
)
async def get_dev_task(slug: str) -> dict:
return await qa_client.get_task(slug)
@mcp.tool(
description=(
"Список заданий для фронта/бэка (без текста, новые первыми). "
"Используй, когда пользователь прислал карточку Trello — чтобы "
"проверить, нет ли по ней заданий для твоей стороны; или чтобы "
"найти slug уже опубликованного задания перед обновлением. "
"Вызывай без спроса, это только чтение.\n\n"
f"Вход: card_url — опционально, target ({TASK_TARGETS}) — опционально.\n"
"Выход: [{slug, url, title, target, card_url, created_at, updated_at}]."
)
)
async def list_dev_tasks(
card_url: str | None = None, target: Literal["front", "back"] | None = None
) -> list[dict]:
return await qa_client.list_tasks(card_url, target)
if __name__ == "__main__":
mcp.run()
+283
View File
@@ -0,0 +1,283 @@
import json
import sys
from pathlib import Path
import httpx
import pytest
import respx
from mcp.server.mcpserver import AcceptedElicitation, DeclinedElicitation
from mcp.shared.exceptions import MCPError
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import server # noqa: E402
import trello # noqa: E402
LISTS_URL = "https://api.trello.com/1/boards/board1/lists"
CARDS_URL = "https://api.trello.com/1/cards"
BOARD_LISTS = [
{"id": "list-backlog", "name": "Бэклог"},
{"id": "list-bugs", "name": "Баги с прода"},
{"id": "list-done", "name": "Готово"},
]
@pytest.fixture(autouse=True)
def trello_env(monkeypatch):
monkeypatch.setenv("TRELLO_API_KEY", "key123")
monkeypatch.setenv("TRELLO_TOKEN", "tok456")
monkeypatch.setenv("TRELLO_BOARD_ID", "board1")
monkeypatch.setenv("TRELLO_CREATE_LISTS", " Бэклог , Баги с прода,")
class FakeContext:
def __init__(self, answer=None, error=None):
self.answer = answer
self.error = error
self.messages = []
async def elicit(self, message, schema):
self.messages.append(message)
if self.error:
raise self.error
return self.answer
def test_create_targets_from_env():
assert trello.create_targets() == ["Бэклог", "Баги с прода"]
@pytest.mark.asyncio
async def test_resolve_rejects_list_not_in_env():
# «Готово» есть на доске, но не разрешён — в API даже не ходим.
with respx.mock:
with pytest.raises(ValueError, match="не разрешён"):
await trello.resolve_create_list("Готово")
@pytest.mark.asyncio
async def test_resolve_finds_list_case_insensitive():
with respx.mock:
respx.get(LISTS_URL).mock(return_value=httpx.Response(200, json=BOARD_LISTS))
target = await trello.resolve_create_list("баги С прода")
assert target == {"id": "list-bugs", "name": "Баги с прода"}
@pytest.mark.asyncio
async def test_resolve_list_missing_on_board():
with respx.mock:
respx.get(LISTS_URL).mock(return_value=httpx.Response(200, json=BOARD_LISTS[2:]))
with pytest.raises(ValueError, match="не найден"):
await trello.resolve_create_list("Бэклог")
@pytest.mark.asyncio
async def test_confirmed_card_is_created_with_exact_text():
ctx = FakeContext(AcceptedElicitation(data=server.CardConfirmation(confirm=True)))
with respx.mock:
respx.get(LISTS_URL).mock(return_value=httpx.Response(200, json=BOARD_LISTS))
create = respx.post(CARDS_URL).mock(
return_value=httpx.Response(
200, json={"id": "c1", "name": "Падает экспорт", "shortUrl": "https://trello.com/c/xyz"}
)
)
result = await server.create_card_confirmed(ctx, "Бэклог", "Падает экспорт", "Шаги: ...")
assert json.loads(create.calls.last.request.content) == {
"idList": "list-backlog",
"name": "Падает экспорт",
"desc": "Шаги: ...",
}
assert result == {
"created": True,
"list": "Бэклог",
"template": None,
"labels": [],
"id": "c1",
"name": "Падает экспорт",
"url": "https://trello.com/c/xyz",
}
assert "Падает экспорт" in ctx.messages[0] and "Шаги: ..." in ctx.messages[0]
@pytest.mark.parametrize(
"ctx",
[
FakeContext(DeclinedElicitation()),
FakeContext(AcceptedElicitation(data=server.CardConfirmation(confirm=False))),
FakeContext(error=MCPError(code=-32601, message="elicitation not supported")),
],
)
@pytest.mark.asyncio
async def test_card_not_created_without_confirmation(ctx):
with respx.mock:
respx.get(LISTS_URL).mock(return_value=httpx.Response(200, json=BOARD_LISTS))
create = respx.post(CARDS_URL)
result = await server.create_card_confirmed(ctx, "Бэклог", "T", "D")
assert result["created"] is False
assert not create.called
BOARD_CARDS_URL = "https://api.trello.com/1/boards/board1/cards"
BOARD_CARDS = [
{"id": "card-plain", "name": "Обычная", "desc": "", "idList": "list-backlog", "labels": [], "isTemplate": False},
{
"id": "tpl-bug",
"name": "баг",
"desc": "## Шаги\n## Ожидаемо",
"idList": "list-backlog",
"labels": [{"name": "Bug", "color": "red"}],
"isTemplate": True,
"idMembers": ["m1"],
"cover": {"color": "red", "size": "normal", "brightness": "light", "yPosition": 0.5},
},
# Архивный шаблон в архивном списке — тоже нужен.
{"id": "tpl-old", "name": "старый", "desc": "", "idList": "list-archived", "labels": [], "isTemplate": True},
]
TEMPLATE_CHECKLISTS = {
"tpl-bug": [{"name": "DoD", "checkItems": [{"name": "Тесты"}]}],
"tpl-old": [],
}
ALL_LISTS = BOARD_LISTS + [{"id": "list-archived", "name": "How to use this Board"}]
BOARD_MEMBERS_URL = "https://api.trello.com/1/boards/board1/members"
BOARD_LABELS_URL = "https://api.trello.com/1/boards/board1/labels"
BOARD_LABELS = [
{"id": "lb-bug", "name": "bug", "color": "orange", "uses": 102},
{"id": "lb-np", "name": "Нова Пошта", "color": "yellow_light", "uses": 41},
{"id": "lb-np-old", "name": "нова пошта", "color": "yellow", "uses": 1},
{"id": "lb-empty", "name": "", "color": "purple", "uses": 2},
]
def _mock_board():
respx.get(LISTS_URL).mock(return_value=httpx.Response(200, json=ALL_LISTS))
respx.get(BOARD_MEMBERS_URL).mock(
return_value=httpx.Response(200, json=[{"id": "m1", "fullName": "Alexey Poddubniy"}])
)
respx.get(BOARD_LABELS_URL).mock(return_value=httpx.Response(200, json=BOARD_LABELS))
for card_id, checklists in TEMPLATE_CHECKLISTS.items():
respx.get(f"{CARDS_URL}/{card_id}/checklists").mock(
return_value=httpx.Response(200, json=checklists)
)
return respx.get(BOARD_CARDS_URL).mock(return_value=httpx.Response(200, json=BOARD_CARDS))
@pytest.mark.asyncio
async def test_list_templates_only_template_cards():
with respx.mock:
cards = _mock_board()
templates = await trello.list_templates()
assert cards.calls.last.request.url.params["filter"] == "all"
assert templates == [
{
"id": "tpl-bug",
"name": "баг",
"list": "Бэклог",
"desc": "## Шаги\n## Ожидаемо",
"members": ["Alexey Poddubniy"],
"cover": {"color": "red", "size": "normal", "brightness": "light"},
"labels": [{"name": "Bug", "color": "red"}],
"checklists": [{"name": "DoD", "items": ["Тесты"]}],
},
{
"id": "tpl-old",
"name": "старый",
"list": "How to use this Board",
"desc": "",
"members": [],
"cover": None,
"labels": [],
"checklists": [],
},
]
@pytest.mark.asyncio
async def test_resolve_template_unknown_lists_available():
with respx.mock:
_mock_board()
with pytest.raises(ValueError, match="Есть: баг, старый"):
await trello.resolve_template("Фича")
@pytest.mark.asyncio
async def test_card_from_template_copies_and_fixes_desc_and_flag():
ctx = FakeContext(AcceptedElicitation(data=server.CardConfirmation()))
with respx.mock:
_mock_board()
create = respx.post(CARDS_URL).mock(
return_value=httpx.Response(
200,
json={
"id": "c1",
"name": "Падает экспорт",
"shortUrl": "https://trello.com/c/xyz",
# Trello скопировал описание и флаг шаблона.
"desc": "## Шаги\n## Ожидаемо",
"isTemplate": True,
"idLabels": ["lb-tpl"],
},
)
)
fix = respx.put(f"{CARDS_URL}/c1").mock(return_value=httpx.Response(200, json={}))
result = await server.create_card_confirmed(
ctx,
"Бэклог",
"Падает экспорт",
"## Шаги\n1. Нажать",
template_name="БАГ",
label_names=["НОВА ПОШТА", "bug"],
)
sent = json.loads(create.calls.last.request.content)
assert sent["idCardSource"] == "tpl-bug"
assert {"checklists", "labels", "members"} <= set(sent["keepFromSource"].split(","))
assert json.loads(fix.calls.last.request.content) == {
"desc": "## Шаги\n1. Нажать",
"isTemplate": False,
"cover": {"color": "red", "size": "normal", "brightness": "light"},
# Метки шаблона сохраняются, выбранные добавляются; при дублях
# названий берётся самая используемая метка.
"idLabels": "lb-tpl,lb-np,lb-bug",
}
assert result["template"] == "баг"
assert result["labels"] == ["Нова Пошта", "bug"]
preview = ctx.messages[0]
assert "Шаблон: баг" in preview and "чек-лист «DoD»" in preview
assert "участники: Alexey Poddubniy" in preview and "обложка: red" in preview
assert "Метки: Нова Пошта, bug" in preview
@pytest.mark.asyncio
async def test_list_labels_skips_unnamed_and_sorts():
with respx.mock:
_mock_board()
labels = await trello.list_labels()
assert [label["name"] for label in labels] == ["bug", "Нова Пошта", "нова пошта"]
@pytest.mark.asyncio
async def test_unknown_label_fails_before_confirmation():
ctx = FakeContext(AcceptedElicitation(data=server.CardConfirmation()))
with respx.mock:
_mock_board()
create = respx.post(CARDS_URL)
with pytest.raises(ValueError, match="'Нет такой'"):
await server.create_card_confirmed(
ctx, "Бэклог", "T", "D", label_names=["bug", "Нет такой"]
)
assert not ctx.messages and not create.called
def test_confirmation_checked_by_default():
assert server.CardConfirmation().confirm is True
+78
View File
@@ -0,0 +1,78 @@
import sys
from pathlib import Path
import pytest
from mcp.server.mcpserver import Image
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import server # noqa: E402
def _fake_attachment(monkeypatch, **fields):
attachment = {
"id": "a1",
"name": "f",
"kind": "file",
"mime": None,
"size": 0,
"date": None,
"url": "https://trello.com/1/x",
**fields,
}
async def fake_get_attachment(card_id, attachment_id):
return dict(attachment)
monkeypatch.setattr(server.trello, "get_attachment", fake_get_attachment)
@pytest.mark.asyncio
async def test_png_returned_as_image(monkeypatch):
_fake_attachment(monkeypatch, kind="image", mime="image/png", data=b"\x89PNG", path="/tmp/x.png")
meta, image = await server.trello_get_attachment("card1", "a1")
assert isinstance(image, Image)
assert "data" not in meta and meta["path"] == "/tmp/x.png"
@pytest.mark.asyncio
async def test_svg_only_path(monkeypatch):
_fake_attachment(monkeypatch, kind="image", mime="image/svg+xml", data=b"<svg/>", path="/tmp/x.svg")
result = await server.trello_get_attachment("card1", "a1")
assert isinstance(result, dict) and result["path"] == "/tmp/x.svg"
@pytest.mark.asyncio
async def test_small_text_file_inlined(monkeypatch):
_fake_attachment(monkeypatch, mime="text/plain", data="ошибка 500".encode(), path="/tmp/x.txt")
result = await server.trello_get_attachment("card1", "a1")
assert result["text"] == "ошибка 500"
@pytest.mark.asyncio
async def test_pdf_only_path(monkeypatch):
_fake_attachment(monkeypatch, mime="application/pdf", data=b"%PDF", path="/tmp/x.pdf")
result = await server.trello_get_attachment("card1", "a1")
assert "text" not in result and "data" not in result
@pytest.mark.asyncio
async def test_linked_card_fetched(monkeypatch):
_fake_attachment(monkeypatch, kind="card", url="https://trello.com/c/zzz/1-x")
async def fake_get_card(url):
return {"id": "linked", "url": url}
monkeypatch.setattr(server.trello, "get_card", fake_get_card)
result = await server.trello_get_attachment("card1", "a1")
assert result["card"] == {"id": "linked", "url": "https://trello.com/c/zzz/1-x"}
+144
View File
@@ -0,0 +1,144 @@
import json
import sys
from pathlib import Path
import httpx
import pytest
import respx
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import qa_client # noqa: E402
@pytest.fixture(autouse=True)
def qa_env(monkeypatch):
monkeypatch.setenv("QA_SERVICE_URL", "https://qa.example.com")
monkeypatch.setenv("QA_SERVICE_API_KEY", "svc-key-789")
@pytest.mark.asyncio
async def test_publish_doc_posts_to_docs():
with respx.mock:
route = respx.post("https://qa.example.com/docs").mock(
return_value=httpx.Response(200, json={"slug": "abc", "url": "https://qa.example.com/docs/abc"})
)
result = await qa_client.publish_doc("Заголовок", "текст", "https://trello.com/c/x")
assert route.called
request = route.calls.last.request
assert request.headers["X-API-Key"] == "svc-key-789"
assert result == {"slug": "abc", "url": "https://qa.example.com/docs/abc"}
@pytest.mark.asyncio
async def test_update_doc_puts_to_docs_slug():
with respx.mock:
route = respx.put("https://qa.example.com/docs/abc").mock(
return_value=httpx.Response(200, json={"slug": "abc", "url": "https://qa.example.com/docs/abc"})
)
result = await qa_client.update_doc(
"abc", "Новый заголовок", "новый текст", "https://trello.com/c/x"
)
assert route.called
request = route.calls.last.request
assert request.headers["X-API-Key"] == "svc-key-789"
assert result == {"slug": "abc", "url": "https://qa.example.com/docs/abc"}
@pytest.mark.asyncio
async def test_update_doc_missing_slug_raises():
with respx.mock:
respx.put("https://qa.example.com/docs/does-not-exist").mock(
return_value=httpx.Response(404, json={"detail": "Doc not found"})
)
with pytest.raises(httpx.HTTPStatusError):
await qa_client.update_doc("does-not-exist", "T", "текст")
def test_update_doc_requires_config(monkeypatch):
monkeypatch.delenv("QA_SERVICE_URL", raising=False)
monkeypatch.delenv("QA_SERVICE_API_KEY", raising=False)
with pytest.raises(qa_client.QaServiceConfigError):
qa_client._config()
@pytest.mark.asyncio
async def test_get_doc_fetches_raw_by_slug_or_url():
doc = {"slug": "abc", "markdown": "текст", "scenarios": []}
with respx.mock:
route = respx.get("https://qa.example.com/docs/abc/raw").mock(
return_value=httpx.Response(200, json=doc)
)
by_slug = await qa_client.get_doc("abc")
by_url = await qa_client.get_doc("https://qa.example.com/docs/abc/")
assert route.call_count == 2
assert route.calls.last.request.headers["X-API-Key"] == "svc-key-789"
assert by_slug == by_url == doc
@pytest.mark.asyncio
async def test_publish_doc_sends_problem():
with respx.mock:
route = respx.post("https://qa.example.com/docs").mock(
return_value=httpx.Response(200, json={"slug": "abc", "url": "u"})
)
await qa_client.publish_doc("T", "текст", problem="Падает экспорт")
body = json.loads(route.calls.last.request.content)
assert body["problem"] == "Падает экспорт"
assert body["card_url"] is None
@pytest.mark.asyncio
async def test_publish_task_uses_bearer_api_key():
with respx.mock:
route = respx.post("https://qa.example.com/tasks").mock(
return_value=httpx.Response(200, json={"slug": "t1"})
)
await qa_client.publish_task("front", "Кнопка", "текст", "https://trello.com/c/x")
request = route.calls.last.request
assert request.headers["Authorization"] == "Bearer svc-key-789"
assert json.loads(request.content) == {
"title": "Кнопка",
"markdown": "текст",
"target": "front",
"card_url": "https://trello.com/c/x",
}
@pytest.mark.asyncio
async def test_tasks_work_with_tasks_token_only(monkeypatch):
# Claude на фронте: основного ключа нет, только токен заданий.
monkeypatch.delenv("QA_SERVICE_API_KEY")
monkeypatch.setenv("QA_TASKS_TOKEN", "tasks-tok")
with respx.mock:
route = respx.get("https://qa.example.com/tasks/t1/raw").mock(
return_value=httpx.Response(200, json={"slug": "t1"})
)
await qa_client.get_task("https://qa.example.com/tasks/t1")
assert route.calls.last.request.headers["Authorization"] == "Bearer tasks-tok"
with pytest.raises(qa_client.QaServiceConfigError):
qa_client._config()
@pytest.mark.asyncio
async def test_update_and_list_tasks():
with respx.mock:
put = respx.put("https://qa.example.com/tasks/t1").mock(
return_value=httpx.Response(200, json={"slug": "t1"})
)
listing = respx.get("https://qa.example.com/tasks").mock(
return_value=httpx.Response(200, json=[])
)
await qa_client.update_task("t1", "back", "T", "v2")
await qa_client.list_tasks(card_url="https://trello.com/c/abc", target=None)
assert json.loads(put.calls.last.request.content)["target"] == "back"
assert dict(listing.calls.last.request.url.params) == {"card": "https://trello.com/c/abc"}
+266
View File
@@ -0,0 +1,266 @@
import sys
from pathlib import Path
import httpx
import pytest
import respx
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import trello # noqa: E402
@pytest.mark.parametrize(
"url, expected",
[
("https://trello.com/c/abc123/45-название-карточки", "abc123"),
("https://trello.com/c/XyZ987", "XyZ987"),
],
)
def test_parse_short_link_ok(url, expected):
assert trello.parse_short_link(url) == expected
def test_parse_short_link_invalid():
with pytest.raises(ValueError):
trello.parse_short_link("https://example.com/not-a-trello-link")
@pytest.mark.asyncio
async def test_get_card_builds_request(monkeypatch):
monkeypatch.setenv("TRELLO_API_KEY", "key123")
monkeypatch.setenv("TRELLO_TOKEN", "tok456")
with respx.mock:
route = respx.get("https://api.trello.com/1/cards/abc123").mock(
return_value=httpx.Response(
200,
json={
"id": "5f0000000000000000000001",
"name": "Карточка",
"desc": "Описание",
"shortUrl": "https://trello.com/c/abc123",
"idBoard": "board1",
"idList": "list1",
"labels": [
{"id": "l1", "name": "Баг", "color": "red"},
{"id": "l2", "name": "", "color": "green"},
],
"attachments": [
{
"id": "a1",
"name": "screen.png",
"url": "https://trello.com/1/cards/c/attachments/a1/download/screen.png",
"mimeType": "image/png",
"bytes": 1024,
"isUpload": True,
"date": "2026-01-01T00:00:00.000Z",
},
{
"id": "a2",
"name": "Связанная",
"url": "https://trello.com/c/zzz999/7-linked",
"mimeType": "",
"bytes": None,
"isUpload": False,
"date": "2026-01-01T00:00:00.000Z",
},
],
},
)
)
actions_route = respx.get(
"https://api.trello.com/1/cards/abc123/actions"
).mock(
return_value=httpx.Response(
200,
json=[
{
"data": {"text": "второй комментарий"},
"date": "2026-01-02T00:00:00.000Z",
"memberCreator": {"fullName": "Аня"},
},
{
"data": {"text": "первый комментарий"},
"date": "2026-01-01T00:00:00.000Z",
"memberCreator": {"fullName": "Максим"},
},
],
)
)
result = await trello.get_card("https://trello.com/c/abc123/45-title")
assert route.called
assert actions_route.called
request = route.calls.last.request
assert request.url.params["key"] == "key123"
assert request.url.params["token"] == "tok456"
assert request.url.params["attachments"] == "true"
assert result.pop("labels") == [
{"name": "Баг", "color": "red"},
{"name": "", "color": "green"},
]
attachments = result.pop("attachments")
assert [(a["id"], a["kind"]) for a in attachments] == [("a1", "image"), ("a2", "card")]
assert attachments[0]["mime"] == "image/png" and attachments[0]["size"] == 1024
assert result == {
"id": "5f0000000000000000000001",
"name": "Карточка",
"desc": "Описание",
"url": "https://trello.com/c/abc123",
"board_id": "board1",
"list_id": "list1",
"comments": [
{"author": "Максим", "text": "первый комментарий", "date": "2026-01-01T00:00:00.000Z"},
{"author": "Аня", "text": "второй комментарий", "date": "2026-01-02T00:00:00.000Z"},
],
}
@pytest.mark.asyncio
async def test_get_comments_builds_request(monkeypatch):
monkeypatch.setenv("TRELLO_API_KEY", "key123")
monkeypatch.setenv("TRELLO_TOKEN", "tok456")
with respx.mock:
route = respx.get("https://api.trello.com/1/cards/card1/actions").mock(
return_value=httpx.Response(
200,
json=[
{
"data": {"text": "коммент без автора"},
"date": "2026-01-01T00:00:00.000Z",
"memberCreator": None,
}
],
)
)
result = await trello.get_comments("card1")
assert route.called
request = route.calls.last.request
assert request.url.params["filter"] == "commentCard"
assert result == [
{"author": "?", "text": "коммент без автора", "date": "2026-01-01T00:00:00.000Z"}
]
@pytest.mark.asyncio
async def test_add_comment_builds_request(monkeypatch):
monkeypatch.setenv("TRELLO_API_KEY", "key123")
monkeypatch.setenv("TRELLO_TOKEN", "tok456")
with respx.mock:
route = respx.post(
"https://api.trello.com/1/cards/card1/actions/comments"
).mock(return_value=httpx.Response(200, json={"id": "comment1"}))
result = await trello.add_comment("card1", "QA: https://qa.example/docs/xyz")
assert route.called
request = route.calls.last.request
assert request.url.params["text"] == "QA: https://qa.example/docs/xyz"
assert result == {"id": "comment1"}
def test_get_card_requires_credentials(monkeypatch):
monkeypatch.delenv("TRELLO_API_KEY", raising=False)
monkeypatch.delenv("TRELLO_TOKEN", raising=False)
with pytest.raises(trello.TrelloConfigError):
trello._credentials()
@pytest.mark.parametrize(
"attachment, kind",
[
({"url": "https://trello.com/c/abc/1-x", "isUpload": False}, "card"),
({"url": "https://trello.com/1/.../a.png", "isUpload": True, "mimeType": "image/jpeg"}, "image"),
({"url": "https://trello.com/1/.../a.pdf", "isUpload": True, "mimeType": "application/pdf"}, "file"),
({"url": "https://figma.com/file/xyz", "isUpload": False}, "link"),
],
)
def test_attachment_kind(attachment, kind):
assert trello._attachment_kind(attachment) == kind
ATTACHMENT_META_URL = "https://api.trello.com/1/cards/card1/attachments/a1"
DOWNLOAD_URL = "https://trello.com/1/cards/card1/attachments/a1/download/лог ошибки.txt"
@pytest.fixture
def trello_creds(monkeypatch, tmp_path):
monkeypatch.setenv("TRELLO_API_KEY", "key123")
monkeypatch.setenv("TRELLO_TOKEN", "tok456")
monkeypatch.setenv("TRELLO_ATTACHMENTS_DIR", str(tmp_path))
return tmp_path
@pytest.mark.asyncio
async def test_get_attachment_downloads_upload_with_oauth(trello_creds):
meta = {
"id": "a1",
"name": "лог ошибки.txt",
"url": DOWNLOAD_URL,
"mimeType": "text/plain",
"bytes": 5,
"isUpload": True,
}
with respx.mock:
respx.get(ATTACHMENT_META_URL).mock(return_value=httpx.Response(200, json=meta))
download = respx.get(DOWNLOAD_URL).mock(return_value=httpx.Response(200, content=b"oops!"))
result = await trello.get_attachment("card1", "a1")
assert download.calls.last.request.headers["Authorization"] == (
'OAuth oauth_consumer_key="key123", oauth_token="tok456"'
)
assert result["data"] == b"oops!"
assert Path(result["path"]).read_bytes() == b"oops!"
assert Path(result["path"]).parent == trello_creds / "card1"
@pytest.mark.asyncio
async def test_get_attachment_link_not_downloaded(trello_creds):
meta = {"id": "a1", "name": "Макет", "url": "https://figma.com/file/x", "isUpload": False}
with respx.mock:
respx.get(ATTACHMENT_META_URL).mock(return_value=httpx.Response(200, json=meta))
result = await trello.get_attachment("card1", "a1")
assert result["kind"] == "link"
assert "data" not in result and "path" not in result
@pytest.mark.asyncio
async def test_get_attachment_refuses_non_trello_upload_host(trello_creds):
meta = {
"id": "a1",
"name": "x.pdf",
"url": "https://evil.example.com/x.pdf",
"mimeType": "application/pdf",
"bytes": 5,
"isUpload": True,
}
with respx.mock:
respx.get(ATTACHMENT_META_URL).mock(return_value=httpx.Response(200, json=meta))
with pytest.raises(ValueError, match="не на Trello"):
await trello.get_attachment("card1", "a1")
@pytest.mark.asyncio
async def test_get_attachment_too_big(trello_creds):
meta = {
"id": "a1",
"name": "video.mp4",
"url": DOWNLOAD_URL,
"mimeType": "video/mp4",
"bytes": trello.MAX_ATTACHMENT_BYTES + 1,
"isUpload": True,
}
with respx.mock:
respx.get(ATTACHMENT_META_URL).mock(return_value=httpx.Response(200, json=meta))
download = respx.get(DOWNLOAD_URL)
with pytest.raises(ValueError, match="больше"):
await trello.get_attachment("card1", "a1")
assert not download.called
+444
View File
@@ -0,0 +1,444 @@
"""Тонкий async-клиент Trello REST API."""
from __future__ import annotations
import asyncio
import os
import re
import tempfile
from pathlib import Path
import httpx
TRELLO_API_BASE = "https://api.trello.com/1"
_SHORT_LINK_RE = re.compile(r"trello\.com/c/([A-Za-z0-9]+)")
_ATTACHMENT_FIELDS = "id,name,url,mimeType,bytes,isUpload,date"
# Больше этого вложение не качаем — неожиданно тяжёлый файл не должен
# забивать диск и контекст.
MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024
class TrelloConfigError(RuntimeError):
"""Не заданы обязательные переменные окружения для Trello API."""
def _credentials() -> tuple[str, str]:
key = os.environ.get("TRELLO_API_KEY")
token = os.environ.get("TRELLO_TOKEN")
if not key or not token:
raise TrelloConfigError(
"TRELLO_API_KEY / TRELLO_TOKEN не заданы (проверь .env)"
)
return key, token
def parse_short_link(card_url: str) -> str:
"""Достаёт shortLink из ссылки вида https://trello.com/c/abc123/45-title."""
match = _SHORT_LINK_RE.search(card_url)
if not match:
raise ValueError(f"Не похоже на ссылку на карточку Trello: {card_url!r}")
return match.group(1)
async def get_card(card_url: str) -> dict:
"""Возвращает {id, name, desc, url, board_id, list_id, labels, attachments, comments}.
comments — see get_comments(): список {author, text, date}, для контекста задачи.
labels — [{name, color}]. attachments — только список (см. _attachment_summary),
содержимое тянется отдельно через get_attachment().
"""
key, token = _credentials()
short_link = parse_short_link(card_url)
async with httpx.AsyncClient() as client:
response = await client.get(
f"{TRELLO_API_BASE}/cards/{short_link}",
params={
"key": key,
"token": token,
"fields": "name,desc,shortUrl,idBoard,idList,labels",
"attachments": "true",
"attachment_fields": _ATTACHMENT_FIELDS,
},
)
response.raise_for_status()
data = response.json()
comments = await get_comments(short_link)
return {
"id": data["id"],
"name": data["name"],
"desc": data["desc"],
"url": data["shortUrl"],
"board_id": data["idBoard"],
"list_id": data["idList"],
"labels": [
{"name": label.get("name") or "", "color": label.get("color")}
for label in data.get("labels") or []
],
"attachments": [_attachment_summary(item) for item in data.get("attachments") or []],
"comments": comments,
}
def _attachment_kind(attachment: dict) -> str:
"""card — ссылка на другую карточку Trello, image/file — загруженный файл, link — прочая ссылка."""
if _SHORT_LINK_RE.search(attachment.get("url") or ""):
return "card"
if attachment.get("isUpload"):
mime = attachment.get("mimeType") or ""
return "image" if mime.startswith("image/") else "file"
return "link"
def _attachment_summary(attachment: dict) -> dict:
return {
"id": attachment["id"],
"name": attachment.get("name") or "",
"kind": _attachment_kind(attachment),
"mime": attachment.get("mimeType") or None,
"size": attachment.get("bytes"),
"date": attachment.get("date"),
"url": attachment.get("url"),
}
def _attachments_dir() -> Path:
return Path(
os.environ.get("TRELLO_ATTACHMENTS_DIR")
or Path(tempfile.gettempdir()) / "trello-attachments"
)
def _safe_filename(name: str) -> str:
cleaned = re.sub(r"[^\w.\-]+", "_", name, flags=re.UNICODE).strip("._")
return cleaned[:100] or "attachment"
async def get_attachment(card_id: str, attachment_id: str) -> dict:
"""Метаданные одного вложения + содержимое, если это загруженный файл.
Для kind image/file файл скачивается в TRELLO_ATTACHMENTS_DIR (по
умолчанию <tmp>/trello-attachments/<card_id>/) — в ответе path и data
(bytes). Для card/link ничего не качается: как с ними быть, решает
вызывающий код.
"""
key, token = _credentials()
async with httpx.AsyncClient(follow_redirects=True) as client:
response = await client.get(
f"{TRELLO_API_BASE}/cards/{card_id}/attachments/{attachment_id}",
params={"key": key, "token": token, "fields": _ATTACHMENT_FIELDS},
)
response.raise_for_status()
attachment = _attachment_summary(response.json())
if attachment["kind"] not in ("image", "file"):
return attachment
if (attachment["size"] or 0) > MAX_ATTACHMENT_BYTES:
raise ValueError(
f"Вложение {attachment['name']!r} больше "
f"{MAX_ATTACHMENT_BYTES // (1024 * 1024)} МБ — не скачиваю"
)
# Загруженные файлы Trello отдаёт только с OAuth-заголовком,
# key/token в query для download не работают. Заголовок с токеном —
# только на домены Trello (при редиректе на чужой домен httpx его снимает).
host = httpx.URL(attachment["url"]).host
if not (host == "trello.com" or host.endswith(".trello.com")):
raise ValueError(f"Вложение ведёт не на Trello ({host}) — не скачиваю")
download = await client.get(
attachment["url"],
headers={
"Authorization": f'OAuth oauth_consumer_key="{key}", oauth_token="{token}"'
},
)
download.raise_for_status()
data = download.content
path = _attachments_dir() / card_id / f"{attachment_id}-{_safe_filename(attachment['name'])}"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return {**attachment, "size": len(data), "path": str(path), "data": data}
async def get_comments(card_id: str, limit: int = 20) -> list[dict]:
"""Комментарии карточки в хронологическом порядке (старые → новые).
card_id — полный id или shortLink, Trello API принимает оба.
"""
key, token = _credentials()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{TRELLO_API_BASE}/cards/{card_id}/actions",
params={
"key": key,
"token": token,
"filter": "commentCard",
"limit": limit,
},
)
response.raise_for_status()
actions = response.json()
return [
{
"author": (action.get("memberCreator") or {}).get("fullName") or "?",
"text": action["data"]["text"],
"date": action["date"],
}
for action in reversed(actions)
]
async def add_comment(card_id: str, text: str) -> dict:
"""Постит комментарий к карточке, возвращает id созданного комментария."""
key, token = _credentials()
async with httpx.AsyncClient() as client:
response = await client.post(
f"{TRELLO_API_BASE}/cards/{card_id}/actions/comments",
params={"key": key, "token": token, "text": text},
)
response.raise_for_status()
data = response.json()
return {"id": data["id"]}
def create_targets() -> list[str]:
"""Столбцы, в которые разрешено создавать карточки (TRELLO_CREATE_LISTS).
Через запятую, пробелы по краям не важны: "Бэклог, Баги с прода".
"""
raw = os.environ.get("TRELLO_CREATE_LISTS", "")
return [name.strip() for name in raw.split(",") if name.strip()]
def _board_id() -> str:
board_id = os.environ.get("TRELLO_BOARD_ID")
if not board_id:
raise TrelloConfigError("TRELLO_BOARD_ID не задан (проверь .env)")
return board_id
async def resolve_create_list(list_name: str) -> dict:
"""Находит на доске столбец из разрешённых, возвращает {id, name}.
Отказывает, если столбца нет в TRELLO_CREATE_LISTS — даже если на доске
он существует: писать можно только туда, куда разрешил пользователь.
"""
allowed = {name.casefold(): name for name in create_targets()}
if list_name.strip().casefold() not in allowed:
raise ValueError(
f"Столбец {list_name!r} не разрешён для создания карточек. "
f"Разрешены (TRELLO_CREATE_LISTS): {', '.join(allowed.values()) or 'ни одного'}"
)
key, token = _credentials()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{TRELLO_API_BASE}/boards/{_board_id()}/lists",
params={"key": key, "token": token, "fields": "name", "filter": "open"},
)
response.raise_for_status()
lists = response.json()
for item in lists:
if item["name"].strip().casefold() == list_name.strip().casefold():
return {"id": item["id"], "name": item["name"]}
raise ValueError(f"Столбец {list_name!r} не найден на доске TRELLO_BOARD_ID")
# Что переносится из шаблона в новую карточку: участники, метки, чек-листы и
# прочее. Описание не копируем — его пишет Claude. Обложку Trello через
# idCardSource не копирует — её ставим отдельно (см. create_card).
_TEMPLATE_KEEP = "attachments,checklists,customFields,labels,members,stickers"
async def list_templates() -> list[dict]:
"""Все карточки-шаблоны доски TRELLO_BOARD_ID, включая архивные.
[{id, name, list, desc, members: [имя], cover: {color, size, brightness} | None,
labels: [{name, color}], checklists: [{name, items}]}]
Шаблоны на доске лежат в архиве, но используются — поэтому filter=all
(без него Trello их не отдаёт). Карточек с filter=all тысячи, так что
сначала берём их с минимумом полей, а чек-листы — только у шаблонов.
"""
key, token = _credentials()
auth = {"key": key, "token": token}
board_id = _board_id()
async with httpx.AsyncClient() as client:
cards_response = await client.get(
f"{TRELLO_API_BASE}/boards/{board_id}/cards",
params={
**auth,
"fields": "name,desc,idList,labels,isTemplate,idMembers,cover",
"filter": "all",
},
)
cards_response.raise_for_status()
lists_response = await client.get(
f"{TRELLO_API_BASE}/boards/{board_id}/lists",
params={**auth, "fields": "name", "filter": "all"},
)
lists_response.raise_for_status()
members_response = await client.get(
f"{TRELLO_API_BASE}/boards/{board_id}/members",
params={**auth, "fields": "fullName"},
)
members_response.raise_for_status()
templates = [card for card in cards_response.json() if card.get("isTemplate")]
checklist_responses = await asyncio.gather(
*(
client.get(
f"{TRELLO_API_BASE}/cards/{card['id']}/checklists",
params={**auth, "fields": "name", "checkItem_fields": "name"},
)
for card in templates
)
)
for response in checklist_responses:
response.raise_for_status()
list_names = {item["id"]: item["name"] for item in lists_response.json()}
member_names = {item["id"]: item["fullName"] for item in members_response.json()}
return [
{
"id": card["id"],
"name": card["name"],
"list": list_names.get(card["idList"]),
"desc": card.get("desc") or "",
"members": [member_names.get(member_id, member_id) for member_id in card.get("idMembers") or []],
"cover": _template_cover(card.get("cover") or {}),
"labels": [
{"name": label.get("name") or "", "color": label.get("color")}
for label in card.get("labels") or []
],
"checklists": [
{
"name": checklist["name"],
"items": [item["name"] for item in checklist.get("checkItems") or []],
}
for checklist in checklists.json()
],
}
for card, checklists in zip(templates, checklist_responses)
]
def _template_cover(cover: dict) -> dict | None:
"""Цветная обложка шаблона в виде, пригодном для PUT /cards/{id}; None — без цвета."""
if not cover.get("color"):
return None
return {key: cover[key] for key in ("color", "size", "brightness") if cover.get(key)}
async def list_labels() -> list[dict]:
"""Метки доски [{name, color, uses}], самые используемые первыми; безымянные пропускаются."""
key, token = _credentials()
async with httpx.AsyncClient() as client:
response = await client.get(
f"{TRELLO_API_BASE}/boards/{_board_id()}/labels",
params={"key": key, "token": token, "fields": "name,color,uses", "limit": 1000},
)
response.raise_for_status()
labels = [
{"id": label["id"], "name": label["name"], "color": label.get("color"), "uses": label.get("uses", 0)}
for label in response.json()
if (label.get("name") or "").strip()
]
return sorted(labels, key=lambda label: -label["uses"])
async def resolve_labels(names: list[str]) -> list[dict]:
"""Метки доски по названиям (без учёта регистра) — или ошибка с неизвестными."""
by_name: dict[str, dict] = {}
for label in await list_labels():
# При дублях названий берём самую используемую — список уже отсортирован.
by_name.setdefault(label["name"].strip().casefold(), label)
found, missing = [], []
for name in names:
label = by_name.get(name.strip().casefold())
if label is None:
missing.append(name)
elif label not in found:
found.append(label)
if missing:
raise ValueError(
f"Меток нет на доске: {', '.join(map(repr, missing))}. "
"Список — trello_list_labels."
)
return found
async def resolve_template(template_name: str) -> dict:
"""Шаблон по названию (без учёта регистра) — или ошибка со списком доступных."""
templates = await list_templates()
wanted = template_name.strip().casefold()
for template in templates:
if template["name"].strip().casefold() == wanted:
return template
available = ", ".join(t["name"] for t in templates) or "ни одного"
raise ValueError(f"Шаблон {template_name!r} не найден. Есть: {available}")
async def create_card(
list_id: str,
name: str,
desc: str,
template: dict | None = None,
label_ids: list[str] | None = None,
) -> dict:
"""Создаёт карточку в столбце, возвращает {id, name, url}.
template — шаблон из list_templates: карточка создаётся его копией
(участники, метки, чек-листы…) и получает его обложку. label_ids —
метки, которые добавляются к меткам шаблона.
"""
key, token = _credentials()
auth = {"key": key, "token": token}
payload = {"idList": list_id, "name": name, "desc": desc}
if template:
payload |= {"idCardSource": template["id"], "keepFromSource": _TEMPLATE_KEEP}
async with httpx.AsyncClient() as client:
response = await client.post(f"{TRELLO_API_BASE}/cards", params=auth, json=payload)
response.raise_for_status()
data = response.json()
# Копия шаблона может унаследовать его описание, флаг isTemplate или
# архивность (шаблоны на доске архивные) — добиваем до того, что
# подтвердил пользователь.
fixes = {}
if data.get("closed"):
fixes["closed"] = False
if data.get("desc", desc) != desc:
fixes["desc"] = desc
if data.get("isTemplate"):
fixes["isTemplate"] = False
if template and template.get("cover"):
fixes["cover"] = template["cover"]
extra_labels = [label_id for label_id in label_ids or [] if label_id not in data.get("idLabels", [])]
if extra_labels:
fixes["idLabels"] = ",".join([*data.get("idLabels", []), *extra_labels])
if fixes:
fix_response = await client.put(
f"{TRELLO_API_BASE}/cards/{data['id']}", params=auth, json=fixes
)
fix_response.raise_for_status()
return {"id": data["id"], "name": data["name"], "url": data["shortUrl"]}