diff --git a/app/ai/integration_tools.py b/app/ai/integration_tools.py new file mode 100644 index 0000000..4849f85 --- /dev/null +++ b/app/ai/integration_tools.py @@ -0,0 +1,170 @@ +"""Integration tools — Agent → Workflow + Agent → Knowledge. + +Registers AI tools that allow agents to start workflows and ask knowledge questions. +Builds on: workflow_service, knowledge services, tool_registry. +""" +from __future__ import annotations +import logging +import uuid +from typing import Any + +logger = logging.getLogger(__name__) + + +def register_integration_tools() -> None: + """Register workflow + knowledge tools in the AI tool registry.""" + from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry + registry = get_tool_registry() + + # ── I-AW: Agent → Workflow ── + async def _start_workflow_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: + """Start a workflow by ID.""" + from app.services.workflow_service import create_instance + from app.core.db import get_worker_session_factory + workflow_id = arguments.get("workflow_id", "") + tenant_id = context.get("tenant_id") + user_id = context.get("user_id") + if not workflow_id or not tenant_id: + return {"error": "workflow_id and tenant_id required"} + factory = get_worker_session_factory() + async with factory() as db: + instance = await create_instance( + db=db, + tenant_id=uuid.UUID(str(tenant_id)), + workflow_id=uuid.UUID(workflow_id), + initiated_by=uuid.UUID(str(user_id)) if user_id else None, + ) + await db.commit() + return {"instance_id": str(instance.get("id", "")), "status": instance.get("status", "created")} + + registry.register( + name="start_workflow", + description="Start a workflow by its ID. Returns the instance ID and status.", + parameters={ + "type": "object", + "properties": { + "workflow_id": {"type": "string", "description": "UUID of the workflow to start"}, + }, + "required": ["workflow_id"], + }, + handler=_start_workflow_handler, + plugin_name="integration", + required_permission="workflows:read", + category="workflow", + ) + + async def _check_workflow_status_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: + """Check the status of a workflow instance.""" + from sqlalchemy import select + from app.models.workflow import WorkflowInstance + from app.core.db import get_worker_session_factory + instance_id = arguments.get("instance_id", "") + tenant_id = context.get("tenant_id") + if not instance_id or not tenant_id: + return {"error": "instance_id and tenant_id required"} + factory = get_worker_session_factory() + async with factory() as db: + result = await db.execute( + select(WorkflowInstance).where( + WorkflowInstance.id == uuid.UUID(instance_id), + WorkflowInstance.tenant_id == uuid.UUID(str(tenant_id)), + ) + ) + inst = result.scalar_one_or_none() + if not inst: + return {"error": "Instance not found"} + return { + "instance_id": str(inst.id), + "status": inst.status, + "current_step": inst.current_step_index, + "completed_at": inst.completed_at.isoformat() if inst.completed_at else None, + } + + registry.register( + name="check_workflow_status", + description="Check the status of a workflow instance by its ID.", + parameters={ + "type": "object", + "properties": { + "instance_id": {"type": "string", "description": "UUID of the workflow instance"}, + }, + "required": ["instance_id"], + }, + handler=_check_workflow_status_handler, + plugin_name="integration", + required_permission="workflows:read", + category="workflow", + ) + + # ── I-AK: Agent → Knowledge ── + async def _ask_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: + """Ask a knowledge question.""" + from app.plugins.builtins.knowledge.services import ask_knowledge + from app.core.db import get_worker_session_factory + question = arguments.get("question", "") + tenant_id = context.get("tenant_id") + if not question or not tenant_id: + return {"error": "question and tenant_id required"} + factory = get_worker_session_factory() + async with factory() as db: + result = await ask_knowledge(db=db, tenant_id=uuid.UUID(str(tenant_id)), question=question) + return {"answer": result.get("answer", ""), "evidence_count": len(result.get("evidence", []))} + + registry.register( + name="ask_knowledge", + description="Ask a knowledge question. Searches wiki articles and graph relationships for context.", + parameters={ + "type": "object", + "properties": { + "question": {"type": "string", "description": "The question to ask"}, + }, + "required": ["question"], + }, + handler=_ask_knowledge_handler, + plugin_name="integration", + required_permission="wiki:read", + category="knowledge", + ) + + async def _search_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: + """Search wiki articles via unified search.""" + from app.plugins.builtins.unified_search.provider_registry import get_search_registry + from app.core.db import get_worker_session_factory + query = arguments.get("query", "") + tenant_id = context.get("tenant_id") + if not query or not tenant_id: + return {"error": "query and tenant_id required"} + factory = get_worker_session_factory() + async with factory() as db: + registry = get_search_registry() + wiki_provider = registry.get("wiki_article") + if not wiki_provider: + return {"error": "Wiki search provider not available"} + results = await wiki_provider._search_fts_filtered( + db=db, tsquery=query, tenant_id=uuid.UUID(str(tenant_id)), limit=5, visible_ids=None + ) + return { + "results": [ + {"title": r.get("title", ""), "summary": (r.get("summary") or r.get("content", "")[:200] or "")} + for r in results + ], + "count": len(results), + } + + registry.register( + name="search_knowledge", + description="Search wiki articles by keyword. Returns matching articles with title and summary.", + parameters={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Search query"}, + }, + "required": ["query"], + }, + handler=_search_knowledge_handler, + plugin_name="integration", + required_permission="wiki:read", + category="knowledge", + ) + + logger.info("Registered integration tools: start_workflow, check_workflow_status, ask_knowledge, search_knowledge") diff --git a/app/plugins/builtins/automation/plugin.py b/app/plugins/builtins/automation/plugin.py index 911fff4..e91c4bd 100644 --- a/app/plugins/builtins/automation/plugin.py +++ b/app/plugins/builtins/automation/plugin.py @@ -210,6 +210,12 @@ class AutomationPlugin(BasePlugin): register_agent_coordinator_tools() except Exception: logger.exception("Failed to register agent coordinator tools") + # Register integration tools (I-AW: Agent→Workflow, I-AK: Agent→Knowledge) + try: + from app.ai.integration_tools import register_integration_tools + register_integration_tools() + except Exception: + logger.exception("Failed to register integration tools") # Register MiniApps from manifest try: from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry