diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 0e88c2b4..4cccee4f 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -264,6 +264,18 @@ class TransferCallConfig(BaseModel): 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): """Configuration for a customer MCP server tool definition.""" diff --git a/api/services/tool_management.py b/api/services/tool_management.py index a3f4ce06..ad8bb96a 100644 --- a/api/services/tool_management.py +++ b/api/services/tool_management.py @@ -82,7 +82,7 @@ def _credential_uuids_from_definition(definition: dict[str, Any]) -> list[str]: resolver = config.get("resolver") if isinstance(resolver, dict): resolver_credential_uuid = resolver.get("credential_uuid") - if isinstance(resolver_credential_uuid, str): + if isinstance(resolver_credential_uuid, str) and resolver_credential_uuid: credential_uuids.append(resolver_credential_uuid) return list(dict.fromkeys(credential_uuids)) diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index f989ec7b..d596cfc9 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -317,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( @@ -328,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.""" @@ -586,6 +608,13 @@ class CustomToolManager: ) 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 @@ -613,9 +642,7 @@ 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 + 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.", @@ -635,9 +662,7 @@ class CustomToolManager: "action": "transfer_failed", "reason": "no_destination", } - if resolver_phase_muted: - self._engine.set_mute_pipeline(False) - resolver_phase_muted = False + clear_transfer_setup_mute_state() await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -667,9 +692,7 @@ 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 + clear_transfer_setup_mute_state() await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -720,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", @@ -819,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 e8816d44..cdecc68c 100644 --- a/api/services/workflow/tools/custom_tool.py +++ b/api/services/workflow/tools/custom_tool.py @@ -43,9 +43,10 @@ def tool_to_function_schema(tool: Any) -> Dict[str, Any]: and config.get("destination_source", "static") == "dynamic" ): resolver = config.get("resolver") - parameters = ( - resolver.get("parameters", []) if isinstance(resolver, dict) else [] - ) + if isinstance(resolver, dict): + parameters = resolver.get("parameters", []) or [] + else: + parameters = [] # Build properties and required list from parameters properties = {} @@ -284,7 +285,7 @@ async def execute_http_tool( params = None if method in ("POST", "PUT", "PATCH"): body = resolved_arguments - elif method in ("GET", "DELETE"): + elif method in ("GET", "DELETE") and resolved_arguments: params = resolved_arguments logger.info( diff --git a/api/tests/test_custom_tools.py b/api/tests/test_custom_tools.py index eb78cc93..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.""" @@ -336,6 +355,32 @@ class TestToolToFunctionSchema: 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.""" @@ -573,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.""" @@ -823,6 +905,34 @@ class TestCoerceParameterValue: 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 @@ -1563,6 +1673,89 @@ class TestCustomToolManagerUnit: 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