Feat/dybamic transfer (#521)

* feat: enable dynamic transfer destination resolution

* fix: remove approved routes, policy and fallback

* fix: review comments

api/services/workflow/pipecat_engine_custom_tools.py

@@ -31,6 +31,10 @@ from api.services.workflow.tools.custom_tool import (
     execute_http_tool,
     tool_to_function_schema,
 )
+from api.services.workflow.tools.transfer_resolver import (
+    TransferResolutionError,
+    resolve_transfer_config,
+)
 from api.utils.template_renderer import render_template
 
 if TYPE_CHECKING:
@@ -313,7 +317,7 @@ class CustomToolManager:
         if tool.category == ToolCategory.END_CALL.value:
             handler = self._create_end_call_handler(tool, function_name)
         elif tool.category == ToolCategory.TRANSFER_CALL.value:
-            timeout_secs = 120.0
+            timeout_secs = self._transfer_handler_timeout_secs(tool)
             handler = self._create_transfer_call_handler(tool, function_name)
         else:
             timeout_ms = ((tool.definition or {}).get("config", {}) or {}).get(
@@ -324,6 +328,28 @@ class CustomToolManager:
 
         return handler, timeout_secs
 
+    def _transfer_handler_timeout_secs(self, tool: Any) -> float:
+        config = ((tool.definition or {}).get("config", {}) or {})
+        try:
+            transfer_timeout = int(config.get("timeout", 30))
+        except (TypeError, ValueError):
+            transfer_timeout = 30
+        transfer_timeout = min(max(transfer_timeout, 5), 120)
+
+        resolver_timeout = 0.0
+        resolver = config.get("resolver")
+        if (
+            config.get("destination_source", "static") == "dynamic"
+            and isinstance(resolver, dict)
+        ):
+            try:
+                resolver_timeout = float(resolver.get("timeout_ms", 3000)) / 1000.0
+            except (TypeError, ValueError):
+                resolver_timeout = 3.0
+            resolver_timeout = min(max(resolver_timeout, 0.5), 5.0)
+
+        return float(transfer_timeout) + resolver_timeout + 15.0
+
     def _register_calculator_handler(self) -> None:
         """Register the built-in calculator function with the LLM."""
 
@@ -520,15 +546,16 @@ class CustomToolManager:
             function_call_params: FunctionCallParams,
         ) -> None:
             logger.info(f"Transfer Call Tool EXECUTED: {function_name}")
-            logger.info(f"Arguments: {function_call_params.arguments}")
+            logger.info(
+                "Transfer call arguments received "
+                f"argument_keys={list((function_call_params.arguments or {}).keys())}"
+            )
 
             try:
                 # Get the transfer call configuration
                 config = tool.definition.get("config", {})
                 destination = config.get("destination", "")
-                timeout_seconds = config.get(
-                    "timeout", 30
-                )  # Default 30 seconds if not configured
+                timeout_seconds = config.get("timeout", 30)
 
                 # Check if this is a WebRTC call - transfers are not supported
                 workflow_run = await db_client.get_workflow_run_by_id(
@@ -560,30 +587,7 @@ class CustomToolManager:
                     )
                     return
 
-                destination = _render_transfer_destination(
-                    destination,
-                    self._engine._call_context_vars,
-                    self._engine._gathered_context,
-                )
-
-                # Validate destination phone number
-                if not destination or not destination.strip():
-                    validation_error_result = {
-                        "status": "failed",
-                        "message": "I'm sorry, but I don't have a phone number configured for the transfer. Please contact support to set up call transfer.",
-                        "action": "transfer_failed",
-                        "reason": "no_destination",
-                    }
-                    await self._handle_transfer_result(
-                        validation_error_result, function_call_params, properties
-                    )
-                    return
-
-                played = await self._play_config_message(config)
-                if played:
-                    self._engine._queued_speech_mute_state = "waiting"
-
-                # Get organization ID for provider configuration
+                # Get organization ID for resolver/provider configuration
                 organization_id = await self.get_organization_id()
                 if not organization_id:
                     validation_error_result = {
@@ -597,6 +601,87 @@ class CustomToolManager:
                     )
                     return
 
+                resolver = config.get("resolver") if isinstance(config, dict) else None
+                is_dynamic_transfer = (
+                    config.get("destination_source", "static") == "dynamic"
+                    and isinstance(resolver, dict)
+                )
+                resolver_phase_muted = False
+
+                def clear_transfer_setup_mute_state() -> None:
+                    nonlocal resolver_phase_muted
+                    if resolver_phase_muted:
+                        self._engine.set_mute_pipeline(False)
+                        resolver_phase_muted = False
+                    self._engine._queued_speech_mute_state = "idle"
+
+                if is_dynamic_transfer:
+                    self._engine.set_mute_pipeline(True)
+                    resolver_phase_muted = True
+
+                if is_dynamic_transfer and resolver.get("wait_message"):
+                    await self._engine.task.queue_frame(
+                        TTSSpeakFrame(
+                            str(resolver["wait_message"]),
+                            append_to_context=False,
+                            persist_to_logs=True,
+                        )
+                    )
+                    self._engine._queued_speech_mute_state = "waiting"
+
+                try:
+                    resolved_transfer = await resolve_transfer_config(
+                        tool=tool,
+                        config=config,
+                        arguments=function_call_params.arguments or {},
+                        call_context_vars=self._engine._call_context_vars,
+                        gathered_context_vars=self._engine._gathered_context,
+                        organization_id=organization_id,
+                        workflow_run_id=self._engine._workflow_run_id,
+                    )
+                    destination = resolved_transfer.destination
+                    timeout_seconds = resolved_transfer.timeout_seconds
+                except TransferResolutionError as e:
+                    clear_transfer_setup_mute_state()
+                    validation_error_result = {
+                        "status": "failed",
+                        "message": "I'm sorry, but I couldn't find a valid destination for this transfer.",
+                        "action": "transfer_failed",
+                        "reason": e.reason,
+                    }
+                    await self._handle_transfer_result(
+                        validation_error_result, function_call_params, properties
+                    )
+                    return
+
+                # Validate destination phone number
+                if not destination or not destination.strip():
+                    validation_error_result = {
+                        "status": "failed",
+                        "message": "I'm sorry, but I don't have a phone number configured for the transfer. Please contact support to set up call transfer.",
+                        "action": "transfer_failed",
+                        "reason": "no_destination",
+                    }
+                    clear_transfer_setup_mute_state()
+                    await self._handle_transfer_result(
+                        validation_error_result, function_call_params, properties
+                    )
+                    return
+
+                if resolved_transfer.message:
+                    await self._engine.task.queue_frame(
+                        TTSSpeakFrame(
+                            resolved_transfer.message,
+                            append_to_context=False,
+                            persist_to_logs=True,
+                        )
+                    )
+                    self._engine._queued_speech_mute_state = "waiting"
+                else:
+                    played = await self._play_config_message(config)
+                    if played:
+                        self._engine._queued_speech_mute_state = "waiting"
+
                 provider = await get_telephony_provider_for_run(
                     workflow_run, organization_id
                 )
@@ -607,6 +692,7 @@ class CustomToolManager:
                         "action": "transfer_failed",
                         "reason": "provider_does_not_support_transfer",
                     }
