Merge pull request #4371 from Stoic-Angel/feat-global-context

Add a global context for tool calls: tool_resources

changelog/4371.added.md

@@ -0,0 +1 @@
+- Added `tool_resources` to `PipelineTask` and `FunctionCallParams`. Pass an application-defined object (DB handles, clients, state, etc.) to `PipelineTask(..., tool_resources=...)` and access it from any tool handler via `params.tool_resources`. Passed by reference; the caller retains their handle and can read mutations after the task finishes. Resolves #4256.

src/pipecat/pipeline/task.py

@@ -207,6 +207,7 @@ class PipelineTask(BasePipelineTask):
         rtvi_processor: RTVIProcessor | None = None,
         rtvi_observer_params: RTVIObserverParams | None = None,
         task_manager: BaseTaskManager | None = None,
+        tool_resources: Any = None,
     ):
         """Initialize the PipelineTask.
 
@@ -234,6 +235,11 @@ class PipelineTask(BasePipelineTask):
             rtvi_observer_params: The RTVI observer parameter to use if RTVI is enabled.
             rtvi_processor: The RTVI processor to add if RTVI is enabled.
             task_manager: Optional task manager for handling asyncio tasks.
+            tool_resources: Optional application-defined bag of resources (DB handles,
+                clients, state, etc.) passed by reference to every tool handler via
+                ``FunctionCallParams.tool_resources``. The framework never copies or
+                clears this object; the caller retains their handle and can read any
+                mutations after the task finishes.
         """
         super().__init__()
         self._params = params or PipelineParams()
@@ -246,6 +252,7 @@ class PipelineTask(BasePipelineTask):
         self._enable_tracing = enable_tracing and is_tracing_available()
         self._enable_turn_tracking = enable_turn_tracking
         self._idle_timeout_secs = idle_timeout_secs
+        self._tool_resources = tool_resources
         observers = observers or []
         self._turn_tracking_observer: TurnTrackingObserver | None = None
         self._user_bot_latency_observer: UserBotLatencyObserver | None = None
@@ -723,6 +730,7 @@ class PipelineTask(BasePipelineTask):
             clock=self._clock,
             task_manager=self._task_manager,
             observer=self._observer,
+            tool_resources=self._tool_resources,
         )
         await self._pipeline.setup(setup)
 

src/pipecat/processors/frame_processor.py

@@ -71,11 +71,14 @@ class FrameProcessorSetup:
         clock: The clock instance for timing operations.
         task_manager: The task manager for handling async operations.
         observer: Optional observer for monitoring frame processing events.
+        tool_resources: Application-defined resources shared with processors
+            for this pipeline run.
     """
 
     clock: BaseClock
     task_manager: BaseTaskManager
     observer: BaseObserver | None = None
+    tool_resources: Any = None
 
 
 class FrameProcessorQueue(asyncio.PriorityQueue):

src/pipecat/services/llm_service.py

@@ -51,7 +51,7 @@ from pipecat.processors.aggregators.llm_context import (
     LLMContext,
     LLMSpecificMessage,
 )
-from pipecat.processors.frame_processor import FrameDirection
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
 from pipecat.services.ai_service import AIService
 from pipecat.services.settings import LLMSettings, assert_given
 from pipecat.services.websocket_service import WebsocketService
@@ -107,6 +107,9 @@ class FunctionCallParams:
             For async function calls (``cancel_on_interruption=False``), call
             it with ``properties=FunctionCallResultProperties(is_final=False)``
             to push intermediate updates before the final result.
+        tool_resources: Application-defined bag of resources (DB handles, clients,
+            state, etc.) shared across tool calls for the pipeline session. Set
+            via ``PipelineTask(..., tool_resources=...)`` and passed by reference.
     """
 
     function_name: str
@@ -115,6 +118,7 @@ class FunctionCallParams:
     llm: LLMService
     context: LLMContext
     result_callback: FunctionCallResultCallback
+    tool_resources: Any = None
 
 
 @dataclass
@@ -252,6 +256,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         self._sequential_runner_task: asyncio.Task | None = None
         self._skip_tts: bool | None = None
         self._summary_task: asyncio.Task | None = None
+        self._tool_resources: Any = None
 
         self._register_event_handler("on_function_calls_started")
         self._register_event_handler("on_function_calls_cancelled")
@@ -298,6 +303,15 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         """
         raise NotImplementedError(f"run_inference() not supported by {self.__class__.__name__}")
 
+    async def setup(self, setup: FrameProcessorSetup):
+        """Set up the LLM service.
+
+        Args:
+            setup: The frame processor setup data.
+        """
+        await super().setup(setup)
+        self._tool_resources = setup.tool_resources
+
     async def start(self, frame: StartFrame):
         """Start the LLM service.
 
