leocrm/app/ai/integration_tools.py

"""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")