From a2f1c670aa94c39a16e45145781db1a9169110aa Mon Sep 17 00:00:00 2001 From: Sabiha Khan Date: Fri, 10 Jul 2026 19:50:07 +0530 Subject: [PATCH] fix: remove approved routes, policy and fallback --- api/schemas/tool.py | 57 +-- .../workflow/pipecat_engine_custom_tools.py | 31 +- api/services/workflow/tools/custom_tool.py | 15 +- .../workflow/tools/transfer_resolver.py | 328 ++++++----------- api/tests/test_custom_tools.py | 124 ++++--- .../components/TransferCallToolConfig.tsx | 340 ++++++------------ ui/src/app/tools/[toolUuid]/page.tsx | 187 +++++----- ui/src/app/tools/config.tsx | 24 +- ui/src/client/index.ts | 2 +- ui/src/client/types.gen.ts | 76 ++-- 10 files changed, 475 insertions(+), 709 deletions(-) diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 159fba0d..0e88c2b4 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -20,11 +20,6 @@ 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", @@ -185,33 +180,15 @@ 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.") + 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.", @@ -226,9 +203,16 @@ class HttpTransferResolverConfig(BaseModel): 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.", + 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") @@ -242,7 +226,12 @@ class HttpTransferResolverConfig(BaseModel): 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}}." @@ -274,14 +263,6 @@ class TransferCallConfig(BaseModel): 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/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index d7d6ab82..f989ec7b 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -524,7 +524,10 @@ 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 @@ -577,7 +580,17 @@ class CustomToolManager: return resolver = config.get("resolver") if isinstance(config, dict) else None - if isinstance(resolver, dict) and resolver.get("wait_message"): + is_dynamic_transfer = ( + config.get("destination_source", "static") == "dynamic" + and isinstance(resolver, dict) + ) + resolver_phase_muted = False + + 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"]), @@ -600,6 +613,9 @@ class CustomToolManager: destination = resolved_transfer.destination timeout_seconds = resolved_transfer.timeout_seconds except TransferResolutionError as e: + if resolver_phase_muted: + self._engine.set_mute_pipeline(False) + resolver_phase_muted = False validation_error_result = { "status": "failed", "message": "I'm sorry, but I couldn't find a valid destination for this transfer.", @@ -619,6 +635,9 @@ class CustomToolManager: "action": "transfer_failed", "reason": "no_destination", } + if resolver_phase_muted: + self._engine.set_mute_pipeline(False) + resolver_phase_muted = False await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -648,6 +667,9 @@ class CustomToolManager: "action": "transfer_failed", "reason": "provider_does_not_support_transfer", } + if resolver_phase_muted: + self._engine.set_mute_pipeline(False) + resolver_phase_muted = False await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -687,7 +709,6 @@ class CustomToolManager: "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( @@ -720,7 +741,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 diff --git a/api/services/workflow/tools/custom_tool.py b/api/services/workflow/tools/custom_tool.py index 6e97225c..e8816d44 100644 --- a/api/services/workflow/tools/custom_tool.py +++ b/api/services/workflow/tools/custom_tool.py @@ -33,6 +33,19 @@ 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") + parameters = ( + resolver.get("parameters", []) if isinstance(resolver, dict) else [] + ) # Build properties and required list from parameters properties = {} @@ -271,7 +284,7 @@ async def execute_http_tool( params = None if method in ("POST", "PUT", "PATCH"): body = resolved_arguments - elif method in ("GET", "DELETE") and resolved_arguments: + elif method in ("GET", "DELETE"): params = resolved_arguments logger.info( diff --git a/api/services/workflow/tools/transfer_resolver.py b/api/services/workflow/tools/transfer_resolver.py index fc0d0b98..2e781fac 100644 --- a/api/services/workflow/tools/transfer_resolver.py +++ b/api/services/workflow/tools/transfer_resolver.py @@ -1,4 +1,4 @@ -"""Resolve transfer-call destinations from static config or dynamic resolvers.""" +"""Resolve transfer-call destinations from static config or HTTP resolvers.""" from __future__ import annotations @@ -11,6 +11,7 @@ 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 @@ -21,7 +22,6 @@ 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) @@ -53,8 +53,17 @@ def _render_value( 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 = _normalize_destination_value(destination) + value = "" if destination is None else str(destination).strip() if not value: return "" if len(value) <= 4: @@ -62,7 +71,24 @@ def _mask_destination(destination: Any) -> str: return f"***{value[-4:]}" -def _safe_log_value(value: Any) -> Any: +_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): @@ -78,104 +104,56 @@ def _safe_log_value(value: Any) -> Any: 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() + 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]], - *, - 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( +def _resolver_arguments( *, - route_key: str, - config: dict[str, Any], + resolver: dict[str, Any], + arguments: 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) +) -> dict[str, Any]: 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 {}), - ) + 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], - tool: Any, - arguments: dict[str, Any], - call_context_vars: Optional[Dict[str, Any]], - gathered_context_vars: Optional[Dict[str, Any]], + resolved_arguments: 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"} + 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( @@ -189,31 +167,25 @@ async def _execute_http_resolver( "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 {}), - } + body = resolved_arguments + 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())}" + 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.post(url, headers=headers, json=payload) + 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( @@ -256,96 +228,39 @@ async def _execute_http_resolver( 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())}" - ) + transfer_context = response_data.get("transfer_context") + if not isinstance(transfer_context, dict): raise TransferResolutionError( - "route_required", - "Transfer resolver must return an approved route for this policy", + "invalid_resolver_response", + "Transfer resolver response must contain transfer_context object", ) - destination = _normalize_destination_value(response_data.get("destination")) - try: - timeout = int(response_data.get("timeout_seconds", _base_timeout(config))) - except (TypeError, ValueError) as exc: + destination = transfer_context.get("destination") + if not isinstance(destination, str) or not destination.strip(): raise TransferResolutionError( - "invalid_timeout", "Transfer resolver returned invalid timeout" - ) from exc + "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, - timeout_seconds=min(max(timeout, 5), 120), - message=response_data.get("message"), - source="http_resolver_raw_destination", + destination=destination.strip(), + timeout_seconds=_base_timeout(config), + message=custom_message.strip() if custom_message else None, + source="http_resolver", resolution_id=resolution_id, - metadata=dict(response_data.get("metadata") or {}), ) @@ -362,7 +277,9 @@ async def resolve_transfer_config( """Resolve transfer destination and options for a transfer tool call.""" resolver = config.get("resolver") - if not isinstance(resolver, dict) or resolver.get("type") != "http": + if config.get("destination_source", "static") != "dynamic" or not isinstance( + resolver, dict + ): resolved = _resolve_static_transfer( config, call_context_vars, gathered_context_vars ) @@ -374,67 +291,36 @@ async def resolve_transfer_config( 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'))}" + "method=POST " + f"timeout_ms={resolver.get('timeout_ms', 3000)}" ) - 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" - ) + 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"route={resolved.route or ''} destination={_mask_destination(resolved.destination)} " - f"timeout={resolved.timeout_seconds}" + 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 08d681d9..eb78cc93 100644 --- a/api/tests/test_custom_tools.py +++ b/api/tests/test_custom_tools.py @@ -305,21 +305,26 @@ class TestToolToFunctionSchema: "schema_version": 1, "type": "transfer_call", "config": { + "destination_source": "dynamic", "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, - }, - ], + "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, + }, + ], + }, }, }, ) @@ -816,10 +821,10 @@ class TestCoerceParameterValue: class TestTransferResolver: - """Tests for dynamic transfer resolution policy behavior.""" + """Tests for dynamic transfer resolution behavior.""" @pytest.mark.asyncio - async def test_http_resolver_uses_static_fallback_when_policy_allows(self): + async def test_http_resolver_resolves_transfer_context_destination(self): from api.services.workflow.tools.transfer_resolver import resolve_transfer_config tool = MockToolModel( @@ -830,20 +835,33 @@ class TestTransferResolver: definition={}, ) config = { - "destination": "{{initial_context.default_destination}}", + "destination_source": "dynamic", + "destination": "", "timeout": 30, "resolver": { "type": "http", "url": "https://crm.example.com/resolve-transfer", - "policy": "approved_routes_or_static_fallback", + "headers": {"X-Tenant": "ovation"}, "timeout_ms": 3000, + "preset_parameters": [ + { + "name": "lead_id", + "type": "string", + "value_template": "{{initial_context.lead_id}}", + "required": True, + } + ], }, - "approved_routes": {}, } mock_response = Mock() mock_response.status_code = 200 - mock_response.json.return_value = {"route": "missing_route"} + mock_response.json.return_value = { + "transfer_context": { + "destination": "+14155550123", + "custom_message": "I will connect you with our Texas partner now.", + } + } with ( patch( @@ -854,24 +872,31 @@ class TestTransferResolver: ) as mock_client_class, ): mock_client = AsyncMock() - mock_client.post.return_value = mock_response + 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={"default_destination": "+14155550999"}, + call_context_vars={"lead_id": "lead-123"}, gathered_context_vars={}, organization_id=1, workflow_run_id=1, ) - assert resolved.destination == "+14155550999" + 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_rejects_raw_destination_for_route_only_policy(self): + async def test_http_resolver_requires_transfer_context_destination(self): from api.services.workflow.tools.transfer_resolver import ( TransferResolutionError, resolve_transfer_config, @@ -885,22 +910,19 @@ class TestTransferResolver: definition={}, ) config = { + "destination_source": "dynamic", "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"} - } + mock_response.json.return_value = {"transfer_context": {}} with ( patch( @@ -911,7 +933,7 @@ class TestTransferResolver: ) as mock_client_class, ): mock_client = AsyncMock() - mock_client.post.return_value = mock_response + mock_client.request.return_value = mock_response mock_client_class.return_value.__aenter__.return_value = mock_client with pytest.raises(TransferResolutionError) as exc_info: @@ -925,7 +947,7 @@ class TestTransferResolver: workflow_run_id=1, ) - assert exc_info.value.reason == "route_required" + assert exc_info.value.reason == "no_destination" class TestAuthHeaders: @@ -1404,8 +1426,8 @@ class TestCustomToolManagerUnit: 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.""" + 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() @@ -1430,21 +1452,22 @@ class TestCustomToolManagerUnit: "schema_version": 1, "type": "transfer_call", "config": { - "destination": "+15550000000", + "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.", - "policy": "approved_routes_only", - }, - "approved_routes": { - "referral_tx": { - "destination": "+14155550123", - "message": "I will connect you with our Texas partner now.", - "timeout_seconds": 45, - } + "parameters": [ + { + "name": "state", + "type": "string", + "description": "State to resolve referral partner", + "required": True, + } + ], }, }, }, @@ -1485,8 +1508,10 @@ class TestCustomToolManagerUnit: mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { - "route": "referral_tx", - "metadata": {"partner_name": "Texas Injury Partners"}, + "transfer_context": { + "destination": "+14155550123", + "custom_message": "I will connect you with our Texas partner now.", + } } with ( @@ -1514,16 +1539,23 @@ class TestCustomToolManagerUnit: ) as mock_client_class, ): mock_client = AsyncMock() - mock_client.post.return_value = mock_response + 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"] == 45 + 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 diff --git a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx index ac7d7bd9..76b87fa9 100644 --- a/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx +++ b/ui/src/app/tools/[toolUuid]/components/TransferCallToolConfig.tsx @@ -1,22 +1,27 @@ "use client"; import type { RecordingResponseSchema } from "@/client/types.gen"; -import { CredentialSelector, ParameterEditor, type ToolParameter } from "@/components/http"; +import { + CredentialSelector, + KeyValueEditor, + type KeyValueItem, + ParameterEditor, + PresetParameterEditor, + type PresetToolParameter, + type ToolParameter, + UrlInput, +} 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 { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Textarea } from "@/components/ui/textarea"; -import { Plus, Trash2 } from "lucide-react"; import { type EndCallMessageType, - type TransferApprovedRouteRow, - type TransferResolverPolicy, + type TransferDestinationSource, } from "../../config"; export interface TransferCallToolConfigProps { @@ -24,6 +29,8 @@ export interface TransferCallToolConfigProps { onNameChange: (name: string) => void; description: string; onDescriptionChange: (description: string) => void; + destinationSource: TransferDestinationSource; + onDestinationSourceChange: (source: TransferDestinationSource) => void; destination: string; onDestinationChange: (destination: string) => void; messageType: EndCallMessageType; @@ -33,26 +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; - resolverEnabled: boolean; - onResolverEnabledChange: (enabled: boolean) => 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; - resolverPolicy: TransferResolverPolicy; - onResolverPolicyChange: (policy: TransferResolverPolicy) => void; - approvedRoutes: TransferApprovedRouteRow[]; - onApprovedRoutesChange: (routes: TransferApprovedRouteRow[]) => void; - fallbackRoute: string; - onFallbackRouteChange: (route: string) => void; parameters: ToolParameter[]; onParametersChange: (parameters: ToolParameter[]) => void; + presetParameters: PresetToolParameter[]; + onPresetParametersChange: (parameters: PresetToolParameter[]) => void; } export function TransferCallToolConfig({ @@ -60,6 +63,8 @@ export function TransferCallToolConfig({ onNameChange, description, onDescriptionChange, + destinationSource, + onDestinationSourceChange, destination, onDestinationChange, messageType, @@ -71,52 +76,21 @@ export function TransferCallToolConfig({ recordings = [], timeout, onTimeoutChange, - resolverEnabled, - onResolverEnabledChange, resolverUrl, onResolverUrlChange, resolverCredentialUuid, onResolverCredentialUuidChange, + resolverHeaders, + onResolverHeadersChange, resolverTimeoutMs, onResolverTimeoutMsChange, resolverWaitMessage, onResolverWaitMessageChange, - resolverPolicy, - onResolverPolicyChange, - approvedRoutes, - onApprovedRoutesChange, - fallbackRoute, - onFallbackRouteChange, parameters, onParametersChange, + presetParameters, + onPresetParametersChange, }: 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 ( @@ -151,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" @@ -260,36 +211,63 @@ export function TransferCallToolConfig({
-
-
- -

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

-
- +
+ +

+ 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 + - {resolverEnabled && ( -
+
- + +
+

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}}"} +
    • +
+
onResolverUrlChange(e.target.value)} - placeholder="https://crm.example.com/resolve-transfer" + value={destination} + onChange={(e) => 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. +

+
+ +
+ + + +
@@ -309,6 +287,13 @@ export function TransferCallToolConfig({
+ +