from typing import Any from app.projects.service import ProjectService from app.tools._dispatch import NOT_HANDLED, ToolContext TOOL_NAMES = frozenset({ "sync_taiga_projects", "list_taiga_projects", "list_taiga_tasks", "create_work_item", "list_work_items", }) TOOL_DEFINITIONS: list[dict[str, Any]] = [ { "type": "function", "function": { "name": "sync_taiga_projects", "description": "Синхронизировать список проектов из Taiga API. Вызывай если проекты неизвестны.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_projects", "description": "Список проектов Taiga с привязкой Gitea.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "list_taiga_tasks", "description": ( "ОБЯЗАТЕЛЬНО при вопросах «какие задачи», «покажи задачи проекта», «что открыто в Taiga». " "Живые user stories и tasks из Taiga API. НЕ путать с list_work_items." ), "parameters": { "type": "object", "properties": { "project_slug": { "type": "string", "description": "Slug проекта, например home_assistant. Пусто = все проекты.", }, "limit": {"type": "integer", "description": "Макс. на проект, по умолчанию 20"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "create_work_item", "description": ( "Создать фичу/баг из вольного текста: структурировать через LLM, " "создать Taiga story + Gitea issue. Вызывай при «заведи баг», «оформи фичу», «добавь в таигу»." ), "parameters": { "type": "object", "properties": { "text": {"type": "string", "description": "Полное описание от пользователя"}, "project_slug": { "type": "string", "description": "Slug проекта Taiga, если известен", }, }, "required": ["text"], }, }, }, { "type": "function", "function": { "name": "list_work_items", "description": ( "Только задачи, созданные ЭТИМ ассистентом через create_work_item (локальная БД). " "НЕ использовать для общего вопроса «какие задачи в Taiga» — для того list_taiga_tasks." ), "parameters": { "type": "object", "properties": { "status": {"type": "string", "description": "open или closed"}, "limit": {"type": "integer"}, }, "required": [], }, }, }, ] async def execute(name: str, arguments: dict[str, Any], ctx: ToolContext) -> Any: if name not in TOOL_NAMES: return NOT_HANDLED projects = ProjectService(ctx.db, ctx.user_id) if name == "sync_taiga_projects": from app.projects.context import invalidate_projects_snapshot_cache result = projects.sync_taiga_projects() invalidate_projects_snapshot_cache(ctx.user_id) return result if name == "list_taiga_projects": return projects.list_projects() if name == "list_taiga_tasks": return projects.list_taiga_open_tasks( project_slug=arguments.get("project_slug"), limit=arguments.get("limit", 20), ) if name == "create_work_item": return await projects.create_work_item( arguments.get("text", ""), project_slug=arguments.get("project_slug"), ) if name == "list_work_items": return projects.list_work_items( limit=arguments.get("limit", 20), status=arguments.get("status"), ) return NOT_HANDLED