"""Integration tools — Agent → Workflow and Agent → Knowledge (I-AW, I-AK). Provides AI agent tools for: - Starting and checking workflow status (I-AW) - Querying knowledge base with evidence (I-AK) These tools are registered in the AI tool registry and can be used by agents via the ReAct loop. Each tool respects tenant_id and permissions. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # ─── I-AW: Agent → Workflow Tools ──────────────────────────────────────────── async def start_workflow_tool( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, workflow_id: str, context: dict[str, Any] | None = None, ) -> dict[str, Any]: """Agent tool: Start a workflow instance. Args: workflow_id: The workflow definition ID. context: Optional initial context variables. Returns: Dict with instance_id, status, and workflow info. """ from app.services.workflow_service import create_instance try: result = await create_instance( db, tenant_id, user_id, workflow_id=workflow_id, context=context or {}, ) if result is None: return {"error": "Workflow not found", "status": "not_found"} return { "instance_id": result.get("id"), "status": result.get("status"), "workflow_id": workflow_id, "message": f"Workflow started successfully", } except Exception as e: logger.warning("start_workflow_tool failed: %s", e) return {"error": str(e), "status": "failed"} async def check_workflow_status_tool( db: AsyncSession, tenant_id: uuid.UUID, instance_id: str, ) -> dict[str, Any]: """Agent tool: Check the status of a workflow instance. Args: instance_id: The workflow instance ID. Returns: Dict with status, current_step, and step history. """ from app.services.workflow_service import get_instance from app.models.workflow import WorkflowStepHistory from sqlalchemy import select try: instance = await get_instance(db, tenant_id, instance_id) if instance is None: return {"error": "Instance not found", "status": "not_found"} # Get step history history_result = await db.execute( select(WorkflowStepHistory) .where( WorkflowStepHistory.tenant_id == tenant_id, WorkflowStepHistory.instance_id == uuid.UUID(instance_id), ) .order_by(WorkflowStepHistory.created_at.desc()) .limit(5) ) recent_steps = [ { "step_index": h.step_index, "step_type": h.step_type, "action": h.action, } for h in history_result.scalars().all() ] return { "instance_id": instance_id, "status": instance.get("status"), "current_step_index": instance.get("current_step_index"), "recent_steps": recent_steps, } except Exception as e: logger.warning("check_workflow_status_tool failed: %s", e) return {"error": str(e), "status": "failed"} # ─── I-AK: Agent → Knowledge Tools ─────────────────────────────────────────── async def ask_knowledge_tool( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, query: str, source_types: list[str] | None = None, max_results: int = 5, ) -> dict[str, Any]: """Agent tool: Query the knowledge base with evidence-backed results. Args: query: Natural language query. source_types: Optional filter (wiki, dms, mail, communication). max_results: Maximum results to return. Returns: Dict with answer, evidence references, and workstream blocks. """ from app.ai.knowledge_lifecycle import ask_knowledge return await ask_knowledge( db=db, tenant_id=tenant_id, user_id=user_id, query=query, source_types=source_types, max_results=max_results, ) async def search_knowledge_tool( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, query: str, entity_type: str | None = None, limit: int = 10, ) -> dict[str, Any]: """Agent tool: Search across all knowledge sources. Args: query: Search query. entity_type: Optional entity type filter. limit: Maximum results. Returns: Dict with search results and evidence references. """ try: from app.plugins.builtins.unified_search.contracts import UnifiedSearchContract contract = UnifiedSearchContract search_fn = contract.get_function("unified_search") if search_fn is None: return {"error": "Search not available", "results": []} results = await search_fn( db=db, tenant_id=tenant_id, query=query, entity_type=entity_type, limit=limit, ) # Build evidence references from app.ai.knowledge_sources import build_evidence_references refs = build_evidence_references(results or [], max_results=limit) return { "results": [r.to_dict() for r in refs], "total": len(refs), "query": query, } except Exception as e: logger.warning("search_knowledge_tool failed: %s", e) return {"error": str(e), "results": []} # ─── Tool Registration ─────────────────────────────────────────────────────── def register_integration_tools(registry: Any) -> None: """Register integration tools in the AI tool registry. Called during plugin initialization to make workflow and knowledge tools available to AI agents. """ # These would be registered as ToolDefinition objects in the registry. # The actual registration depends on the ToolRegistry API. # For now, we expose the functions for manual registration. pass __all__ = [ "start_workflow_tool", "check_workflow_status_tool", "ask_knowledge_tool", "search_knowledge_tool", "register_integration_tools", ]