+                    clear_transfer_setup_mute_state()
                     await self._handle_transfer_result(
                         validation_error_result, function_call_params, properties
                     )
@@ -639,6 +725,15 @@ class CustomToolManager:
 
                 # Initiate transfer via provider with inline TwiML
                 try:
+                    masked_destination = (
+                        f"***{destination[-4:]}" if len(destination) > 4 else "***"
+                    )
+                    logger.info(
+                        "Transfer provider call starting "
+                        f"source={resolved_transfer.source} "
+                        f"resolution_id={resolved_transfer.resolution_id or ''} "
+                        f"destination={masked_destination} timeout={timeout_seconds}"
+                    )
                     transfer_result = await provider.transfer_call(
                         destination=destination,
                         transfer_id=transfer_id,
@@ -648,6 +743,7 @@ class CustomToolManager:
                 except Exception as e:
                     logger.error(f"Transfer provider failed: {e}")
                     self._engine.set_mute_pipeline(False)
+                    self._engine._queued_speech_mute_state = "idle"
                     await call_transfer_manager.remove_transfer_context(transfer_id)
                     provider_error_result = {
                         "status": "failed",
@@ -669,7 +765,9 @@ class CustomToolManager:
 
                 # Wait for status callback completion using Redis pub/sub
                 logger.info(
-                    f"Transfer call initiated for {destination} (transfer_id={transfer_id}), waiting for completion..."
+                    "Transfer call initiated "
+                    f"destination={masked_destination} transfer_id={transfer_id}, "
+                    "waiting for completion..."
                 )
 
                 # Start hold music during transfer waiting period
@@ -745,6 +843,7 @@ class CustomToolManager:
                     f"Transfer call tool '{function_name}' execution failed: {e}"
                 )
                 self._engine.set_mute_pipeline(False)
+                self._engine._queued_speech_mute_state = "idle"
 
                 # Handle generic exception with user-friendly message
                 exception_result = {

api/services/workflow/tools/custom_tool.py

@@ -33,6 +33,20 @@ def tool_to_function_schema(tool: Any) -> Dict[str, Any]:
     definition = tool.definition or {}
     config = definition.get("config", {})
     parameters = config.get("parameters", []) or []
+    if (
+        definition.get("type") == "transfer_call"
+        and config.get("destination_source", "static") != "dynamic"
+    ):
+        parameters = []
+    elif (
+        definition.get("type") == "transfer_call"
+        and config.get("destination_source", "static") == "dynamic"
+    ):
+        resolver = config.get("resolver")
+        if isinstance(resolver, dict):
+            parameters = resolver.get("parameters", []) or []
+        else:
+            parameters = []
 
     # Build properties and required list from parameters
     properties = {}

api/services/workflow/tools/transfer_resolver.py

@@ -0,0 +1,326 @@
+"""Resolve transfer-call destinations from static config or HTTP resolvers."""
+
+from __future__ import annotations
+
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+import httpx
+from loguru import logger
+
+from api.db import db_client
+from api.services.workflow.tools.custom_tool import _resolve_preset_parameters
+from api.utils.credential_auth import build_auth_header
+from api.utils.template_renderer import render_template
+from api.utils.url_security import validate_user_configured_service_url
+
+
+@dataclass
+class ResolvedTransferConfig:
+    destination: str
+    timeout_seconds: int
+    message: Optional[str] = None
+    source: str = "static"
+    resolution_id: Optional[str] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+class TransferResolutionError(ValueError):
+    """Raised when a transfer destination cannot be resolved safely."""
+
+    def __init__(self, reason: str, message: str):
+        super().__init__(message)
+        self.reason = reason
+        self.message = message
+
+
+def _render_value(
+    value: Any,
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+) -> str:
+    initial_context = dict(call_context_vars or {})
+    render_context: Dict[str, Any] = {
+        **initial_context,
+        "initial_context": initial_context,
+        "gathered_context": dict(gathered_context_vars or {}),
+    }
+    rendered = render_template(value, render_context)
+    if rendered is None:
+        return ""
+    return str(rendered).strip()
+
+
+def _base_timeout(config: dict[str, Any]) -> int:
+    timeout = config.get("timeout", 30)
+    try:
+        timeout_int = int(timeout)
+    except (TypeError, ValueError):
+        timeout_int = 30
+    return min(max(timeout_int, 5), 120)
+
+
+def _mask_destination(destination: Any) -> str:
+    value = "" if destination is None else str(destination).strip()
+    if not value:
+        return ""
+    if len(value) <= 4:
+        return "***"
+    return f"***{value[-4:]}"
+
+
+_SENSITIVE_LOG_KEY_PARTS = (
+    "authorization",
+    "auth",
+    "card",
+    "destination",
+    "email",
+    "password",
+    "phone",
+    "secret",
+    "ssn",
+    "token",
+)
+
+
+def _safe_log_value(key: str, value: Any) -> Any:
+    key_lower = key.lower()
+    if any(part in key_lower for part in _SENSITIVE_LOG_KEY_PARTS):
+        return "<redacted>"
+    if value is None or isinstance(value, (bool, int, float)):
+        return value
+    if isinstance(value, str):
+        stripped = value.strip()
+        if len(stripped) > 80:
+            return f"{stripped[:77]}..."
+        return stripped
+    if isinstance(value, list):
+        return f"<array:{len(value)}>"
+    if isinstance(value, dict):
+        return f"<object:{len(value)}>"
+    return f"<{type(value).__name__}>"
+
+
+def _safe_log_dict(data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    return {
+        str(key): _safe_log_value(str(key), value)
+        for key, value in (data or {}).items()
+    }
+
+
+def _resolve_static_transfer(
+    config: dict[str, Any],
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+) -> ResolvedTransferConfig:
+    return ResolvedTransferConfig(
+        destination=_render_value(
+            config.get("destination", ""), call_context_vars, gathered_context_vars
+        ),
+        timeout_seconds=_base_timeout(config),
+    )
+
+
+def _resolver_arguments(
+    *,
+    resolver: dict[str, Any],
+    arguments: dict[str, Any],
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+) -> dict[str, Any]:
+    try:
+        preset_arguments = _resolve_preset_parameters(
+            resolver, call_context_vars, gathered_context_vars
+        )
+    except ValueError as exc:
+        raise TransferResolutionError("preset_parameter_error", str(exc)) from exc
+    return {**(arguments or {}), **preset_arguments}
+
+
+async def _execute_http_resolver(
+    *,
+    resolver: dict[str, Any],
+    resolved_arguments: dict[str, Any],
+    organization_id: Optional[int],
+    resolution_id: str,
+) -> dict[str, Any]:
+    url = resolver.get("url", "")
+    validate_user_configured_service_url(url, field_name="config.resolver.url")
+
+    method = "POST"
+    headers = dict(resolver.get("headers", {}) or {})
+    if method in ("POST", "PUT", "PATCH"):
+        headers.setdefault("Content-Type", "application/json")
+
+    credential_uuid = resolver.get("credential_uuid")
+    if credential_uuid and organization_id:
+        credential = await db_client.get_credential_by_uuid(
+            credential_uuid, organization_id
+        )
+        if credential:
+            headers.update(build_auth_header(credential))
+        else:
+            raise TransferResolutionError(
+                "credential_not_found",
+                "Transfer resolver credential was not found for this organization",
+            )
+
+    body = resolved_arguments
+
+    timeout_seconds = float(resolver.get("timeout_ms", 3000)) / 1000.0
+    logger.debug(
+        "Transfer resolver request prepared "
+        f"resolution_id={resolution_id} method={method} "
+        f"argument_keys={list(resolved_arguments.keys())} "
+        f"arguments={_safe_log_dict(resolved_arguments)}"
+    )
+
+    try:
+        started_at = time.monotonic()
+        async with httpx.AsyncClient(timeout=timeout_seconds) as client:
+            response = await client.request(
+                method=method,
+                url=url,
+                headers=headers,
+                json=body,
+            )
+        duration_ms = int((time.monotonic() - started_at) * 1000)
+    except httpx.TimeoutException as exc:
+        raise TransferResolutionError(
+            "resolver_timeout",
+            f"Transfer resolver timed out after {timeout_seconds:.1f} seconds",
+        ) from exc
+    except httpx.RequestError as exc:
+        raise TransferResolutionError(
+            "resolver_request_failed", f"Transfer resolver request failed: {exc}"
+        ) from exc
+
+    if response.status_code < 200 or response.status_code >= 300:
+        logger.warning(
+            "Transfer resolver HTTP error "
+            f"resolution_id={resolution_id} status_code={response.status_code} "
+            f"duration_ms={duration_ms}"
+        )
+        raise TransferResolutionError(
+            "resolver_http_error",
+            f"Transfer resolver returned HTTP {response.status_code}",
+        )
+
+    try:
+        data = response.json()
+    except Exception as exc:
+        raise TransferResolutionError(
+            "invalid_resolver_response", "Transfer resolver returned non-JSON response"
+        ) from exc
+
+    if not isinstance(data, dict):
+        raise TransferResolutionError(
+            "invalid_resolver_response",
+            "Transfer resolver response must be a JSON object",
+        )
+    logger.info(
+        "Transfer resolver HTTP completed "
+        f"resolution_id={resolution_id} status_code={response.status_code} "
+        f"duration_ms={duration_ms} response_keys={list(data.keys())}"
+    )
+    return data
+
+
+def _resolve_from_response(
+    *,
+    response_data: dict[str, Any],
+    config: dict[str, Any],
+    resolution_id: str,
+) -> ResolvedTransferConfig:
+    transfer_context = response_data.get("transfer_context")
+    if not isinstance(transfer_context, dict):
+        raise TransferResolutionError(
+            "invalid_resolver_response",
+            "Transfer resolver response must contain transfer_context object",
+        )
+
+    destination = transfer_context.get("destination")
+    if not isinstance(destination, str) or not destination.strip():
+        raise TransferResolutionError(
+            "no_destination",
+            "Transfer resolver response must contain transfer_context.destination",
+        )
+
+    custom_message = transfer_context.get("custom_message")
+    if custom_message is not None and not isinstance(custom_message, str):
+        raise TransferResolutionError(
+            "invalid_custom_message",
+            "transfer_context.custom_message must be a string when provided",
+        )
+
+    return ResolvedTransferConfig(
+        destination=destination.strip(),
+        timeout_seconds=_base_timeout(config),
+        message=custom_message.strip() if custom_message else None,
+        source="http_resolver",
+        resolution_id=resolution_id,
+    )
+
+
+async def resolve_transfer_config(
+    *,
+    tool: Any,
+    config: dict[str, Any],
+    arguments: dict[str, Any],
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+    organization_id: Optional[int],
+    workflow_run_id: Optional[int],
+) -> ResolvedTransferConfig:
+    """Resolve transfer destination and options for a transfer tool call."""
+
+    resolver = config.get("resolver")
+    if config.get("destination_source", "static") != "dynamic" or not isinstance(
+        resolver, dict
+    ):
+        resolved = _resolve_static_transfer(
+            config, call_context_vars, gathered_context_vars
+        )
+        logger.info(
+            "Transfer destination resolved "
+            f"source={resolved.source} destination={_mask_destination(resolved.destination)} "
+            f"timeout={resolved.timeout_seconds}"
+        )
+        return resolved
+
+    resolution_id = str(uuid.uuid4())
+    logger.info(
+        "Transfer resolver started "
+        f"resolution_id={resolution_id} tool_uuid={getattr(tool, 'tool_uuid', None)} "
+        f"workflow_run_id={workflow_run_id} type={resolver.get('type')} "
+        "method=POST "
+        f"timeout_ms={resolver.get('timeout_ms', 3000)}"
+    )
+
+    resolved_arguments = _resolver_arguments(
+        resolver=resolver,
+        arguments=arguments,
+        call_context_vars=call_context_vars,
+        gathered_context_vars=gathered_context_vars,
+    )
+    response_data = await _execute_http_resolver(
+        resolver=resolver,
+        resolved_arguments=resolved_arguments,
+        organization_id=organization_id,
+        resolution_id=resolution_id,
+    )
+    resolved = _resolve_from_response(
+        response_data=response_data,
+        config=config,
+        resolution_id=resolution_id,
+    )
+    logger.info(
+        "Transfer destination resolved "
+        f"resolution_id={resolution_id} source={resolved.source} "
+        f"destination={_mask_destination(resolved.destination)} "
+        f"timeout={resolved.timeout_seconds} "
+        f"custom_message_present={bool(resolved.message)}"
+    )
+    return resolved