From 6ce1bbcbf8d43add6abffdfc91a413bc298075af Mon Sep 17 00:00:00 2001 From: Sabiha Khan Date: Fri, 10 Jul 2026 14:11:05 +0530 Subject: [PATCH] feat: enable dynamic transfer destination resolution --- api/schemas/tool.py | 78 ++++ api/services/tool_management.py | 45 +- .../workflow/pipecat_engine_custom_tools.py | 99 +++- .../workflow/tools/transfer_resolver.py | 440 ++++++++++++++++++ api/tests/test_custom_tools.py | 278 +++++++++++ .../components/TransferCallToolConfig.tsx | 271 ++++++++++- ui/src/app/tools/[toolUuid]/page.tsx | 184 +++++++- ui/src/app/tools/config.tsx | 29 ++ ui/src/client/index.ts | 2 +- ui/src/client/types.gen.ts | 102 ++++ 10 files changed, 1476 insertions(+), 52 deletions(-) create mode 100644 api/services/workflow/tools/transfer_resolver.py diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 82f23f51..159fba0d 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -20,6 +20,11 @@ DEFAULT_MCP_SSE_READ_TIMEOUT_SECS = 300 ToolParameterType = Literal["string", "number", "boolean", "object", "array"] HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE"] +TransferResolverPolicy = Literal[ + "approved_routes_only", + "approved_routes_or_static_fallback", + "allow_raw_destination", +] ToolCategoryValue = Literal[ "http_api", "end_call", @@ -180,6 +185,60 @@ class EndCallConfig(BaseModel): ) +class TransferApprovedRoute(BaseModel): + """A pre-approved destination the transfer resolver may select.""" + + destination: str = Field( + description="Phone number, SIP endpoint, or template for this route." + ) + message: Optional[str] = Field( + default=None, + description="Optional message to play before transferring to this route.", + ) + timeout_seconds: Optional[int] = Field( + default=None, + ge=5, + le=120, + description="Optional route-specific transfer answer timeout.", + ) + metadata: Optional[Dict[str, Any]] = Field( + default=None, + description="Optional non-secret route metadata for logs and resolver context.", + ) + + +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.") + 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.", + ) + policy: TransferResolverPolicy = Field( + default="approved_routes_only", + description="Controls what resolver responses are allowed to select.", + ) + + @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.""" @@ -204,6 +263,25 @@ 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.", + ) + approved_routes: Optional[Dict[str, TransferApprovedRoute]] = Field( + default=None, + description="Approved route keys that a resolver may select.", + ) + fallback_route: Optional[str] = Field( + default=None, + description="Approved route key to use when resolver resolution fails.", + ) class McpToolConfig(BaseModel): diff --git a/api/services/tool_management.py b/api/services/tool_management.py index 12161a00..a3f4ce06 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): + 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..d7d6ab82 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: @@ -526,9 +530,7 @@ class CustomToolManager: # 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,11 +562,54 @@ class CustomToolManager: ) return - destination = _render_transfer_destination( - destination, - self._engine._call_context_vars, - self._engine._gathered_context, - ) + # Get organization ID for resolver/provider configuration + organization_id = await self.get_organization_id() + if not organization_id: + validation_error_result = { + "status": "failed", + "message": "I'm sorry, there's an issue with this call transfer. Please contact support.", + "action": "transfer_failed", + "reason": "no_organization_id", + } + await self._handle_transfer_result( + validation_error_result, function_call_params, properties + ) + return + + resolver = config.get("resolver") if isinstance(config, dict) else None + if isinstance(resolver, dict) 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: + 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(): @@ -579,23 +624,19 @@ class CustomToolManager: ) return - played = await self._play_config_message(config) - if played: - self._engine._queued_speech_mute_state = "waiting" - - # Get organization ID for provider configuration - organization_id = await self.get_organization_id() - if not organization_id: - validation_error_result = { - "status": "failed", - "message": "I'm sorry, there's an issue with this call transfer. Please contact support.", - "action": "transfer_failed", - "reason": "no_organization_id", - } - await self._handle_transfer_result( - validation_error_result, function_call_params, properties + if resolved_transfer.message: + await self._engine.task.queue_frame( + TTSSpeakFrame( + resolved_transfer.message, + append_to_context=False, + persist_to_logs=True, + ) ) - return + 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 @@ -639,6 +680,16 @@ 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"route={resolved_transfer.route or ''} " + f"destination={masked_destination} timeout={timeout_seconds}" + ) transfer_result = await provider.transfer_call( destination=destination, transfer_id=transfer_id, diff --git a/api/services/workflow/tools/transfer_resolver.py b/api/services/workflow/tools/transfer_resolver.py new file mode 100644 index 00000000..fc0d0b98 --- /dev/null +++ b/api/services/workflow/tools/transfer_resolver.py @@ -0,0 +1,440 @@ +"""Resolve transfer-call destinations from static config or dynamic 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.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 + route: 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 _mask_destination(destination: Any) -> str: + value = _normalize_destination_value(destination) + if not value: + return "" + if len(value) <= 4: + return "***" + return f"***{value[-4:]}" + + +def _safe_log_value(value: Any) -> Any: + 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(value) for key, value in (data or {}).items()} + + +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 _normalize_destination_value(value: Any) -> str: + if isinstance(value, dict): + value = value.get("value") + if value is None: + return "" + return str(value).strip() + + +def _resolve_static_transfer( + config: dict[str, Any], + call_context_vars: Optional[Dict[str, Any]], + gathered_context_vars: Optional[Dict[str, Any]], + *, + resolution_id: Optional[str] = None, + source: str = "static", +) -> ResolvedTransferConfig: + return ResolvedTransferConfig( + destination=_render_value( + config.get("destination", ""), call_context_vars, gathered_context_vars + ), + timeout_seconds=_base_timeout(config), + source=source, + resolution_id=resolution_id, + ) + + +def _expand_approved_route( + *, + route_key: str, + config: dict[str, Any], + call_context_vars: Optional[Dict[str, Any]], + gathered_context_vars: Optional[Dict[str, Any]], + resolution_id: Optional[str] = None, + source: str = "approved_route", +) -> ResolvedTransferConfig: + approved_routes = config.get("approved_routes") or {} + if not isinstance(approved_routes, dict) or route_key not in approved_routes: + raise TransferResolutionError( + "unknown_route", f"Resolver returned unknown transfer route '{route_key}'" + ) + + route = approved_routes[route_key] or {} + if not isinstance(route, dict): + raise TransferResolutionError( + "invalid_route", f"Transfer route '{route_key}' is not configured correctly" + ) + + destination = _render_value( + route.get("destination", ""), call_context_vars, gathered_context_vars + ) + timeout = route.get("timeout_seconds") + if timeout is None: + timeout = _base_timeout(config) + try: + timeout_int = int(timeout) + except (TypeError, ValueError) as exc: + raise TransferResolutionError( + "invalid_timeout", f"Transfer route '{route_key}' has invalid timeout" + ) from exc + + return ResolvedTransferConfig( + destination=destination, + timeout_seconds=min(max(timeout_int, 5), 120), + message=route.get("message"), + route=route_key, + source=source, + resolution_id=resolution_id, + metadata=dict(route.get("metadata") or {}), + ) + + +async def _execute_http_resolver( + *, + resolver: dict[str, Any], + tool: 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], + resolution_id: str, +) -> dict[str, Any]: + url = resolver.get("url", "") + validate_user_configured_service_url(url, field_name="config.resolver.url") + + headers = {"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", + ) + + payload = { + "event": "transfer_resolution_requested", + "workflow_run_id": workflow_run_id, + "tool": { + "tool_uuid": getattr(tool, "tool_uuid", None), + "name": getattr(tool, "name", None), + }, + "arguments": arguments or {}, + "initial_context": dict(call_context_vars or {}), + "gathered_context": dict(gathered_context_vars or {}), + } + timeout_seconds = float(resolver.get("timeout_ms", 3000)) / 1000.0 + logger.debug( + "Transfer resolver request prepared " + f"resolution_id={resolution_id} " + f"argument_keys={list((arguments or {}).keys())} " + f"arguments={_safe_log_dict(arguments)} " + f"initial_context_keys={list((call_context_vars or {}).keys())} " + f"gathered_context_keys={list((gathered_context_vars or {}).keys())}" + ) + + try: + started_at = time.monotonic() + async with httpx.AsyncClient(timeout=timeout_seconds) as client: + response = await client.post(url, headers=headers, json=payload) + 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 _fallback_resolution( + *, + config: dict[str, Any], + resolver: dict[str, Any], + call_context_vars: Optional[Dict[str, Any]], + gathered_context_vars: Optional[Dict[str, Any]], + resolution_id: Optional[str] = None, +) -> Optional[ResolvedTransferConfig]: + fallback_route = config.get("fallback_route") + if fallback_route: + return _expand_approved_route( + route_key=str(fallback_route), + config=config, + call_context_vars=call_context_vars, + gathered_context_vars=gathered_context_vars, + resolution_id=resolution_id, + source="fallback_route", + ) + + if resolver.get("policy") == "approved_routes_or_static_fallback": + return _resolve_static_transfer( + config, + call_context_vars, + gathered_context_vars, + resolution_id=resolution_id, + source="static_fallback", + ) + + return None + + +def _resolve_from_response( + *, + response_data: dict[str, Any], + config: dict[str, Any], + resolver: dict[str, Any], + call_context_vars: Optional[Dict[str, Any]], + gathered_context_vars: Optional[Dict[str, Any]], + resolution_id: str, +) -> ResolvedTransferConfig: + route_key = response_data.get("route") + if route_key: + resolved = _expand_approved_route( + route_key=str(route_key), + config=config, + call_context_vars=call_context_vars, + gathered_context_vars=gathered_context_vars, + resolution_id=resolution_id, + source="http_resolver_route", + ) + resolved.metadata.update(dict(response_data.get("metadata") or {})) + if response_data.get("message"): + resolved.message = str(response_data["message"]) + if response_data.get("timeout_seconds") is not None: + try: + resolved.timeout_seconds = min( + max(int(response_data["timeout_seconds"]), 5), 120 + ) + except (TypeError, ValueError) as exc: + raise TransferResolutionError( + "invalid_timeout", "Transfer resolver returned invalid timeout" + ) from exc + return resolved + + policy = resolver.get("policy", "approved_routes_only") + if policy != "allow_raw_destination": + logger.warning( + "Transfer resolver rejected response " + f"resolution_id={resolution_id} reason=route_required " + f"policy={policy} response_keys={list(response_data.keys())}" + ) + raise TransferResolutionError( + "route_required", + "Transfer resolver must return an approved route for this policy", + ) + + destination = _normalize_destination_value(response_data.get("destination")) + try: + timeout = int(response_data.get("timeout_seconds", _base_timeout(config))) + except (TypeError, ValueError) as exc: + raise TransferResolutionError( + "invalid_timeout", "Transfer resolver returned invalid timeout" + ) from exc + return ResolvedTransferConfig( + destination=destination, + timeout_seconds=min(max(timeout, 5), 120), + message=response_data.get("message"), + source="http_resolver_raw_destination", + resolution_id=resolution_id, + metadata=dict(response_data.get("metadata") or {}), + ) + + +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 not isinstance(resolver, dict) or resolver.get("type") != "http": + 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()) + approved_routes = config.get("approved_routes") or {} + 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')} " + f"policy={resolver.get('policy', 'approved_routes_only')} " + f"timeout_ms={resolver.get('timeout_ms', 3000)} " + f"route_count={len(approved_routes) if isinstance(approved_routes, dict) else 0} " + f"fallback_route={config.get('fallback_route') or ''} " + f"static_fallback_available={bool(config.get('destination'))}" + ) + + try: + response_data = await _execute_http_resolver( + resolver=resolver, + tool=tool, + arguments=arguments, + call_context_vars=call_context_vars, + gathered_context_vars=gathered_context_vars, + organization_id=organization_id, + workflow_run_id=workflow_run_id, + resolution_id=resolution_id, + ) + resolved = _resolve_from_response( + response_data=response_data, + config=config, + resolver=resolver, + call_context_vars=call_context_vars, + gathered_context_vars=gathered_context_vars, + resolution_id=resolution_id, + ) + except TransferResolutionError as exc: + fallback = _fallback_resolution( + config=config, + resolver=resolver, + call_context_vars=call_context_vars, + gathered_context_vars=gathered_context_vars, + resolution_id=resolution_id, + ) + if fallback: + logger.warning( + "Transfer resolver failed; using configured fallback " + f"resolution_id={resolution_id} reason={exc.reason} " + f"fallback_source={fallback.source} route={fallback.route or ''} " + f"destination={_mask_destination(fallback.destination)}" + ) + return fallback + logger.warning( + "Transfer resolver failed without fallback " + f"resolution_id={resolution_id} reason={exc.reason}" + ) + raise + + if not resolved.destination: + raise TransferResolutionError( + "no_destination", "Transfer resolver did not provide a destination" + ) + logger.info( + "Transfer destination resolved " + f"resolution_id={resolution_id} source={resolved.source} " + f"route={resolved.route or ''} destination={_mask_destination(resolved.destination)} " + f"timeout={resolved.timeout_seconds}" + ) + return resolved diff --git a/api/tests/test_custom_tools.py b/api/tests/test_custom_tools.py index e3523edc..08d681d9 100644 --- a/api/tests/test_custom_tools.py +++ b/api/tests/test_custom_tools.py @@ -294,6 +294,43 @@ 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": "", + "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"] + class TestExecuteHttpTool: """Tests for execute_http_tool function.""" @@ -778,6 +815,119 @@ class TestCoerceParameterValue: _coerce_parameter_value(value, "array") +class TestTransferResolver: + """Tests for dynamic transfer resolution policy behavior.""" + + @pytest.mark.asyncio + async def test_http_resolver_uses_static_fallback_when_policy_allows(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": "{{initial_context.default_destination}}", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "policy": "approved_routes_or_static_fallback", + "timeout_ms": 3000, + }, + "approved_routes": {}, + } + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"route": "missing_route"} + + 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.post.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={"default_destination": "+14155550999"}, + gathered_context_vars={}, + organization_id=1, + workflow_run_id=1, + ) + + assert resolved.destination == "+14155550999" + assert resolved.timeout_seconds == 30 + + @pytest.mark.asyncio + async def test_http_resolver_rejects_raw_destination_for_route_only_policy(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": "", + "timeout": 30, + "resolver": { + "type": "http", + "url": "https://crm.example.com/resolve-transfer", + "policy": "approved_routes_only", + "timeout_ms": 3000, + }, + "approved_routes": {}, + } + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "destination": {"type": "phone_number", "value": "+14155550123"} + } + + 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.post.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 == "route_required" + + class TestAuthHeaders: """Tests for auth header building utilities.""" @@ -1253,6 +1403,134 @@ 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_approved_route(self): + """HTTP resolver route responses expand to approved transfer destinations.""" + 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": "+15550000000", + "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.", + "policy": "approved_routes_only", + }, + "approved_routes": { + "referral_tx": { + "destination": "+14155550123", + "message": "I will connect you with our Texas partner now.", + "timeout_seconds": 45, + } + }, + }, + }, + ) + 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 = { + "route": "referral_tx", + "metadata": {"partner_name": "Texas Injury Partners"}, + } + + 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.post.return_value = mock_response + mock_client_class.return_value.__aenter__.return_value = mock_client + + await handler(mock_params) + + provider.transfer_call.assert_awaited_once() + transfer_kwargs = provider.transfer_call.await_args.kwargs + assert transfer_kwargs["destination"] == "+14155550123" + assert transfer_kwargs["timeout"] == 45 + assert result_received["status"] == "transfer_failed" + + 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_propagates_provider_destination_error(self): """Provider-specific destination failures are returned through the tool result.""" diff --git a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx index 1268b99d..ac7d7bd9 100644 --- a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx +++ b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx @@ -1,14 +1,23 @@ "use client"; import type { RecordingResponseSchema } from "@/client/types.gen"; +import { CredentialSelector, ParameterEditor, type ToolParameter } from "@/components/http"; import { RecordingSelect, StaticTextWarning } from "@/components/flow/TextOrAudioInput"; +import { Button } from "@/components/ui/button"; 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 { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; +import { Switch } from "@/components/ui/switch"; import { Textarea } from "@/components/ui/textarea"; +import { Plus, Trash2 } from "lucide-react"; -import { type EndCallMessageType } from "../../config"; +import { + type EndCallMessageType, + type TransferApprovedRouteRow, + type TransferResolverPolicy, +} from "../../config"; export interface TransferCallToolConfigProps { name: string; @@ -26,6 +35,24 @@ export interface TransferCallToolConfigProps { recordings?: RecordingResponseSchema[]; timeout?: number; // Make optional to match API type onTimeoutChange: (timeout: number) => void; + resolverEnabled: boolean; + onResolverEnabledChange: (enabled: boolean) => void; + resolverUrl: string; + onResolverUrlChange: (url: string) => void; + resolverCredentialUuid: string; + onResolverCredentialUuidChange: (uuid: string) => void; + resolverTimeoutMs: number; + onResolverTimeoutMsChange: (timeoutMs: number) => void; + resolverWaitMessage: string; + onResolverWaitMessageChange: (message: string) => void; + resolverPolicy: TransferResolverPolicy; + onResolverPolicyChange: (policy: TransferResolverPolicy) => void; + approvedRoutes: TransferApprovedRouteRow[]; + onApprovedRoutesChange: (routes: TransferApprovedRouteRow[]) => void; + fallbackRoute: string; + onFallbackRouteChange: (route: string) => void; + parameters: ToolParameter[]; + onParametersChange: (parameters: ToolParameter[]) => void; } export function TransferCallToolConfig({ @@ -44,7 +71,52 @@ export function TransferCallToolConfig({ recordings = [], timeout, onTimeoutChange, + resolverEnabled, + onResolverEnabledChange, + resolverUrl, + onResolverUrlChange, + resolverCredentialUuid, + onResolverCredentialUuidChange, + resolverTimeoutMs, + onResolverTimeoutMsChange, + resolverWaitMessage, + onResolverWaitMessageChange, + resolverPolicy, + onResolverPolicyChange, + approvedRoutes, + onApprovedRoutesChange, + fallbackRoute, + onFallbackRouteChange, + parameters, + onParametersChange, }: TransferCallToolConfigProps) { + const updateRoute = ( + index: number, + patch: Partial, + ) => { + onApprovedRoutesChange( + approvedRoutes.map((route, i) => + i === index ? { ...route, ...patch } : route, + ), + ); + }; + + const removeRoute = (index: number) => { + onApprovedRoutesChange(approvedRoutes.filter((_, i) => i !== index)); + }; + + const addRoute = () => { + onApprovedRoutesChange([ + ...approvedRoutes, + { + key: "", + destination: "", + message: "", + timeout_seconds: 30, + }, + ]); + }; + return ( @@ -186,6 +258,203 @@ export function TransferCallToolConfig({ Default: 30 seconds + +
+
+
+ +

+ Resolve the destination from an HTTP endpoint when the transfer tool is called. +

+
+ +
+ + {resolverEnabled && ( +
+
+ + onResolverUrlChange(e.target.value)} + placeholder="https://crm.example.com/resolve-transfer" + /> +
+ + + +
+ + { + const value = parseInt(e.target.value) || 3000; + onResolverTimeoutMsChange(Math.min(Math.max(value, 500), 5000)); + }} + min="500" + max="5000" + className="w-36" + /> + +
+ +
+ +