From 108e32eb726605d6103c3d18ef085f8c59ddf8bf Mon Sep 17 00:00:00 2001 From: Aayush Jain Date: Sat, 25 Apr 2026 02:12:40 +0530 Subject: [PATCH 1/2] Add a global context for tool calls - tool_resources, as a parameter to PipelineTask and FrameProcessorSetup --- src/pipecat/pipeline/task.py | 8 ++ src/pipecat/processors/frame_processor.py | 3 + src/pipecat/services/llm_service.py | 18 ++- tests/test_tool_resources.py | 140 ++++++++++++++++++++++ 4 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 tests/test_tool_resources.py diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 08e08ea00..8b1b7dfb6 100644 --- a/src/pipecat/pipeline/task.py +++ b/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) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 92cf920ae..fe5f49b10 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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): diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 2a6e0668e..f0078e269 100644 --- a/src/pipecat/services/llm_service.py +++ b/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 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. @@ -869,6 +883,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): llm=self, context=runner_item.context, result_callback=function_call_result_callback, + tool_resources=self._tool_resources, ), ) else: @@ -880,6 +895,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: diff --git a/tests/test_tool_resources.py b/tests/test_tool_resources.py new file mode 100644 index 000000000..3f5026ddc --- /dev/null +++ b/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) From 65b15a85288ffb4b4133bdd159ad1c9789fdb48d Mon Sep 17 00:00:00 2001 From: Aayush Jain Date: Sat, 25 Apr 2026 02:23:25 +0530 Subject: [PATCH 2/2] add changelog --- changelog/4371.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4371.added.md diff --git a/changelog/4371.added.md b/changelog/4371.added.md new file mode 100644 index 000000000..92c7cc9e1 --- /dev/null +++ b/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.