@@ -871,6 +885,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
                         llm=self,
                         context=runner_item.context,
                         result_callback=function_call_result_callback,
+                        tool_resources=self._tool_resources,
                     ),
                 )
             else:
@@ -882,6 +897,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
                     llm=self,
                     context=runner_item.context,
                     result_callback=function_call_result_callback,
+                    tool_resources=self._tool_resources,
                 )
                 await item.handler(params)
         except Exception as e:

tests/test_tool_resources.py

@@ -0,0 +1,140 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import unittest
+from dataclasses import dataclass, field
+from typing import Any
+from unittest.mock import AsyncMock
+
+from pipecat.adapters.schemas.direct_function import DirectFunctionWrapper
+from pipecat.clocks.system_clock import SystemClock
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.frame_processor import FrameProcessorSetup
+from pipecat.services.llm_service import (
+    FunctionCallParams,
+    FunctionCallRegistryItem,
+    FunctionCallRunnerItem,
+    LLMService,
+)
+from pipecat.services.settings import LLMSettings
+from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
+
+
+@dataclass
+class _Resources:
+    user_name: str
+    db: dict[str, Any] = field(default_factory=dict)
+
+
+class _MockLLMService(LLMService):
+    def __init__(self, **kwargs):
+        super().__init__(settings=LLMSettings(), **kwargs)
+
+
+class TestFunctionCallParamsToolResources(unittest.TestCase):
+    def test_default_is_none(self):
+        params = FunctionCallParams(
+            function_name="f",
+            tool_call_id="1",
+            arguments={},
+            llm=None,  # type: ignore[arg-type]
+            context=LLMContext(),
+            result_callback=AsyncMock(),
+        )
+        self.assertIsNone(params.tool_resources)
+
+    def test_holds_reference(self):
+        resources = _Resources(user_name="John")
+        params = FunctionCallParams(
+            function_name="f",
+            tool_call_id="1",
+            arguments={},
+            llm=None,  # type: ignore[arg-type]
+            context=LLMContext(),
+            result_callback=AsyncMock(),
+            tool_resources=resources,
+        )
+        self.assertIs(params.tool_resources, resources)
+
+
+class TestLLMServiceCachesToolResources(unittest.IsolatedAsyncioTestCase):
+    async def test_setup_caches_tool_resources(self):
+        service = _MockLLMService()
+        resources = _Resources(user_name="John")
+        task_manager = TaskManager()
+        task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
+
+        await service.setup(
+            FrameProcessorSetup(
+                clock=SystemClock(),
+                task_manager=task_manager,
+                tool_resources=resources,
+            )
+        )
+        await asyncio.sleep(0)
+        await service.cleanup()
+
+        self.assertIs(service._tool_resources, resources)
+
+    async def test_function_call_params_receives_tool_resources(self):
+        service = _MockLLMService()
+        resources = _Resources(user_name="John")
+        service._tool_resources = resources
+
+        captured: dict[str, Any] = {}
+
+        async def handler(params: FunctionCallParams):
+            captured["params"] = params
+            params.tool_resources.db["hit"] = True
+            await params.result_callback({"ok": True})
+
+        service._functions["lookup"] = FunctionCallRegistryItem(
+            function_name="lookup",
+            handler=handler,
+            cancel_on_interruption=True,
+        )
+        service.broadcast_frame = AsyncMock()  # type: ignore[method-assign]
+
+        runner_item = FunctionCallRunnerItem(
+            registry_item=service._functions["lookup"],
+            function_name="lookup",
+            tool_call_id="call-1",
+            arguments={},
+            context=LLMContext(),
+        )
+        await service._run_function_call(runner_item)
+
+        self.assertIs(captured["params"].tool_resources, resources)
+        self.assertTrue(resources.db["hit"])
+
+    async def test_direct_function_params_receives_tool_resources(self):
+        service = _MockLLMService()
+        resources = _Resources(user_name="John")
+        service._tool_resources = resources
+        captured: dict[str, Any] = {}
+
+        async def lookup(params: FunctionCallParams):
+            captured["params"] = params
+
+        wrapper = DirectFunctionWrapper(lookup)
+        service._functions[wrapper.name] = FunctionCallRegistryItem(
+            function_name=wrapper.name,
+            handler=wrapper,
+            cancel_on_interruption=True,
+        )
+        service.broadcast_frame = AsyncMock()  # type: ignore[method-assign]
+
+        runner_item = FunctionCallRunnerItem(
+            registry_item=service._functions[wrapper.name],
+            function_name=wrapper.name,
+            tool_call_id="call-1",
+            arguments={},
+            context=LLMContext(),
+        )
+        await service._run_function_call(runner_item)
+
+        self.assertIs(captured["params"].tool_resources, resources)