ai-video-fullstack/backend/services/tool_policy.py

"""Runtime behavior shared by HTTP, MCP, and client tools.

The stored definitions are intentionally plain dictionaries.  Keeping policy
normalization here gives old definitions stable defaults without spreading
fallback rules across PromptBrain, WorkflowBrain, and the pipeline.
"""

from __future__ import annotations

from dataclasses import dataclass

from models import RuntimeTool


@dataclass(frozen=True)
class ToolRuntimePolicy:
    """Normalized interaction policy for one reusable tool."""

    allow_interruptions: bool
    execution_mode: str
    wait_for_response: bool
    response_wait_mode: str

    @property
    def cancel_on_interruption(self) -> bool:
        """Pipecat uses this flag to distinguish waiting from async tools."""
        return self.execution_mode == "immediate"

    @property
    def runs_llm_after_result(self) -> bool:
        """Late async results update context without starting another reply."""
        return self.execution_mode == "immediate"


def policy_for_tool(tool: RuntimeTool) -> ToolRuntimePolicy:
    """Read policy with backwards-compatible defaults per tool kind."""
    config = (tool.definition or {}).get("config") or {}
    default_mode = "async" if tool.type == "client" else "immediate"
    mode = str(config.get("execution_mode") or default_mode)
    if mode not in {"immediate", "async"}:
        mode = default_mode
    response_wait_mode = str(config.get("response_wait_mode") or "timeout")
    if response_wait_mode not in {"timeout", "session"}:
        response_wait_mode = "timeout"
    return ToolRuntimePolicy(
        allow_interruptions=bool(config.get("allow_interruptions", True)),
        execution_mode=mode,
        wait_for_response=bool(config.get("wait_for_response", True)),
        response_wait_mode=response_wait_mode,
    )