Merge remote-tracking branch 'origin/main' into pr-381

api/services/workflow/dto.py

@@ -244,7 +244,8 @@ class _ToolDocumentRefsMixin(BaseModel):
             "display_name": "Greeting Text",
             "description": (
                 "Text spoken via TTS at the start of the call. Supports "
-                "{{template_variables}}. Leave empty to skip the greeting."
+                "{{template_variables}}. Leave empty to skip the greeting. "
+                "Not supported with realtime (speech-to-speech) models."
             ),
             "display_options": DisplayOptions(show={"greeting_type": ["text"]}),
             "placeholder": "Hi {{first_name}}, this is Sarah from Acme.",

api/services/workflow/mcp_tool_session.py

@@ -79,8 +79,12 @@ class McpToolSession:
         self.available: bool = False
 
     async def start(self) -> None:
-        """Connect, initialize, and cache the tool list. Never raises —
-        on any failure the session is marked unavailable."""
+        """Connect, initialize, and cache the tool list.
+
+        Never raises on a connect failure — a dead/unreachable MCP server
+        leaves the session marked unavailable (``available = False``). Genuine
+        external cancellation, KeyboardInterrupt, and SystemExit are re-raised
+        (see the CancelledError handling below and ``_degrade``)."""
         try:
             params = build_streamable_http_params(
                 url=self._url,

api/services/workflow/pipecat_engine.py

@@ -10,7 +10,7 @@ from pipecat.frames.frames import (
     LLMContextFrame,
     TTSSpeakFrame,
 )
-from pipecat.pipeline.task import PipelineTask
+from pipecat.pipeline.worker import PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.services.settings import LLMSettings
@@ -60,7 +60,7 @@ class PipecatEngine:
     def __init__(
         self,
         *,
-        task: Optional[PipelineTask] = None,
+        task: Optional[PipelineWorker] = None,
         llm: Optional["LLMService"] = None,
         inference_llm: Optional["LLMService"] = None,
         context: Optional[LLMContext] = None,
@@ -851,7 +851,7 @@ class PipecatEngine:
         """
         self.context = context
 
-    def set_task(self, task: PipelineTask) -> None:
+    def set_task(self, task: PipelineWorker) -> None:
         """Set the pipeline task.
 
         This allows setting the task after the engine has been created,
@@ -964,7 +964,15 @@ class PipecatEngine:
                 exc_info=True,
             )
 
-    async def _close_mcp_sessions(self) -> None:
+    async def close_mcp_sessions(self) -> None:
+        """Close all open MCP tool sessions.
+
+        Must run in the same task that ran initialize() (which opened the
+        sessions via _open_mcp_sessions). The MCP client's underlying anyio
+        cancel scopes are task-affine — they must be exited from the task that
+        entered them — so this is invoked from _run_pipeline's finally, not
+        from cleanup() (which runs in a pipecat event-handler task).
+        """
         for tool_uuid, session in list(self._mcp_sessions.items()):
             try:
                 await session.close()
@@ -973,7 +981,14 @@ class PipecatEngine:
         self._mcp_sessions = {}
 
     async def cleanup(self):
-        """Clean up engine resources on disconnect."""
+        """Clean up engine resources on disconnect.
+
+        MCP tool sessions are intentionally NOT closed here — see
+        close_mcp_sessions(). This method runs in a pipecat event-handler task
+        (on_pipeline_finished), a different task than the one that opened the
+        MCP sessions; closing them here raises "Attempted to exit cancel scope
+        in a different task than it was entered in".
+        """
         # Cancel any pending timeout tasks
         if (
             self._user_response_timeout_task
@@ -982,11 +997,5 @@ class PipecatEngine:
             self._user_response_timeout_task.cancel()
 
         # Cancel any in-flight background summarization.
-        # MCP sessions are closed in a finally block so they are guaranteed to
-        # run even if the summarization cleanup raises an exception.
-        try:
-            if self._context_summarization_manager:
-                await self._context_summarization_manager.cleanup()
-        finally:
-            # Close any open MCP tool sessions
-            await self._close_mcp_sessions()
+        if self._context_summarization_manager:
+            await self._context_summarization_manager.cleanup()

api/services/workflow/pipecat_engine_callbacks.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 """Callback factory helpers for :pyclass:`~api.services.workflow.pipecat_engine.PipecatEngine`.
 
 Each helper takes a :class:`PipecatEngine` instance and returns an async
@@ -10,6 +8,8 @@ encapsulating the callback implementations here for easier maintenance and
 unit-testing.
 """
 
+from __future__ import annotations
+
 import re
 from typing import TYPE_CHECKING
 
@@ -73,11 +73,14 @@ def create_user_idle_handler(engine: "PipecatEngine") -> UserIdleHandler:
 
 
 def create_max_duration_callback(engine: "PipecatEngine"):
-    """Return a callback that ends the task when the max call duration is exceeded."""
+    """Return a callback that cancels the task when the hard call limit is exceeded."""
 
     async def handle_max_duration():
         logger.debug("Max call duration exceeded. Terminating call")
-        await engine.end_call_with_reason(EndTaskReason.CALL_DURATION_EXCEEDED.value)
+        await engine.end_call_with_reason(
+            EndTaskReason.CALL_DURATION_EXCEEDED.value,
+            abort_immediately=True,
+        )
 
     return handle_max_duration
 

api/services/workflow/text_chat_runner.py

@@ -22,7 +22,6 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -45,6 +44,10 @@ from api.services.pipecat.tracing_config import (
     build_remote_parent_context,
     get_trace_url,
 )
+from api.services.pipecat.worker_runner import (
+    run_pipeline_worker,
+    wait_for_pipeline_worker_started,
+)
 from api.services.workflow.dto import ReactFlowDTO
 from api.services.workflow.pipecat_engine import PipecatEngine
 from api.services.workflow.workflow_graph import WorkflowGraph
@@ -534,8 +537,7 @@ async def execute_text_chat_pending_turn(
         conversation_type="text",
         additional_span_attributes=trace_span_attributes,
     )
-    runner = PipelineRunner(handle_sigint=False, handle_sigterm=False)
-    runner_task = asyncio.create_task(runner.run(task))
+    runner_task = asyncio.create_task(run_pipeline_worker(task))
 
     engine.set_task(task)
     engine.set_audio_config(audio_config)
@@ -548,7 +550,7 @@ async def execute_text_chat_pending_turn(
     )
 
     try:
-        await asyncio.wait_for(task._pipeline_start_event.wait(), timeout=5.0)
+        await wait_for_pipeline_worker_started(task, timeout=5.0, run_task=runner_task)
 
         await engine.initialize()
 

api/services/workflow/tools/custom_tool.py

@@ -16,6 +16,8 @@ TYPE_MAP = {
     "string": "string",
     "number": "number",
     "boolean": "boolean",
+    "object": "object",
+    "array": "array",
 }
 
 
@@ -45,10 +47,24 @@ def tool_to_function_schema(tool: Any) -> Dict[str, Any]:
         if not param_name:
             continue
 
-        properties[param_name] = {
-            "type": TYPE_MAP.get(param_type, "string"),
-            "description": param_desc,
-        }
+        schema_type = TYPE_MAP.get(param_type, "string")
+        if schema_type == "object":
+            properties[param_name] = {
+                "type": "object",
+                "additionalProperties": True,
+                "description": param_desc,
+            }
+        elif schema_type == "array":
+            properties[param_name] = {
+                "type": "array",
+                "items": {},
+                "description": param_desc,
+            }
+        else:
+            properties[param_name] = {
+                "type": schema_type,
+                "description": param_desc,
+            }
 
         if param_required:
             required.append(param_name)
@@ -127,6 +143,26 @@ def _coerce_parameter_value(value: Any, param_type: str) -> Any:
 
         raise ValueError(f"Cannot convert '{value}' to boolean")
 
+    if param_type == "object":
+        if isinstance(value, str):
+            try:
+                value = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"Cannot convert '{value}' to object") from exc
+        if isinstance(value, dict):
+            return value
+        raise ValueError(f"Cannot convert '{value}' to object")
+
+    if param_type == "array":
+        if isinstance(value, str):
+            try:
+                value = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"Cannot convert '{value}' to array") from exc
+        if isinstance(value, list):
+            return value
+        raise ValueError(f"Cannot convert '{value}' to array")
+
     return value
 
 

