diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 82f23f51..4cccee4f 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -180,10 +180,58 @@ class EndCallConfig(BaseModel): ) +class HttpTransferResolverConfig(BaseModel): + """HTTP endpoint used to resolve transfer destination at call time.""" + + type: Literal["http"] = Field(default="http", description="Resolver type.") + url: str = Field(description="HTTP or HTTPS endpoint for transfer resolution.") + headers: Optional[Dict[str, str]] = Field( + default=None, + description="Static headers to include with every resolver request.", + ) + credential_uuid: Optional[str] = Field( + default=None, + description="Reference to an external credential for resolver authentication.", + ) + timeout_ms: int = Field( + default=3000, + ge=500, + le=5000, + description="Resolver request timeout in milliseconds.", + ) + wait_message: Optional[str] = Field( + default=None, + description="Optional short message played while Dograh resolves routing.", + ) + parameters: Optional[List[ToolParameter]] = Field( + default=None, + description="Parameters the model may provide when calling this transfer tool.", + ) + preset_parameters: Optional[List[PresetToolParameter]] = Field( + default=None, + description=( + "Parameters injected by Dograh from fixed values or workflow context " + "templates." + ), + ) + + @field_validator("url") + @classmethod + def validate_url(cls, v: str) -> str: + if not isinstance(v, str) or not v.startswith(("http://", "https://")): + raise ValueError("config.resolver.url must be an http(s) URL") + return v + + class TransferCallConfig(BaseModel): """Configuration for Transfer Call tools.""" + destination_source: Literal["static", "dynamic"] = Field( + default="static", + description="Whether transfer destination is static/template or resolved by HTTP.", + ) destination: str = Field( + default="", description=( "Phone number, SIP endpoint, or template to transfer the call to, e.g. " "+1234567890, PJSIP/1234, or {{initial_context.transfer_destination}}." @@ -204,6 +252,29 @@ class TransferCallConfig(BaseModel): le=120, description="Maximum seconds to wait for the destination to answer.", ) + parameters: Optional[List[ToolParameter]] = Field( + default=None, + description=( + "Parameters the model may provide when calling this transfer tool, " + "for example state, department, or transfer reason." + ), + ) + resolver: Optional[HttpTransferResolverConfig] = Field( + default=None, + description="Optional resolver that determines transfer routing at call time.", + ) + + @model_validator(mode="after") + def validate_destination_source_config(self): + if self.destination_source == "static" and not self.destination.strip(): + raise ValueError( + "config.destination is required when destination_source is static" + ) + if self.destination_source == "dynamic" and self.resolver is None: + raise ValueError( + "config.resolver is required when destination_source is dynamic" + ) + return self class McpToolConfig(BaseModel): diff --git a/api/services/tool_management.py b/api/services/tool_management.py index 12161a00..ad8bb96a 100644 --- a/api/services/tool_management.py +++ b/api/services/tool_management.py @@ -71,6 +71,23 @@ def _credential_uuid_from_definition(definition: dict[str, Any]) -> Optional[str return credential_uuid if isinstance(credential_uuid, str) else None +def _credential_uuids_from_definition(definition: dict[str, Any]) -> list[str]: + credential_uuids: list[str] = [] + top_level = _credential_uuid_from_definition(definition) + if top_level: + credential_uuids.append(top_level) + + config = definition.get("config") + if isinstance(config, dict): + resolver = config.get("resolver") + if isinstance(resolver, dict): + resolver_credential_uuid = resolver.get("credential_uuid") + if isinstance(resolver_credential_uuid, str) and resolver_credential_uuid: + credential_uuids.append(resolver_credential_uuid) + + return list(dict.fromkeys(credential_uuids)) + + async def fetch_credential(credential_uuid: Optional[str], organization_id: int): """Best-effort credential lookup for MCP auth/discovery.""" if not credential_uuid: @@ -86,22 +103,20 @@ async def validate_tool_credential_references( definition: dict[str, Any], *, organization_id: int ) -> None: """Ensure credential UUID references belong to the caller's organization.""" - credential_uuid = _credential_uuid_from_definition(definition) - if not credential_uuid: - return - - credential = await db_client.get_credential_by_uuid( - credential_uuid, organization_id - ) - if not credential: - raise ToolManagementError( - "credential_not_found", - ( - f"Credential '{credential_uuid}' was not found in this organization. " - "Create it in the UI first, then retry with its credential_uuid." - ), - status_code=404, + for credential_uuid in _credential_uuids_from_definition(definition): + credential = await db_client.get_credential_by_uuid( + credential_uuid, organization_id ) + if not credential: + raise ToolManagementError( + "credential_not_found", + ( + f"Credential '{credential_uuid}' was not found in this " + "organization. Create it in the UI first, then retry with its " + "credential_uuid." + ), + status_code=404, + ) async def populate_discovered_tools( diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index cb72ac46..d596cfc9 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/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 = { diff --git a/api/services/workflow/tools/custom_tool.py b/api/services/workflow/tools/custom_tool.py index 6e97225c..cdecc68c 100644 --- a/api/services/workflow/tools/custom_tool.py +++ b/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 = {} diff --git a/api/services/workflow/tools/transfer_resolver.py b/api/services/workflow/tools/transfer_resolver.py new file mode 100644 index 00000000..2e781fac --- /dev/null +++ b/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 "" + 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"" + if isinstance(value, dict): + return f"" + 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 diff --git a/api/tests/test_custom_tools.py b/api/tests/test_custom_tools.py index e3523edc..96062a71 100644 --- a/api/tests/test_custom_tools.py +++ b/api/tests/test_custom_tools.py @@ -49,6 +49,25 @@ class MockToolModel: definition: Dict[str, Any] +def test_empty_resolver_credential_uuid_is_ignored(): + from api.services.tool_management import _credential_uuids_from_definition + + definition = { + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "credential_uuid": "", + }, + }, + } + + assert _credential_uuids_from_definition(definition) == [] + + class TestToolToFunctionSchema: """Tests for tool_to_function_schema function.""" @@ -294,6 +313,74 @@ class TestToolToFunctionSchema: assert schema["function"]["description"] == "Execute My Tool tool" + def test_transfer_tool_schema_includes_configured_parameters(self): + """Transfer tools can expose resolver inputs to the model.""" + tool = MockToolModel( + tool_uuid="transfer-uuid", + name="Transfer To Partner", + description="Transfer to the correct referral partner", + category="transfer_call", + definition={ + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "destination": "", + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "parameters": [ + { + "name": "state", + "type": "string", + "description": "The caller's US state.", + "required": True, + }, + { + "name": "reason", + "type": "string", + "description": "Why transfer is needed.", + "required": False, + }, + ], + }, + }, + }, + ) + + schema = tool_to_function_schema(tool) + + params = schema["function"]["parameters"] + assert params["properties"]["state"]["type"] == "string" + assert params["properties"]["reason"]["type"] == "string" + assert params["required"] == ["state"] + + def test_transfer_tool_schema_handles_null_resolver_parameters(self): + """Dynamic transfer tools should remain available with no parameters.""" + tool = MockToolModel( + tool_uuid="transfer-uuid", + name="Transfer To Partner", + description="Transfer to the correct referral partner", + category="transfer_call", + definition={ + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "parameters": None, + }, + }, + }, + ) + + schema = tool_to_function_schema(tool) + + assert schema["function"]["parameters"]["properties"] == {} + assert schema["function"]["parameters"]["required"] == [] + class TestExecuteHttpTool: """Tests for execute_http_tool function.""" @@ -531,6 +618,43 @@ class TestExecuteHttpTool: assert result["status"] == "success" + @pytest.mark.asyncio + async def test_get_request_without_arguments_preserves_url_query_params(self): + """Empty runtime args should not override query params already in the URL.""" + tool = MockToolModel( + tool_uuid="test-uuid", + name="Search Users", + description="Search for users", + category="http_api", + definition={ + "schema_version": 1, + "type": "http_api", + "config": { + "method": "GET", + "url": "https://api.example.com/users/search?tenant=abc", + "timeout_ms": 5000, + }, + }, + ) + + with patch( + "api.services.workflow.tools.custom_tool.httpx.AsyncClient" + ) as mock_client_class: + mock_client = AsyncMock() + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"users": []} + mock_client.request.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + result = await execute_http_tool(tool, {}) + + call_kwargs = mock_client.request.call_args.kwargs + assert call_kwargs["method"] == "GET" + assert call_kwargs["url"].endswith("?tenant=abc") + assert call_kwargs["params"] is None + assert result["status"] == "success" + @pytest.mark.asyncio async def test_delete_request_sends_query_params(self): """Test that DELETE requests send arguments as query parameters.""" @@ -778,6 +902,164 @@ class TestCoerceParameterValue: _coerce_parameter_value(value, "array") +class TestTransferResolver: + """Tests for dynamic transfer resolution behavior.""" + + def test_dynamic_transfer_handler_timeout_includes_resolver_budget(self): + from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager + + manager = CustomToolManager(Mock()) + tool = MockToolModel( + tool_uuid="transfer-tool-uuid", + name="Transfer Call", + description="Transfer the caller", + category="transfer_call", + definition={ + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "timeout": 120, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "timeout_ms": 5000, + }, + }, + }, + ) + + _handler, timeout_secs = manager._create_handler(tool, "transfer_call") + + assert timeout_secs == 140.0 + + @pytest.mark.asyncio + async def test_http_resolver_resolves_transfer_context_destination(self): + from api.services.workflow.tools.transfer_resolver import resolve_transfer_config + + tool = MockToolModel( + tool_uuid="transfer-tool-uuid", + name="Transfer Call", + description="Transfer the caller", + category="transfer_call", + definition={}, + ) + config = { + "destination_source": "dynamic", + "destination": "", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "headers": {"X-Tenant": "ovation"}, + "timeout_ms": 3000, + "preset_parameters": [ + { + "name": "lead_id", + "type": "string", + "value_template": "{{initial_context.lead_id}}", + "required": True, + } + ], + }, + } + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "transfer_context": { + "destination": "+14155550123", + "custom_message": "I will connect you with our Texas partner now.", + } + } + + with ( + patch( + "api.services.workflow.tools.transfer_resolver.validate_user_configured_service_url" + ), + patch( + "api.services.workflow.tools.transfer_resolver.httpx.AsyncClient" + ) as mock_client_class, + ): + mock_client = AsyncMock() + mock_client.request.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + resolved = await resolve_transfer_config( + tool=tool, + config=config, + arguments={"state": "TX"}, + call_context_vars={"lead_id": "lead-123"}, + gathered_context_vars={}, + organization_id=1, + workflow_run_id=1, + ) + + assert resolved.destination == "+14155550123" + assert resolved.message == "I will connect you with our Texas partner now." + assert resolved.timeout_seconds == 30 + assert resolved.source == "http_resolver" + mock_client.request.assert_awaited_once() + request_kwargs = mock_client.request.await_args.kwargs + assert request_kwargs["method"] == "POST" + assert request_kwargs["json"] == {"state": "TX", "lead_id": "lead-123"} + assert request_kwargs["headers"]["X-Tenant"] == "ovation" + + @pytest.mark.asyncio + async def test_http_resolver_requires_transfer_context_destination(self): + from api.services.workflow.tools.transfer_resolver import ( + TransferResolutionError, + resolve_transfer_config, + ) + + tool = MockToolModel( + tool_uuid="transfer-tool-uuid", + name="Transfer Call", + description="Transfer the caller", + category="transfer_call", + definition={}, + ) + config = { + "destination_source": "dynamic", + "destination": "", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "timeout_ms": 3000, + }, + } + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"transfer_context": {}} + + with ( + patch( + "api.services.workflow.tools.transfer_resolver.validate_user_configured_service_url" + ), + patch( + "api.services.workflow.tools.transfer_resolver.httpx.AsyncClient" + ) as mock_client_class, + ): + mock_client = AsyncMock() + mock_client.request.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + with pytest.raises(TransferResolutionError) as exc_info: + await resolve_transfer_config( + tool=tool, + config=config, + arguments={}, + call_context_vars={}, + gathered_context_vars={}, + organization_id=1, + workflow_run_id=1, + ) + + assert exc_info.value.reason == "no_destination" + + class TestAuthHeaders: """Tests for auth header building utilities.""" @@ -1253,6 +1535,227 @@ class TestCustomToolManagerUnit: assert first_context.target_number == "+14155550123" assert result_received["status"] == "transfer_failed" + @pytest.mark.asyncio + async def test_transfer_call_http_resolver_uses_transfer_context_destination(self): + """HTTP resolver transfer_context.destination is passed to the provider.""" + from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager + + mock_engine = Mock() + mock_engine._workflow_run_id = 1 + mock_engine._call_context_vars = {} + mock_engine._gathered_context = {"state": "TX"} + mock_engine._fetch_recording_audio = None + mock_engine._audio_config = SimpleNamespace(transport_out_sample_rate=8000) + mock_engine._transport_output = SimpleNamespace(queue_frame=AsyncMock()) + mock_engine._get_organization_id = AsyncMock(return_value=1) + mock_engine.task = SimpleNamespace(queue_frame=AsyncMock()) + mock_engine.set_mute_pipeline = Mock() + mock_engine.end_call_with_reason = AsyncMock() + + manager = CustomToolManager(mock_engine) + tool = MockToolModel( + tool_uuid="transfer-tool-uuid", + name="Transfer Call", + description="Transfer the caller", + category="transfer_call", + definition={ + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "destination": "", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "timeout_ms": 3000, + "wait_message": "One moment while I find the right team.", + "parameters": [ + { + "name": "state", + "type": "string", + "description": "State to resolve referral partner", + "required": True, + } + ], + }, + }, + }, + ) + handler, _timeout_secs = manager._create_handler(tool, "transfer_call") + + workflow_run = SimpleNamespace( + mode=WorkflowRunMode.TWILIO.value, + gathered_context={"call_id": "caller-call-sid"}, + ) + provider = Mock() + provider.supports_transfers.return_value = True + provider.validate_config.return_value = True + provider.transfer_call = AsyncMock(return_value={"call_sid": "dest-call-sid"}) + + transfer_event = Mock() + transfer_event.to_result_dict.return_value = { + "status": "failed", + "action": "transfer_failed", + "reason": "test_complete", + } + transfer_manager = Mock() + transfer_manager.store_transfer_context = AsyncMock() + transfer_manager.wait_for_transfer_completion = AsyncMock( + return_value=transfer_event + ) + + result_received = None + + async def mock_result_callback(result, properties=None): + nonlocal result_received + result_received = result + + mock_params = Mock() + mock_params.arguments = {"state": "TX"} + mock_params.result_callback = mock_result_callback + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "transfer_context": { + "destination": "+14155550123", + "custom_message": "I will connect you with our Texas partner now.", + } + } + + with ( + patch( + "api.services.workflow.pipecat_engine_custom_tools.db_client.get_workflow_run_by_id", + new=AsyncMock(return_value=workflow_run), + ), + patch( + "api.services.workflow.pipecat_engine_custom_tools.get_telephony_provider_for_run", + new=AsyncMock(return_value=provider), + ), + patch( + "api.services.workflow.pipecat_engine_custom_tools.get_call_transfer_manager", + new=AsyncMock(return_value=transfer_manager), + ), + patch( + "api.services.workflow.pipecat_engine_custom_tools.play_audio_loop", + new=AsyncMock(return_value=None), + ), + patch( + "api.services.workflow.tools.transfer_resolver.validate_user_configured_service_url" + ), + patch( + "api.services.workflow.tools.transfer_resolver.httpx.AsyncClient" + ) as mock_client_class, + ): + mock_client = AsyncMock() + mock_client.request.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + await handler(mock_params) + + mock_client.request.assert_awaited_once() + assert mock_client.request.await_args.kwargs["json"] == {"state": "TX"} + provider.transfer_call.assert_awaited_once() + transfer_kwargs = provider.transfer_call.await_args.kwargs + assert transfer_kwargs["destination"] == "+14155550123" + assert transfer_kwargs["timeout"] == 30 + assert result_received["status"] == "transfer_failed" + assert [call.args[0] for call in mock_engine.set_mute_pipeline.call_args_list] == [ + True, + True, + False, + ] + + spoken_texts = [ + call.args[0].text for call in mock_engine.task.queue_frame.await_args_list + ] + assert "One moment while I find the right team." in spoken_texts + assert "I will connect you with our Texas partner now." in spoken_texts + + @pytest.mark.asyncio + async def test_transfer_call_resolver_failure_resets_queued_speech_mute(self): + """Resolver failure after a wait message should not leave user input muted.""" + from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager + + mock_engine = Mock() + mock_engine._workflow_run_id = 1 + mock_engine._call_context_vars = {} + mock_engine._gathered_context = {"state": "TX"} + mock_engine._fetch_recording_audio = None + mock_engine._get_organization_id = AsyncMock(return_value=1) + mock_engine.task = SimpleNamespace(queue_frame=AsyncMock()) + mock_engine.set_mute_pipeline = Mock() + mock_engine.end_call_with_reason = AsyncMock() + mock_engine._queued_speech_mute_state = "idle" + + manager = CustomToolManager(mock_engine) + tool = MockToolModel( + tool_uuid="transfer-tool-uuid", + name="Transfer Call", + description="Transfer the caller", + category="transfer_call", + definition={ + "schema_version": 1, + "type": "transfer_call", + "config": { + "destination_source": "dynamic", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "timeout_ms": 3000, + "wait_message": "One moment while I find the right team.", + }, + }, + }, + ) + handler, _timeout_secs = manager._create_handler(tool, "transfer_call") + + workflow_run = SimpleNamespace( + mode=WorkflowRunMode.TWILIO.value, + gathered_context={"call_id": "caller-call-sid"}, + ) + result_received = None + + async def mock_result_callback(result, properties=None): + nonlocal result_received + result_received = result + + mock_params = Mock() + mock_params.arguments = {"state": "TX"} + mock_params.result_callback = mock_result_callback + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"transfer_context": {}} + + with ( + patch( + "api.services.workflow.pipecat_engine_custom_tools.db_client.get_workflow_run_by_id", + new=AsyncMock(return_value=workflow_run), + ), + patch( + "api.services.workflow.tools.transfer_resolver.validate_user_configured_service_url" + ), + patch( + "api.services.workflow.tools.transfer_resolver.httpx.AsyncClient" + ) as mock_client_class, + ): + mock_client = AsyncMock() + mock_client.request.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + await handler(mock_params) + + assert result_received["status"] == "transfer_failed" + assert result_received["reason"] == "no_destination" + assert mock_engine._queued_speech_mute_state == "idle" + assert [call.args[0] for call in mock_engine.set_mute_pipeline.call_args_list] == [ + True, + False, + ] + @pytest.mark.asyncio async def test_transfer_call_propagates_provider_destination_error(self): """Provider-specific destination failures are returned through the tool result.""" diff --git a/api/tests/test_tool_schema.py b/api/tests/test_tool_schema.py index 61afc227..fd97f62d 100644 --- a/api/tests/test_tool_schema.py +++ b/api/tests/test_tool_schema.py @@ -1,3 +1,5 @@ +import pytest + from api.schemas.tool import TransferCallConfig @@ -13,3 +15,28 @@ def test_transfer_call_destination_accepts_provider_specific_literal(): config = TransferCallConfig(destination="provider-specific-destination") assert config.destination == "provider-specific-destination" + + +def test_transfer_call_static_requires_destination(): + with pytest.raises(ValueError, match="destination is required"): + TransferCallConfig(destination_source="static", destination="") + + +def test_transfer_call_dynamic_requires_resolver(): + with pytest.raises(ValueError, match="resolver is required"): + TransferCallConfig(destination_source="dynamic", destination="") + + +def test_transfer_call_dynamic_accepts_resolver_without_destination(): + config = TransferCallConfig( + destination_source="dynamic", + destination="", + resolver={ + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + }, + ) + + assert config.destination_source == "dynamic" + assert config.destination == "" + assert config.resolver is not None diff --git a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx index 1268b99d..76b87fa9 100644 --- a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx +++ b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx @@ -1,20 +1,36 @@ "use client"; import type { RecordingResponseSchema } from "@/client/types.gen"; +import { + CredentialSelector, + KeyValueEditor, + type KeyValueItem, + ParameterEditor, + PresetParameterEditor, + type PresetToolParameter, + type ToolParameter, + UrlInput, +} from "@/components/http"; import { RecordingSelect, StaticTextWarning } from "@/components/flow/TextOrAudioInput"; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; import { RadioGroup, RadioGroupItem } from "@/components/ui/radio-group"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Textarea } from "@/components/ui/textarea"; -import { type EndCallMessageType } from "../../config"; +import { + type EndCallMessageType, + type TransferDestinationSource, +} from "../../config"; export interface TransferCallToolConfigProps { name: string; onNameChange: (name: string) => void; description: string; onDescriptionChange: (description: string) => void; + destinationSource: TransferDestinationSource; + onDestinationSourceChange: (source: TransferDestinationSource) => void; destination: string; onDestinationChange: (destination: string) => void; messageType: EndCallMessageType; @@ -24,8 +40,22 @@ export interface TransferCallToolConfigProps { audioRecordingId: string; onAudioRecordingIdChange: (id: string) => void; recordings?: RecordingResponseSchema[]; - timeout?: number; // Make optional to match API type + timeout?: number; onTimeoutChange: (timeout: number) => void; + resolverUrl: string; + onResolverUrlChange: (url: string) => void; + resolverCredentialUuid: string; + onResolverCredentialUuidChange: (uuid: string) => void; + resolverHeaders: KeyValueItem[]; + onResolverHeadersChange: (headers: KeyValueItem[]) => void; + resolverTimeoutMs: number; + onResolverTimeoutMsChange: (timeoutMs: number) => void; + resolverWaitMessage: string; + onResolverWaitMessageChange: (message: string) => void; + parameters: ToolParameter[]; + onParametersChange: (parameters: ToolParameter[]) => void; + presetParameters: PresetToolParameter[]; + onPresetParametersChange: (parameters: PresetToolParameter[]) => void; } export function TransferCallToolConfig({ @@ -33,6 +63,8 @@ export function TransferCallToolConfig({ onNameChange, description, onDescriptionChange, + destinationSource, + onDestinationSourceChange, destination, onDestinationChange, messageType, @@ -44,6 +76,20 @@ export function TransferCallToolConfig({ recordings = [], timeout, onTimeoutChange, + resolverUrl, + onResolverUrlChange, + resolverCredentialUuid, + onResolverCredentialUuidChange, + resolverHeaders, + onResolverHeadersChange, + resolverTimeoutMs, + onResolverTimeoutMsChange, + resolverWaitMessage, + onResolverWaitMessageChange, + parameters, + onParametersChange, + presetParameters, + onPresetParametersChange, }: TransferCallToolConfigProps) { return ( @@ -79,31 +125,10 @@ export function TransferCallToolConfig({ /> -
- -
-

Enter one of these destination formats:

-
    -
  • SIP endpoint, e.g. PJSIP/1234
    • -
  • E.164 phone number, e.g. +1234567890
    • -
  • - Template variable, e.g. {"{{initial_context.transfer_destination}}"} -
    • -
-
- onDestinationChange(e.target.value)} - placeholder={ - "+1234567890, PJSIP/1234, or {{initial_context.transfer_destination}}" - } - /> -
-
{ const value = parseInt(e.target.value) || 30; - // Clamp value between 5 and 120 seconds - const clampedValue = Math.min(Math.max(value, 5), 120); - onTimeoutChange(clampedValue); + onTimeoutChange(Math.min(Math.max(value, 5), 120)); }} placeholder="30" min="5" @@ -186,6 +209,142 @@ export function TransferCallToolConfig({ Default: 30 seconds
+ +
+
+ +

+ Choose whether the transfer uses a configured destination or resolves one from an HTTP endpoint. +

+
+ onDestinationSourceChange(v as TransferDestinationSource)} + className="w-full" + > + + Static / Template + Dynamic HTTP Resolver + + + +
+ +
+

Use a fixed number, SIP endpoint, or context template.

+
    +
  • SIP endpoint, e.g. PJSIP/1234
    • +
  • E.164 phone number, e.g. +1234567890
    • +
  • + Template variable, e.g. {"{{initial_context.transfer_destination}}"} +
    • +
+
+ onDestinationChange(e.target.value)} + placeholder="+1234567890, PJSIP/1234, or {{initial_context.transfer_destination}}" + /> +
+ + + +
+ +

+ Dograh sends the resolved argument dictionary to this endpoint. The endpoint must return transfer_context.destination and may return transfer_context.custom_message. +

+
+ +
+ + + +
+ +
+ + { + const value = parseInt(e.target.value) || 3000; + onResolverTimeoutMsChange(Math.min(Math.max(value, 500), 5000)); + }} + min="500" + max="5000" + className="w-36" + /> + +
+ + + +
+ +