dograh/api/tests/test_custom_tools.py

1482 lines
54 KiB
Python

"""Tests for custom tool integration with PipecatEngine.
This module tests:
1. tool_to_function_schema - converting tool models to LLM function schemas
2. execute_http_tool - executing HTTP API tools
3. CustomToolManager - tool registration and handler execution
4. End-to-end LLM generation with custom tool calls
"""
from dataclasses import dataclass
from types import SimpleNamespace
from typing import Any, Dict
from unittest.mock import AsyncMock, Mock, patch
import pytest
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallsFromLLMInfoFrame,
FunctionCallsStartedFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
UserTurnInferenceCompletedFrame,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.services.llm_service import FunctionCallParams
from api.enums import WorkflowRunMode
from api.services.workflow.pipecat_engine_custom_tools import get_function_schema
from api.services.workflow.tools.custom_tool import (
_coerce_parameter_value,
execute_http_tool,
tool_to_function_schema,
)
from pipecat.tests import MockLLMService, run_test
@dataclass
class MockToolModel:
"""Mock tool model for testing."""
tool_uuid: str
name: str
description: str
category: str
definition: Dict[str, Any]
class TestToolToFunctionSchema:
"""Tests for tool_to_function_schema function."""
def test_simple_tool_with_string_parameter(self):
"""Test converting a simple tool with one string parameter."""
tool = MockToolModel(
tool_uuid="test-uuid-1",
name="Get Weather",
description="Get current weather for a location",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "GET",
"url": "https://api.weather.com/current",
"parameters": [
{
"name": "location",
"type": "string",
"description": "City name",
"required": True,
}
],
},
},
)
schema = tool_to_function_schema(tool)
assert schema["type"] == "function"
assert schema["function"]["name"] == "get_weather"
assert schema["function"]["description"] == "Get current weather for a location"
assert schema["function"]["parameters"]["type"] == "object"
assert "location" in schema["function"]["parameters"]["properties"]
assert (
schema["function"]["parameters"]["properties"]["location"]["type"]
== "string"
)
assert (
schema["function"]["parameters"]["properties"]["location"]["description"]
== "City name"
)
assert "location" in schema["function"]["parameters"]["required"]
assert schema["_tool_uuid"] == "test-uuid-1"
def test_tool_with_multiple_parameter_types(self):
"""Test converting a tool with string, number, and boolean parameters."""
tool = MockToolModel(
tool_uuid="test-uuid-2",
name="Book Appointment",
description="Book an appointment with the service",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/appointments",
"parameters": [
{
"name": "customer_name",
"type": "string",
"description": "Customer's full name",
"required": True,
},
{
"name": "duration_minutes",
"type": "number",
"description": "Appointment duration in minutes",
"required": True,
},
{
"name": "is_priority",
"type": "boolean",
"description": "Whether this is a priority appointment",
"required": False,
},
],
},
},
)
schema = tool_to_function_schema(tool)
props = schema["function"]["parameters"]["properties"]
assert props["customer_name"]["type"] == "string"
assert props["duration_minutes"]["type"] == "number"
assert props["is_priority"]["type"] == "boolean"
required = schema["function"]["parameters"]["required"]
assert "customer_name" in required
assert "duration_minutes" in required
assert "is_priority" not in required
def test_tool_with_object_and_array_parameters(self):
"""Test converting a tool with object and array parameters."""
tool = MockToolModel(
tool_uuid="test-uuid-nested",
name="Create Booking",
description="Create a booking with nested details",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/bookings",
"parameters": [
{
"name": "booking",
"type": "object",
"description": "Nested booking payload",
"required": True,
},
{
"name": "attendees",
"type": "array",
"description": "Booking attendees",
"required": False,
},
],
},
},
)
schema = tool_to_function_schema(tool)
props = schema["function"]["parameters"]["properties"]
assert props["booking"] == {
"type": "object",
"additionalProperties": True,
"description": "Nested booking payload",
}
assert props["attendees"] == {
"type": "array",
"items": {},
"description": "Booking attendees",
}
def test_preset_parameters_are_not_exposed_to_llm_schema(self):
"""Test that preset parameters are injected at runtime, not shown to the LLM."""
tool = MockToolModel(
tool_uuid="test-uuid-preset",
name="Lookup Customer",
description="Lookup a customer using contextual identifiers",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/customers/lookup",
"parameters": [
{
"name": "customer_name",
"type": "string",
"description": "Customer name spoken by the caller",
"required": True,
}
],
"preset_parameters": [
{
"name": "phone_number",
"type": "string",
"value_template": "{{initial_context.phone_number}}",
"required": True,
}
],
},
},
)
schema = tool_to_function_schema(tool)
props = schema["function"]["parameters"]["properties"]
assert "customer_name" in props
assert "phone_number" not in props
def test_tool_name_sanitization(self):
"""Test that tool names with special characters are sanitized."""
tool = MockToolModel(
tool_uuid="test-uuid-3",
name="Get User's Account Info!!!",
description="Get account information",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "GET",
"url": "https://api.example.com/account",
"parameters": [],
},
},
)
schema = tool_to_function_schema(tool)
# Name should be lowercase with underscores only
assert schema["function"]["name"] == "get_user_s_account_info"
def test_tool_with_no_parameters(self):
"""Test converting a tool with no parameters."""
tool = MockToolModel(
tool_uuid="test-uuid-4",
name="Ping Server",
description="Check if server is alive",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "GET",
"url": "https://api.example.com/ping",
},
},
)
schema = tool_to_function_schema(tool)
assert schema["function"]["parameters"]["properties"] == {}
assert schema["function"]["parameters"]["required"] == []
def test_tool_without_description_uses_fallback(self):
"""Test that tools without description use fallback."""
tool = MockToolModel(
tool_uuid="test-uuid-5",
name="My Tool",
description=None,
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/tool",
},
},
)
schema = tool_to_function_schema(tool)
assert schema["function"]["description"] == "Execute My Tool tool"
class TestExecuteHttpTool:
"""Tests for execute_http_tool function."""
@pytest.mark.asyncio
async def test_post_request_sends_json_body(self):
"""Test that POST requests send arguments as JSON body."""
tool = MockToolModel(
tool_uuid="test-uuid",
name="Create User",
description="Create a new user",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/users",
"timeout_ms": 5000,
},
},
)
arguments = {"name": "John", "email": "john@example.com"}
with patch(
"api.services.workflow.tools.custom_tool.httpx.AsyncClient"
) as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 201
mock_response.json.return_value = {"id": 123, "name": "John"}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await execute_http_tool(tool, arguments)
# Verify request was made with JSON body
mock_client.request.assert_called_once()
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["method"] == "POST"
assert call_kwargs["url"] == "https://api.example.com/users"
assert call_kwargs["json"] == arguments
assert call_kwargs["params"] is None
assert result["status"] == "success"
assert result["status_code"] == 201
assert result["data"]["id"] == 123
@pytest.mark.asyncio
async def test_post_request_sends_nested_json_body(self):
"""Test that POST requests preserve nested arguments in the JSON body."""
tool = MockToolModel(
tool_uuid="test-uuid-nested",
name="Create Booking",
description="Create a nested booking",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/bookings",
"timeout_ms": 5000,
},
},
)
arguments = {
"booking": {
"start": "2026-05-28T10:00:00Z",
"attendee": {"name": "Jane", "email": "jane@example.com"},
"metadata": {"source": "voice"},
}
}
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 = {"bookingId": "booking-123"}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await execute_http_tool(tool, arguments)
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["json"] == arguments
assert isinstance(call_kwargs["json"]["booking"], dict)
assert isinstance(call_kwargs["json"]["booking"]["attendee"], dict)
assert result["status"] == "success"
@pytest.mark.asyncio
async def test_post_request_injects_preset_parameters(self):
"""Test that preset parameters are resolved from runtime context."""
tool = MockToolModel(
tool_uuid="test-uuid-preset",
name="Create Lead",
description="Create a lead with caller context",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/leads",
"timeout_ms": 5000,
"preset_parameters": [
{
"name": "phone_number",
"type": "string",
"value_template": "{{initial_context.phone_number}}",
"required": True,
},
{
"name": "customer_id",
"type": "number",
"value_template": "{{gathered_context.customer_id}}",
"required": True,
},
{
"name": "is_vip",
"type": "boolean",
"value_template": "{{initial_context.is_vip}}",
"required": False,
},
],
},
},
)
arguments = {"name": "John"}
with patch(
"api.services.workflow.tools.custom_tool.httpx.AsyncClient"
) as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 201
mock_response.json.return_value = {"id": 123}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await execute_http_tool(
tool,
arguments,
call_context_vars={
"phone_number": "+14155550123",
"is_vip": "true",
},
gathered_context_vars={"customer_id": "42"},
)
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["json"] == {
"name": "John",
"phone_number": "+14155550123",
"customer_id": 42,
"is_vip": True,
}
assert result["status"] == "success"
@pytest.mark.asyncio
async def test_missing_required_preset_parameter_returns_error(self):
"""Test that required preset parameters fail before the HTTP request."""
tool = MockToolModel(
tool_uuid="test-uuid-preset-error",
name="Create Lead",
description="Create a lead with caller context",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/leads",
"timeout_ms": 5000,
"preset_parameters": [
{
"name": "phone_number",
"type": "string",
"value_template": "{{initial_context.phone_number}}",
"required": True,
}
],
},
},
)
result = await execute_http_tool(tool, {"name": "John"}, call_context_vars={})
assert result["status"] == "error"
assert "phone_number" in result["error"]
@pytest.mark.asyncio
async def test_get_request_sends_query_params(self):
"""Test that GET requests send arguments as query parameters."""
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",
"timeout_ms": 5000,
},
},
)
arguments = {"query": "john", "limit": 10}
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, arguments)
# Verify request was made with query params
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["method"] == "GET"
assert call_kwargs["json"] is None
assert call_kwargs["params"] == arguments
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."""
tool = MockToolModel(
tool_uuid="test-uuid",
name="Delete User",
description="Delete a user",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "DELETE",
"url": "https://api.example.com/users",
"timeout_ms": 5000,
},
},
)
arguments = {"user_id": "123"}
with patch(
"api.services.workflow.tools.custom_tool.httpx.AsyncClient"
) as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 204
mock_response.json.return_value = {}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
await execute_http_tool(tool, arguments)
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["method"] == "DELETE"
assert call_kwargs["json"] is None
assert call_kwargs["params"] == arguments
@pytest.mark.asyncio
async def test_timeout_error_handling(self):
"""Test that timeout errors are handled gracefully."""
import httpx
tool = MockToolModel(
tool_uuid="test-uuid",
name="Slow API",
description="A slow API call",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/slow",
"timeout_ms": 1000,
},
},
)
with patch(
"api.services.workflow.tools.custom_tool.httpx.AsyncClient"
) as mock_client_class:
mock_client = AsyncMock()
mock_client.request.side_effect = httpx.TimeoutException(
"Request timed out"
)
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await execute_http_tool(tool, {})
assert result["status"] == "error"
assert "timed out" in result["error"]
@pytest.mark.asyncio
async def test_request_includes_custom_headers(self):
"""Test that custom headers are included in the request."""
tool = MockToolModel(
tool_uuid="test-uuid",
name="API with Headers",
description="API that requires headers",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/data",
"headers": {
"X-API-Key": "secret-key",
"X-Custom-Header": "custom-value",
},
"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 = {"success": True}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
await execute_http_tool(tool, {"data": "test"})
call_kwargs = mock_client.request.call_args.kwargs
assert call_kwargs["headers"]["X-API-Key"] == "secret-key"
assert call_kwargs["headers"]["X-Custom-Header"] == "custom-value"
@pytest.mark.asyncio
async def test_request_includes_auth_header_from_credential(self):
"""Test that auth headers from credentials are included in the request."""
tool = MockToolModel(
tool_uuid="test-uuid",
name="Authenticated API",
description="API that requires authentication",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/secure",
"credential_uuid": "cred-uuid-123",
"timeout_ms": 5000,
},
},
)
# Mock credential
mock_credential = Mock()
mock_credential.name = "API Token"
mock_credential.credential_type = "bearer_token"
mock_credential.credential_data = {"token": "my-secret-token"}
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 = {"success": True}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch("api.services.workflow.tools.custom_tool.db_client") as mock_db:
mock_db.get_credential_by_uuid = AsyncMock(return_value=mock_credential)
await execute_http_tool(tool, {"data": "test"}, organization_id=1)
# Verify credential was fetched
mock_db.get_credential_by_uuid.assert_called_once_with(
"cred-uuid-123", 1
)
# Verify auth header was added
call_kwargs = mock_client.request.call_args.kwargs
assert (
call_kwargs["headers"]["Authorization"] == "Bearer my-secret-token"
)
@pytest.mark.asyncio
async def test_no_credential_lookup_without_organization_id(self):
"""Test that credential lookup is skipped without organization_id."""
tool = MockToolModel(
tool_uuid="test-uuid",
name="API with Credential",
description="API with credential configured",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/secure",
"credential_uuid": "cred-uuid-123",
"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 = {"success": True}
mock_client.request.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch("api.services.workflow.tools.custom_tool.db_client") as mock_db:
# Call without organization_id
await execute_http_tool(tool, {"data": "test"})
# Verify credential lookup was NOT called
mock_db.get_credential_by_uuid.assert_not_called()
class TestCoerceParameterValue:
"""Tests for _coerce_parameter_value function."""
def test_object_value_returns_dict_unchanged(self):
"""Test that object parameters preserve dict values."""
value = {"attendee": {"name": "Jane"}}
assert _coerce_parameter_value(value, "object") is value
def test_object_value_parses_json_string(self):
"""Test that object parameters parse JSON string values."""
value = '{"attendee": {"name": "Jane"}}'
assert _coerce_parameter_value(value, "object") == {
"attendee": {"name": "Jane"}
}
def test_array_value_returns_list_unchanged(self):
"""Test that array parameters preserve list values."""
value = [{"name": "Jane"}, {"name": "Sam"}]
assert _coerce_parameter_value(value, "array") is value
def test_array_value_parses_json_string(self):
"""Test that array parameters parse JSON string values."""
value = '[{"name": "Jane"}, {"name": "Sam"}]'
assert _coerce_parameter_value(value, "array") == [
{"name": "Jane"},
{"name": "Sam"},
]
@pytest.mark.parametrize("value", ["not json", "[]", "null"])
def test_object_value_rejects_invalid_or_wrong_shape(self, value):
"""Test that object parameters require a JSON object."""
with pytest.raises(ValueError, match="Cannot convert"):
_coerce_parameter_value(value, "object")
@pytest.mark.parametrize("value", ["not json", "{}", "null"])
def test_array_value_rejects_invalid_or_wrong_shape(self, value):
"""Test that array parameters require a JSON array."""
with pytest.raises(ValueError, match="Cannot convert"):
_coerce_parameter_value(value, "array")
class TestAuthHeaders:
"""Tests for auth header building utilities."""
def test_bearer_token_auth(self):
"""Test building bearer token auth header."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "bearer_token"
mock_credential.credential_data = {"token": "abc123"}
header = build_auth_header(mock_credential)
assert header == {"Authorization": "Bearer abc123"}
def test_api_key_auth(self):
"""Test building API key auth header."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "api_key"
mock_credential.credential_data = {
"header_name": "X-API-Key",
"api_key": "secret-key-123",
}
header = build_auth_header(mock_credential)
assert header == {"X-API-Key": "secret-key-123"}
def test_basic_auth(self):
"""Test building basic auth header."""
import base64
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "basic_auth"
mock_credential.credential_data = {
"username": "user",
"password": "pass123",
}
header = build_auth_header(mock_credential)
expected_encoded = base64.b64encode(b"user:pass123").decode()
assert header == {"Authorization": f"Basic {expected_encoded}"}
def test_custom_header_auth(self):
"""Test building custom header auth."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "custom_header"
mock_credential.credential_data = {
"header_name": "X-Custom-Auth",
"header_value": "custom-value-123",
}
header = build_auth_header(mock_credential)
assert header == {"X-Custom-Auth": "custom-value-123"}
def test_unknown_auth_type_returns_empty(self):
"""Test that unknown auth types return empty dict."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "unknown_type"
mock_credential.credential_data = {}
header = build_auth_header(mock_credential)
assert header == {}
def test_none_credential_type_returns_empty(self):
"""Test that 'none' credential type returns empty dict."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "none"
mock_credential.credential_data = {}
header = build_auth_header(mock_credential)
assert header == {}
def test_build_auth_header_from_data(self):
"""Test building auth header from raw data."""
from api.utils.credential_auth import build_auth_header_from_data
header = build_auth_header_from_data(
credential_type="bearer_token",
credential_data={"token": "my-token"},
)
assert header == {"Authorization": "Bearer my-token"}
def test_api_key_default_header_name(self):
"""Test that API key uses default header name if not specified."""
from api.utils.credential_auth import build_auth_header
mock_credential = Mock()
mock_credential.credential_type = "api_key"
mock_credential.credential_data = {"api_key": "key123"}
header = build_auth_header(mock_credential)
assert header == {"X-API-Key": "key123"}
class TestCustomToolManagerIntegration:
"""Integration tests for CustomToolManager with MockLLMService."""
@pytest.mark.asyncio
async def test_llm_calls_custom_tool_handler(self):
"""Test that when LLM makes a function call, the custom tool handler is executed."""
# Create function call chunks that simulate LLM calling a custom tool
chunks = MockLLMService.create_function_call_chunks(
function_name="book_appointment",
arguments={"customer_name": "John Doe", "date": "2024-01-15"},
tool_call_id="call_custom_123",
)
llm = MockLLMService(mock_chunks=chunks, chunk_delay=0.001)
# Track if our handler was called
handler_called = False
received_arguments = None
async def mock_book_appointment(params: FunctionCallParams):
nonlocal handler_called, received_arguments
handler_called = True
received_arguments = params.arguments
await params.result_callback({"status": "booked", "confirmation": "ABC123"})
# Register the function handler
llm.register_function("book_appointment", mock_book_appointment)
# Create context and run
messages = [
{"role": "user", "content": "Book an appointment for John Doe on Jan 15"}
]
context = LLMContext(messages)
pipeline = Pipeline([llm])
frames_to_send = [LLMContextFrame(context)]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=[
LLMFullResponseStartFrame,
FunctionCallsFromLLMInfoFrame,
UserTurnInferenceCompletedFrame,
FunctionCallsStartedFrame,
LLMFullResponseEndFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
],
)
# Verify handler was called with correct arguments
assert handler_called, "Custom tool handler should have been called"
assert received_arguments == {"customer_name": "John Doe", "date": "2024-01-15"}
@pytest.mark.asyncio
async def test_multiple_custom_tools_can_be_registered(self):
"""Test that multiple custom tools can be registered and called."""
# Create chunks for calling multiple tools
functions = [
{
"name": "get_weather",
"arguments": {"location": "NYC"},
"tool_call_id": "call_weather",
},
{
"name": "book_restaurant",
"arguments": {"restaurant": "Tavern", "party_size": 4},
"tool_call_id": "call_restaurant",
},
]
chunks = MockLLMService.create_multiple_function_call_chunks(functions)
llm = MockLLMService(mock_chunks=chunks, chunk_delay=0.001)
# Track calls
calls_made = []
async def mock_get_weather(params: FunctionCallParams):
calls_made.append(("get_weather", params.arguments))
await params.result_callback({"temp": 72, "condition": "sunny"})
async def mock_book_restaurant(params: FunctionCallParams):
calls_made.append(("book_restaurant", params.arguments))
await params.result_callback({"confirmed": True})
llm.register_function("get_weather", mock_get_weather)
llm.register_function("book_restaurant", mock_book_restaurant)
messages = [{"role": "user", "content": "Check weather and book restaurant"}]
context = LLMContext(messages)
pipeline = Pipeline([llm])
await run_test(
pipeline,
frames_to_send=[LLMContextFrame(context)],
expected_down_frames=None,
)
# Verify both handlers were called
assert len(calls_made) == 2
tool_names = [call[0] for call in calls_made]
assert "get_weather" in tool_names
assert "book_restaurant" in tool_names
class TestCustomToolManagerUnit:
"""Unit tests for CustomToolManager class."""
@pytest.mark.asyncio
async def test_get_tool_schemas_returns_correct_format(self):
"""Test that get_tool_schemas returns FunctionSchema objects."""
# Create a mock engine
from pipecat.adapters.schemas.function_schema import FunctionSchema
from api.services.workflow.pipecat_engine import PipecatEngine
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._organization_id = None
mock_engine._get_organization_id = PipecatEngine._get_organization_id.__get__(
mock_engine
)
manager = CustomToolManager(mock_engine)
# Mock the database client
mock_tool = MockToolModel(
tool_uuid="uuid-1",
name="Test Tool",
description="A test tool",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/test",
"parameters": [
{
"name": "param1",
"type": "string",
"description": "Test param",
"required": True,
}
],
},
},
)
with (
patch(
"api.services.workflow.pipecat_engine_custom_tools.db_client"
) as mock_db,
patch(
"api.db:db_client.get_organization_id_by_workflow_run_id",
new_callable=AsyncMock,
return_value=1,
),
):
mock_db.get_tools_by_uuids = AsyncMock(return_value=[mock_tool])
schemas = await manager.get_tool_schemas(["uuid-1"])
assert len(schemas) == 1
schema = schemas[0]
# Schema should be a FunctionSchema object
assert isinstance(schema, FunctionSchema)
# FunctionSchema should have correct attributes
assert schema.name == "test_tool"
assert "param1" in schema.properties
assert schema.properties["param1"]["type"] == "string"
assert "param1" in schema.required
@pytest.mark.asyncio
async def test_register_handlers_creates_working_handler(self):
"""Test that register_handlers creates handlers that can execute tools."""
from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager
# Create a mock engine with a mock LLM
mock_llm = Mock()
registered_handlers = {}
registered_kwargs = {}
def capture_register(name, handler, **kwargs):
registered_handlers[name] = handler
registered_kwargs[name] = kwargs
mock_llm.register_function = capture_register
from api.services.workflow.pipecat_engine import PipecatEngine
mock_engine = Mock()
mock_engine._workflow_run_id = 1
mock_engine._call_context_vars = {}
mock_engine._organization_id = None
mock_engine._get_organization_id = PipecatEngine._get_organization_id.__get__(
mock_engine
)
mock_engine.llm = mock_llm
manager = CustomToolManager(mock_engine)
mock_tool = MockToolModel(
tool_uuid="uuid-1",
name="API Call",
description="Make an API call",
category="http_api",
definition={
"schema_version": 1,
"type": "http_api",
"config": {
"method": "POST",
"url": "https://api.example.com/call",
"parameters": [],
},
},
)
with (
patch(
"api.services.workflow.pipecat_engine_custom_tools.db_client"
) as mock_db,
patch(
"api.db:db_client.get_organization_id_by_workflow_run_id",
new_callable=AsyncMock,
return_value=1,
),
):
mock_db.get_tools_by_uuids = AsyncMock(return_value=[mock_tool])
await manager.register_handlers(["uuid-1"])
# Verify handler was registered
assert "api_call" in registered_handlers
assert registered_kwargs["api_call"]["timeout_secs"] == pytest.approx(5)
# Now test that the handler works
handler = registered_handlers["api_call"]
result_received = None
async def mock_result_callback(result, properties=None):
nonlocal result_received
result_received = result
mock_params = Mock()
mock_params.arguments = {"key": "value"}
mock_params.result_callback = mock_result_callback
with patch(
"api.services.workflow.pipecat_engine_custom_tools.execute_http_tool"
) as mock_execute:
mock_execute.return_value = {
"status": "success",
"data": {"response": "ok"},
}
await handler(mock_params)
# Verify execute was called
mock_execute.assert_called_once()
# Verify result was returned
assert result_received["status"] == "success"
@pytest.mark.asyncio
async def test_transfer_call_renders_destination_from_initial_context(self):
"""Transfer call tools resolve destination templates before provider calls."""
from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager
mock_engine = Mock()
mock_engine._workflow_run_id = 1
mock_engine._call_context_vars = {
"transfer_destination": "+14155550123",
}
mock_engine._gathered_context = {}
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.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": "{{initial_context.transfer_destination}}",
"timeout": 30,
},
},
)
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 = {}
mock_params.result_callback = mock_result_callback
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),
),
):
await handler(mock_params)
provider.transfer_call.assert_awaited_once()
assert provider.transfer_call.await_args.kwargs["destination"] == "+14155550123"
first_context = transfer_manager.store_transfer_context.await_args_list[0].args[
0
]
assert first_context.target_number == "+14155550123"
assert result_received["status"] == "transfer_failed"
@pytest.mark.asyncio
async def test_transfer_call_propagates_provider_destination_error(self):
"""Provider-specific destination failures are returned through the tool result."""
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 = {}
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.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": "provider-specific-destination",
"timeout": 30,
},
},
)
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(
side_effect=Exception("provider rejected destination")
)
transfer_manager = Mock()
transfer_manager.store_transfer_context = AsyncMock()
transfer_manager.remove_transfer_context = AsyncMock()
result_received = None
async def mock_result_callback(result, properties=None):
nonlocal result_received
result_received = result
mock_params = Mock()
mock_params.arguments = {}
mock_params.result_callback = mock_result_callback
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),
),
):
await handler(mock_params)
provider.transfer_call.assert_awaited_once()
assert (
provider.transfer_call.await_args.kwargs["destination"]
== "provider-specific-destination"
)
transfer_manager.remove_transfer_context.assert_awaited_once()
assert result_received == {
"status": "transfer_failed",
"reason": "provider_error",
"message": "Transfer provider failed: provider rejected destination",
}
def _update_llm_context(context, system_message, functions):
"""Inline helper replicating the old update_llm_context for tests."""
tools_schema = ToolsSchema(standard_tools=functions)
previous_interactions = context.messages
if previous_interactions and previous_interactions[0]["role"] == "system":
messages = [system_message] + previous_interactions[1:]
else:
messages = [system_message] + previous_interactions
context.set_messages(messages)
if functions:
context.set_tools(tools_schema)
class TestUpdateLLMContext:
"""Tests for _update_llm_context inline logic."""
def test_replaces_system_message(self):
"""Test that _update_llm_context replaces existing system messages."""
context = LLMContext()
context.set_messages(
[
{"role": "system", "content": "Old system message"},
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there"},
]
)
new_system = {"role": "system", "content": "New system message"}
_update_llm_context(context, new_system, [])
messages = context.messages
# Should have new system message at the start
assert messages[0]["role"] == "system"
assert messages[0]["content"] == "New system message"
# Should preserve user and assistant messages
assert len(messages) == 3
assert messages[1]["role"] == "user"
assert messages[2]["role"] == "assistant"
def test_preserves_conversation_history(self):
"""Test that user/assistant messages are preserved in order."""
context = LLMContext()
context.set_messages(
[
{"role": "system", "content": "Old prompt"},
{"role": "user", "content": "First question"},
{"role": "assistant", "content": "First answer"},
{"role": "user", "content": "Second question"},
{"role": "assistant", "content": "Second answer"},
]
)
new_system = {"role": "system", "content": "New prompt"}
_update_llm_context(context, new_system, [])
messages = context.messages
assert len(messages) == 5
assert messages[1]["content"] == "First question"
assert messages[2]["content"] == "First answer"
assert messages[3]["content"] == "Second question"
assert messages[4]["content"] == "Second answer"
def test_sets_tools_when_functions_provided(self):
"""Test that tools are set on context when functions are provided."""
context = LLMContext()
context.set_messages([{"role": "system", "content": "Old"}])
# Create function schemas
functions = [
get_function_schema("book_appointment", "Book an appointment"),
get_function_schema("cancel_appointment", "Cancel an appointment"),
]
new_system = {"role": "system", "content": "New prompt with tools"}
_update_llm_context(context, new_system, functions)
# Verify tools were set
tools = context.tools
assert tools is not None
assert len(tools.standard_tools) == 2
def test_does_not_set_tools_when_functions_empty(self):
"""Test that tools are not set when functions list is empty."""
context = LLMContext()
context.set_messages([{"role": "system", "content": "Old"}])
new_system = {"role": "system", "content": "New prompt without tools"}
_update_llm_context(context, new_system, [])
# Tools should not be set (or remain None)
# Note: The function only calls set_tools if functions is truthy
# So we verify the context state is as expected
messages = context.messages
assert len(messages) == 1
assert messages[0]["content"] == "New prompt without tools"
def test_works_with_empty_context(self):
"""Test that update works on a fresh context with no messages."""
context = LLMContext()
new_system = {"role": "system", "content": "Initial prompt"}
functions = [get_function_schema("test_func", "A test function")]
_update_llm_context(context, new_system, functions)
messages = context.messages
assert len(messages) == 1
assert messages[0]["role"] == "system"
assert messages[0]["content"] == "Initial prompt"
def test_function_schema_structure(self):
"""Test that get_function_schema creates correct structure."""
schema = get_function_schema(
"search_products",
"Search for products in the catalog",
properties={
"query": {"type": "string", "description": "Search query"},
"limit": {"type": "integer", "description": "Max results"},
},
required=["query"],
)
assert schema.name == "search_products"
assert schema.description == "Search for products in the catalog"
assert "query" in schema.properties
assert "limit" in schema.properties
assert "query" in schema.required
assert "limit" not in schema.required
def test_function_schema_with_no_parameters(self):
"""Test get_function_schema with no properties or required."""
schema = get_function_schema("ping", "Check if service is alive")
assert schema.name == "ping"
assert schema.description == "Check if service is alive"
assert schema.properties == {}
assert schema.required == []