api/services/workflow/tools/mcp_tool.py

@@ -4,70 +4,27 @@ LLM-function-name namespacing. No I/O, no MCP protocol here."""
 from __future__ import annotations
 
 import re
-from typing import Any, Dict, Literal, Optional
+from typing import Any, Dict
 
-from pydantic import BaseModel, Field, ValidationError, field_validator
+from pydantic import ValidationError
 
-DEFAULT_TIMEOUT_SECS = 30
-DEFAULT_SSE_READ_TIMEOUT_SECS = 300
+from api.schemas.tool import (
+    DEFAULT_MCP_SSE_READ_TIMEOUT_SECS,
+    DEFAULT_MCP_TIMEOUT_SECS,
+    McpToolDefinition,
+)
+from api.schemas.tool import (
+    McpToolConfig as McpToolConfig,
+)
+
+DEFAULT_TIMEOUT_SECS = DEFAULT_MCP_TIMEOUT_SECS
+DEFAULT_SSE_READ_TIMEOUT_SECS = DEFAULT_MCP_SSE_READ_TIMEOUT_SECS
 
 
 class McpDefinitionError(ValueError):
     """Raised when an MCP tool definition is structurally invalid."""
 
 
-class McpToolConfig(BaseModel):
-    """Configuration for an MCP tool definition."""
-
-    transport: Literal["streamable_http"] = Field(
-        default="streamable_http", description="MCP transport protocol"
-    )
-    url: str = Field(description="MCP server URL (must be http:// or https://)")
-    credential_uuid: Optional[str] = Field(
-        default=None, description="Reference to ExternalCredentialModel for auth"
-    )
-    tools_filter: list[str] = Field(
-        default_factory=list,
-        description="Allowlist of MCP tool names to expose (empty = all tools)",
-    )
-    timeout_secs: int = Field(
-        default=DEFAULT_TIMEOUT_SECS, description="Connection timeout in seconds"
-    )
-    sse_read_timeout_secs: int = Field(
-        default=DEFAULT_SSE_READ_TIMEOUT_SECS,
-        description="SSE read timeout in seconds",
-    )
-    discovered_tools: list[dict[str, Any]] = Field(
-        default_factory=list,
-        description=(
-            "Server-managed cache of the MCP server's tool catalog "
-            "[{name, description}]. Populated best-effort by the backend."
-        ),
-    )
-
-    @field_validator("url")
-    @classmethod
-    def validate_url(cls, v: str) -> str:
-        if not isinstance(v, str) or not v.startswith(("http://", "https://")):
-            raise ValueError("config.url must be an http(s) URL")
-        return v
-
-    @field_validator("tools_filter")
-    @classmethod
-    def validate_tools_filter(cls, v: list[str]) -> list[str]:
-        if not all(isinstance(tool_name, str) for tool_name in v):
-            raise ValueError("config.tools_filter must be a list of strings")
-        return v
-
-
-class McpToolDefinition(BaseModel):
-    """Persisted MCP tool definition."""
-
-    schema_version: int = Field(default=1, description="Schema version")
-    type: Literal["mcp"] = Field(description="Tool type")
-    config: McpToolConfig = Field(description="MCP server configuration")
-
-
 def _format_validation_error(error: ValidationError) -> str:
     parts: list[str] = []
     for item in error.errors():