diff --git a/FUTURE_TOOLS.md b/FUTURE_TOOLS.md new file mode 100644 index 000000000..b3f8ed019 --- /dev/null +++ b/FUTURE_TOOLS.md @@ -0,0 +1,87 @@ +# Future Tools Roadmap + +Tools planned for future implementation in Dograh. + +--- + +## 1. `get_current_time` / `convert_time` + +**Status:** Backend fully implemented in `api/services/workflow/tools/timezone.py` — just needs wiring up. + +**What it does:** +- `get_current_time(timezone)` — Agent tells the caller the current time in any timezone +- `convert_time(source_timezone, time, target_timezone)` — Agent converts a time between two timezones + +**Use case:** +Scheduling agents, appointment bots, international call centers. +> *"What time is it in New York right now?"* → Agent calls `get_current_time("America/New_York")` and responds accurately. + +**What needs to be done:** +- Add `TIME = "time"` to `ToolCategory` in `api/enums.py` +- Add a new Alembic migration for the enum +- Register `get_current_time` and `convert_time` handlers in `pipecat_engine_custom_tools.py` +- Add the tool category config in `ui/src/app/tools/config.tsx` +- Add `WaitToolDefinition` equivalent for time tool in `ui/src/client/types.gen.ts` + +--- + +## 2. `play_audio_clip` + +**Status:** Not implemented. Internal `play_audio` utility exists in `api/services/pipecat/audio_playback.py` but is not exposed as an LLM-callable tool. + +**What it does:** +- LLM dynamically decides mid-call to play a specific pre-recorded audio clip +- Different from current usage where audio is always pre-configured and fixed (e.g. hold music during wait, goodbye on end call) + +**Use case:** +Regulated industries — finance, insurance, healthcare — where specific legal disclaimers must be read **verbatim** from a pre-recorded clip, triggered based on conversation context. +> *User asks about refund policy* → LLM calls `play_audio_clip(clip_id="refund_disclaimer")` → exact approved recording plays. + +Also useful for: brand voice recordings, non-verbal audio (beeps/tones), multilingual clips mid-call. + +**What needs to be done:** +- Add `AUDIO_CLIP = "audio_clip"` to `ToolCategory` in `api/enums.py` +- Add a new Alembic migration for the enum +- Implement the tool handler in `pipecat_engine_custom_tools.py` using existing `play_audio()` +- UI config in `config.tsx` with a clip picker (select from existing recordings) +- The clip library would use the existing recordings/uploads system in Dograh + +--- + +## 3. `send_dtmf` + +**Status:** Not implemented. Dograh has inbound DTMF (receive user keypresses) but NOT outbound DTMF (agent sends keypresses). + +**What it does:** +Agent programmatically sends DTMF tones through an outbound call. + +**Use case:** +Agent makes an outbound call and the other end is an IVR system. +> *Clinic IVR: "Press 1 for appointments, press 2 for billing"* +> Agent calls `send_dtmf(digit="1")` → tone is sent → IVR proceeds to appointments menu. + +**What needs to be done:** +- Implement per-provider: Twilio (`calls.update(dtmf=...)`) , Plivo (`send_digits`), Telnyx, etc. +- Each telephony provider has its own API for sending DTMF tones mid-call +- Register handler in `pipecat_engine_custom_tools.py` +- Expose in UI as a built-in tool + +--- + +## 4. `send_sms` + +**Status:** Not implemented. Telephony integrations for Twilio/Plivo exist in Dograh, but there is no SMS sending capability exposed. + +**What it does:** +Agent programmatically sends a text message to the caller (or another number) during or after the call. + +**Use case:** +Agent finishes an action and sends confirmation details. +> *Agent: "I've booked your appointment. I'll text you the confirmation details right now."* +> Agent calls `send_sms(message="Your appointment is confirmed for tomorrow at 3pm.")` → SMS is delivered via the telephony provider. + +**What needs to be done:** +- Implement SMS sending for each provider (Twilio Programmable SMS, Plivo SMS API). +- Register `send_sms` handler in `pipecat_engine_custom_tools.py`. +- Expose in UI as a built-in tool. (Note: May require handling regulatory constraints like DLT in India). + diff --git a/GIT_WORKFLOW.md b/GIT_WORKFLOW.md new file mode 100644 index 000000000..df4fa525d --- /dev/null +++ b/GIT_WORKFLOW.md @@ -0,0 +1,37 @@ +# Dograh Git Workflow (Fork-and-Pull) + +This document outlines the gold standard Git workflow for contributing to the Dograh repository, using the Fork-and-Pull model. + +## Why this is the best approach: +1. **Pristine Local `main` Branch**: By never committing directly to your local `main` branch, you guarantee it will never drift or get into messy conflict states with `upstream/main`. Your local `main` essentially just acts as a clean mirror of the official Dograh repository. *(Note: Your local `main` is also what is deployed to your server, so keeping it clean and stable is critical!)* +2. **Feature Isolation**: By creating a new branch (like `wait-tool`) off your clean `main` and pushing it to your personal fork (`arnofrxdd`), you isolate your work. If you make a mistake, you can always just delete the branch and start over from your clean `main` without losing any official repo code. +3. **Painless Updates**: When you pull updates from `upstream/main` every 1-3 days, you never have to deal with massive merge conflicts. Your local `main` fast-forwards instantly, and any new branches you create are built on top of the latest, most stable code. + +## The Ideal Development Cycle: + +1. **Go to your clean local mirror:** + ```bash + git checkout main + ``` + +2. **Sync your local main with Dograh's latest code:** + ```bash + git pull upstream main + ``` + +3. **Create an isolated space for your new feature/fix:** + ```bash + git checkout -b new-feature + ``` + +4. **Write code...** (Make your commits here) + +5. **Push to your fork and open a PR:** + ```bash + git push arnofrxdd new-feature + ``` + +6. **Go back to safety and repeat!** + ```bash + git checkout main + ``` diff --git a/api/alembic/versions/ceeaf3c37b2f_add_wait_to_tool_category.py b/api/alembic/versions/ceeaf3c37b2f_add_wait_to_tool_category.py new file mode 100644 index 000000000..2092ba581 --- /dev/null +++ b/api/alembic/versions/ceeaf3c37b2f_add_wait_to_tool_category.py @@ -0,0 +1,64 @@ +"""add wait to tool category + +Revision ID: ceeaf3c37b2f +Revises: gg11dd223344 +Create Date: 2026-07-22 12:42:18.521312 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +from alembic_postgresql_enum import TableReference + +# revision identifiers, used by Alembic. +revision: str = 'ceeaf3c37b2f' +down_revision: Union[str, None] = '00b0201ad918' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.sync_enum_values( + enum_schema="public", + enum_name="tool_category", + new_values=[ + "http_api", + "end_call", + "transfer_call", + "calculator", + "native", + "integration", + "mcp", + "wait", + ], + affected_columns=[ + TableReference( + table_schema="public", table_name="tools", column_name="category" + ) + ], + enum_values_to_rename=[], + ) + +def downgrade() -> None: + op.sync_enum_values( + enum_schema="public", + enum_name="tool_category", + new_values=[ + "http_api", + "end_call", + "transfer_call", + "calculator", + "native", + "integration", + "mcp", + ], + affected_columns=[ + TableReference( + table_schema="public", table_name="tools", column_name="category" + ) + ], + enum_values_to_rename=[], + ) diff --git a/api/assets/wait_music_16000.wav b/api/assets/wait_music_16000.wav new file mode 100644 index 000000000..2a64d125c Binary files /dev/null and b/api/assets/wait_music_16000.wav differ diff --git a/api/assets/wait_music_8000.wav b/api/assets/wait_music_8000.wav new file mode 100644 index 000000000..fe6b83a57 Binary files /dev/null and b/api/assets/wait_music_8000.wav differ diff --git a/api/enums.py b/api/enums.py index 4dda8441c..c87906604 100644 --- a/api/enums.py +++ b/api/enums.py @@ -173,6 +173,7 @@ class ToolCategory(Enum): NATIVE = "native" # Built-in integrations (future: dtmf_input) INTEGRATION = "integration" # Third-party integrations (future: Google Calendar, Salesforce, etc.) MCP = "mcp" # Customer-provided MCP server exposing a tool catalog + WAIT = "wait" # Built-in wait_for_user tool class ToolStatus(Enum): diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 87937b276..d065a394b 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -25,6 +25,7 @@ "end_call", "transfer_call", "calculator", + "wait", "native", "integration", "mcp", @@ -435,6 +436,13 @@ class CalculatorToolDefinition(BaseModel): type: Literal["calculator"] = Field(description="Tool type.") +class WaitToolDefinition(BaseModel): + """Tool definition for Wait tools.""" + + schema_version: int = Field(default=1, description="Schema version.") + type: Literal["wait"] = Field(description="Tool type.") + + class McpToolDefinition(BaseModel): """Persisted MCP tool definition.""" @@ -449,6 +457,7 @@ class McpToolDefinition(BaseModel): EndCallToolDefinition, TransferCallToolDefinition, CalculatorToolDefinition, + WaitToolDefinition, McpToolDefinition, ], Field(discriminator="type"), diff --git a/api/services/pipecat/audio_playback.py b/api/services/pipecat/audio_playback.py index ab0e50416..77fd5bbc9 100644 --- a/api/services/pipecat/audio_playback.py +++ b/api/services/pipecat/audio_playback.py @@ -174,19 +174,32 @@ async def play_audio_loop( duration = num_samples / sample_rate logger.debug(f"Audio loop: playing at {sample_rate}Hz") + + # 500ms chunks to allow near-instant cancellation without music bleeding + chunk_duration_sec = 0.5 + chunk_size = int(sample_rate * chunk_duration_sec * 2) # 2 bytes per sample + try: while not stop_event.is_set(): - frame = OutputAudioRawFrame( - audio=audio_data, - sample_rate=sample_rate, - num_channels=1, - ) - await queue_frame(frame) - try: - await asyncio.wait_for(stop_event.wait(), timeout=duration + 1.5) - break - except asyncio.TimeoutError: - pass + for i in range(0, len(audio_data), chunk_size): + if stop_event.is_set(): + break + + chunk = audio_data[i:i + chunk_size] + frame = OutputAudioRawFrame( + audio=chunk, + sample_rate=sample_rate, + num_channels=1, + ) + await queue_frame(frame) + + chunk_play_time = (len(chunk) // 2) / sample_rate + + try: + await asyncio.wait_for(stop_event.wait(), timeout=chunk_play_time) + break + except asyncio.TimeoutError: + pass except Exception as e: logger.error(f"Audio loop error: {e}") logger.debug("Audio loop: stopped") diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index 3bc711674..71ce0692a 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -28,6 +28,7 @@ from api.services.telephony.factory import get_telephony_provider_for_run from api.services.telephony.transfer_event_protocol import TransferContext from api.services.workflow.tools.calculator import get_calculator_tools, safe_calculator +from api.services.workflow.tools.wait import get_wait_tools from api.services.workflow.tools.custom_tool import ( execute_http_tool, tool_to_function_schema, @@ -172,6 +173,21 @@ async def get_tool_schemas( schemas: list[FunctionSchema] = [] for tool in tools: + if tool.category == "wait": + self._register_wait_handler() + logger.debug(f"Registered wait tool handler (tool_uuid: {tool.tool_uuid})") + for tool_def in get_wait_tools(): + func = tool_def["function"] + schemas.append( + get_function_schema( + func["name"], + func["description"], + properties=func["parameters"]["properties"], + required=func["parameters"]["required"], + ) + ) + continue + if tool.category == ToolCategory.CALCULATOR.value: # Built-in calculator: return pre-defined schemas for tool_def in get_calculator_tools(): @@ -250,6 +266,11 @@ async def register_handlers( tools = await db_client.get_tools_by_uuids(tool_uuids, organization_id) for tool in tools: + if tool.category == "wait": + self._register_wait_handler() + logger.debug(f"Registered wait tool handler (tool_uuid: {tool.tool_uuid})") + continue + if tool.category == ToolCategory.CALCULATOR.value: self._register_calculator_handler() logger.debug( @@ -359,6 +380,186 @@ def _transfer_handler_timeout_secs(self, tool: Any) -> float: return float(transfer_timeout) + resolver_timeout + 15.0 + def _register_wait_handler(self) -> None: + """Register the built-in wait function with the LLM.""" + + async def wait_func(function_call_params: FunctionCallParams) -> None: + logger.info("LLM Function Call EXECUTED: wait_for_user") + logger.info(f"Arguments: {function_call_params.arguments}") + + try: + seconds_arg = function_call_params.arguments.get("seconds", 60) + try: + seconds = int(seconds_arg) + if seconds < 0: + raise ValueError("Seconds cannot be negative.") + except (ValueError, TypeError): + await function_call_params.result_callback({"error": f"Invalid 'seconds' parameter: {seconds_arg}. Must be a positive integer."}) + return + + # Use a hard cap of 300 seconds for the built-in wait tool + max_wait = 300 + seconds = min(max(seconds, 1), max_wait) + + message = function_call_params.arguments.get("message") + if message: + from pipecat.frames.frames import TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame + from pipecat.observers.base_observer import BaseObserver as _BaseObserver, FramePushed as _FramePushed + + logger.info(f"Playing wait acknowledgment message: {message}") + + # Register a temporary observer BEFORE queuing the TTS so we + # don't miss TTSStartedFrame if the pipeline is very fast. + # We require TTSStartedFrame before accepting TTSStoppedFrame to + # guard against stale TTSStoppedFrames from the previous response. + tts_done_event = asyncio.Event() + + class _TTSDoneObserver(_BaseObserver): + def __init__(self): + super().__init__() + self._seen_started = False + + async def on_push_frame(self, data: _FramePushed): + if isinstance(data.frame, TTSStartedFrame): + self._seen_started = True + elif isinstance(data.frame, TTSStoppedFrame) and self._seen_started: + logger.info("Wait tool: acknowledgment TTS finished (TTSStoppedFrame received).") + tts_done_event.set() + + tts_done_observer = _TTSDoneObserver() + _task = self._engine.task + if _task: + _task.add_observer(tts_done_observer) + + await self._engine.task.queue_frame( + TTSSpeakFrame( + message, + append_to_context=False, + persist_to_logs=True, + ) + ) + + # Wait for TTS to actually finish, with a generous fallback timeout. + try: + await asyncio.wait_for(tts_done_event.wait(), timeout=15.0) + except asyncio.TimeoutError: + logger.warning("Wait tool: TTS done event timed out after 15s ÔÇö proceeding anyway.") + + if _task: + try: + await _task.remove_observer(tts_done_observer) + except Exception as e: + logger.warning(f"Could not remove TTS done observer: {e}") + + logger.info(f"Pausing for {seconds} seconds...") + + stop_event = asyncio.Event() + hold_music_task = None + + if seconds >= 5 and self._engine._audio_config and self._engine._transport_output: + from api.constants import APP_ROOT_DIR + sample_rate = self._engine._audio_config.pipeline_sample_rate + audio_file = str(APP_ROOT_DIR / "assets" / f"wait_music_{sample_rate}.wav") + + hold_music_task = asyncio.create_task( + play_audio_loop( + stop_event=stop_event, + sample_rate=sample_rate, + queue_frame=self._engine._transport_output.queue_frame, + audio_file=audio_file, + ) + ) + + user_speech_event = asyncio.Event() + # Issue 2 fix: capture the actual transcript text so we can include it + # in the tool result. Without this, the TranscriptionFrame is muted by + # FunctionCallUserMuteStrategy and the LLM never sees what the user said, + # forcing the user to repeat themselves. + import time + user_speech_text: list[str] = [] # mutable container for closure capture + + # We use a pipeline observer (not the mute-filtered aggregator) + # because FunctionCallUserMuteStrategy suppresses all + # TranscriptionFrames and UserStartedSpeakingFrames while a + # tool is executing. Pipeline observers receive frames *before* + # the mute strategy suppresses them. + from pipecat.observers.base_observer import BaseObserver, FramePushed + from pipecat.frames.frames import TranscriptionFrame + + class _WaitInterruptObserver(BaseObserver): + async def on_push_frame(self, data: FramePushed): + if isinstance(data.frame, TranscriptionFrame) and data.frame.text.strip(): + text = data.frame.text.strip() + logger.info(f"Wait tool observer: transcription received: '{text}', signalling interrupt.") + # Capture the text (Issue 2 fix: include in tool result) + if not user_speech_text: + user_speech_text.append(text) + user_speech_event.set() + + wait_observer = _WaitInterruptObserver() + task = self._engine.task + if task: + task.add_observer(wait_observer) + + elapsed = 0.0 + try: + # Fix #1: Never wait less than 15 seconds to avoid overlapping TTS + # acknowledgment and immediate "are you still there" follow-up. + effective_seconds = min(max(float(seconds), 15.0), float(max_wait)) + + while round(elapsed, 1) < effective_seconds: + if self._engine.is_call_disposed(): + break + + if user_speech_event.is_set(): + logger.info("Wait interrupted by user speech.") + break + + await asyncio.sleep(0.1) + elapsed += 0.1 + finally: + stop_event.set() + if hold_music_task: + hold_music_task.cancel() + try: + await hold_music_task + except asyncio.CancelledError: + pass + + if task: + try: + await task.remove_observer(wait_observer) + except Exception as e: + logger.warning(f"Could not remove wait observer: {e}") + + logger.info(f"Wait completed after {elapsed} seconds (requested {seconds}).") + + if function_call_params and function_call_params.result_callback: + clean_elapsed = round(elapsed, 1) + msg = f"Wait finished after {clean_elapsed} seconds. " + if user_speech_event.is_set(): + # Issue 2 fix: The TranscriptionFrame was muted by + # FunctionCallUserMuteStrategy so the LLM context never received + # it. We inject the captured text directly into the tool result + # so the LLM can respond to the user's exact words. + captured = user_speech_text[0] if user_speech_text else None + if captured: + msg += f'The user has returned and said: "{captured}". Respond directly to what they said. Do NOT thank them for waiting, because YOU were the one waiting for THEM.' + else: + msg += "The user has returned and spoken. Please respond naturally to what they just said. Do NOT thank them for waiting, because YOU were the one waiting for THEM." + else: + msg += "The time is up. Please ask the user if they are still there and if they need further assistance. Do NOT thank them for waiting, because YOU were the one waiting for THEM." + + await function_call_params.result_callback( + {"status": "success", "message": msg} + ) + except Exception as e: + logger.error(f"Wait tool error: {e}") + await function_call_params.result_callback({"error": str(e)}) + + # Register with a large timeout to prevent the LLM caller from timing out the wait. + self._engine.llm.register_function("wait_for_user", wait_func, timeout_secs=330.0) + def _register_calculator_handler(self) -> None: """Register the built-in calculator function with the LLM.""" diff --git a/api/services/workflow/tools/wait.py b/api/services/workflow/tools/wait.py new file mode 100644 index 000000000..ae7d2b284 --- /dev/null +++ b/api/services/workflow/tools/wait.py @@ -0,0 +1,38 @@ +from typing import Any, Dict + +def get_wait_tools() -> list[Dict[str, Any]]: + """Get wait tool definitions for LLM function calling.""" + return [ + { + "type": "function", + "function": { + "name": "wait_for_user", + "description": ( + "Wait for the user to return. Use this when the user explicitly asks you to wait or hold on. " + "You MUST specify the duration in seconds to wait. " + "If the user does not specify a duration, default to 60 seconds. " + "NOTE: The minimum wait time is 15 seconds. If the user says 'give me a second' or 'wait a sec', " + "they mean at least 20 seconds. Do not use this tool for very short pauses (< 15s). " + "The absolute maximum wait time is 300 seconds (5 minutes). " + "IMPORTANT: If the user requests a wait time longer than 5 minutes (e.g., an hour), " + "DO NOT silently substitute a smaller value. Instead, do not call this tool. " + "Explain that you can only hold for a maximum of 5 minutes, and ask if they would like you to wait 5 minutes." + ), + "parameters": { + "type": "object", + "properties": { + "seconds": { + "type": "integer", + "description": "The number of seconds to wait. Defaults to 60 if not specified.", + "maximum": 300 + }, + "message": { + "type": "string", + "description": "A short conversational acknowledgment to speak to the user before waiting (e.g. 'Sure, I will wait.').", + } + }, + "required": ["message"], + }, + }, + } + ] diff --git a/ui/src/app/tools/[toolUuid]/page.tsx b/ui/src/app/tools/[toolUuid]/page.tsx index fb919e63a..0162b295d 100644 --- a/ui/src/app/tools/[toolUuid]/page.tsx +++ b/ui/src/app/tools/[toolUuid]/page.tsx @@ -390,7 +390,7 @@ export default function ToolDetailPage() { const normalizedTransferDestination = transferDestination.trim(); // Validation based on tool type - if (tool.category === "calculator") { + if (tool.category === "calculator" || tool.category === "wait") { // No validation needed for built-in tools } else if (tool.category === "transfer_call") { if (transferDestinationSource === "static" && !normalizedTransferDestination) { @@ -512,6 +512,15 @@ export default function ToolDetailPage() { type: "calculator", }, }; + } else if (tool.category === "wait") { + // Built-in tool - only name/description, no config + requestBody = { + name, + description: description || undefined, + definition: { + type: "wait", + } as any, + }; } else if (tool.category === "end_call") { // Build end call request body requestBody = { @@ -786,7 +795,7 @@ const data = await response.json();`; const isEndCallTool = tool.category === "end_call"; const isTransferCallTool = tool.category === "transfer_call"; - const isBuiltinTool = tool.category === "calculator"; + const isBuiltinTool = tool.category === "calculator" || tool.category === "wait"; const isMcpTool = tool.category === "mcp"; const isHttpApiTool = tool.category === "http_api"; const hasUnsavedHttpChanges = @@ -870,8 +879,8 @@ const data = await response.json();`; onNameChange={setName} description={description} onDescriptionChange={setDescription} - title="Calculator Configuration" - subtitle="Built-in calculator for arithmetic operations. No additional configuration needed." + title={tool.category === "wait" ? "Wait Tool" : "Calculator Tool"} + subtitle={tool.category === "wait" ? "A built-in tool that forces the agent to wait for a specified duration." : "A built-in calculator for performing arithmetic operations during the conversation."} /> ) : isEndCallTool ? (