mirror of
https://github.com/dograh-hq/dograh.git
synced 2026-07-25 12:01:04 +02:00
fix: review comments
This commit is contained in:
Sabiha Khan
parent
a2f1c670aa
commit
28fc1953d5
6 changed files with 273 additions and 15 deletions
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue