import asyncio import unittest from models import RuntimeTool from pipecat.frames.frames import ( CancelFrame, EndFrame, InputTransportMessageFrame, OutputTransportMessageUrgentFrame, StopFrame, ) from pipecat.processors.frame_processor import FrameDirection from services.client_tools import ClientToolBroker, ClientToolError from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutor from schemas import ClientToolConfig def client_tool() -> RuntimeTool: return RuntimeTool( id="tool_photo_button", name="显示拍照按钮", function_name="set_photo_button_visible", type="client", definition={ "schema_version": 1, "type": "client", "config": { "parameters": [ { "name": "visible", "type": "boolean", "required": True, } ], "timeout_seconds": 3, "dynamic_variable_assignments": { "photo_button_visible": "visible" }, }, }, ) class FakeClientTools: def __init__(self, result): self.result = result self.calls = [] async def call( self, function_name, arguments, *, timeout_seconds, wait_for_response=True, response_wait_mode="timeout", ): self.calls.append( ( function_name, arguments, timeout_seconds, wait_for_response, response_wait_mode, ) ) return self.result class ClientToolExecutorTests(unittest.IsolatedAsyncioTestCase): def test_response_wait_mode_defaults_and_uses_camel_case(self): legacy = ClientToolConfig() session = ClientToolConfig(responseWaitMode="session") self.assertEqual(legacy.response_wait_mode, "timeout") self.assertEqual(session.response_wait_mode, "session") self.assertEqual( session.model_dump(by_alias=True)["responseWaitMode"], "session", ) def test_fire_and_forget_rejects_result_assignments(self): with self.assertRaisesRegex(ValueError, "不能配置结果变量赋值"): ClientToolConfig( wait_for_response=False, dynamic_variable_assignments={"visible": "visible"}, ) async def test_success_updates_configured_variable(self): store = DynamicVariableStore( {"photo_button_visible": False}, variable_types={"photo_button_visible": "boolean"}, ) port = FakeClientTools({"status": "ok", "data": {"visible": True}}) executor = ToolExecutor(store, client_tools=port) result = await executor.execute(client_tool(), {"visible": True}) self.assertEqual(result["updated_variables"], ["photo_button_visible"]) self.assertIs(store.values["photo_button_visible"], True) self.assertEqual( port.calls, [ ( "set_photo_button_visible", {"visible": True}, 3.0, True, "timeout", ) ], ) async def test_session_wait_mode_is_forwarded(self): tool = client_tool() tool.definition["config"]["response_wait_mode"] = "session" port = FakeClientTools({"status": "ok", "data": {"visible": True}}) executor = ToolExecutor(DynamicVariableStore({}), client_tools=port) await executor.execute(tool, {"visible": True}) self.assertEqual(port.calls[0][-1], "session") async def test_failure_does_not_update_variable(self): store = DynamicVariableStore({"photo_button_visible": False}) executor = ToolExecutor( store, client_tools=FakeClientTools( {"status": "error", "message": "unsupported"} ), ) result = await executor.execute(client_tool(), {"visible": True}) self.assertEqual(result["updated_variables"], []) self.assertIs(store.values["photo_button_visible"], False) class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase): async def test_correlates_result_and_times_out(self): broker = ClientToolBroker() outbound = [] async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): outbound.append((frame, direction)) broker.push_frame = push_frame call = asyncio.create_task( broker.call("set_photo_button_visible", {"visible": True}, timeout_seconds=1) ) await asyncio.sleep(0) message = outbound[0][0].message self.assertIsInstance(outbound[0][0], OutputTransportMessageUrgentFrame) await broker.process_frame( InputTransportMessageFrame( message={ "type": "client-tool-result", "tool_call_id": message["tool_call_id"], "status": "ok", "data": {"visible": True}, } ), FrameDirection.DOWNSTREAM, ) self.assertEqual( await call, {"status": "ok", "data": {"visible": True}}, ) with self.assertRaisesRegex(ClientToolError, "超时"): await broker.call("never_returns", {}, timeout_seconds=0.001) async def test_fire_and_forget_does_not_create_pending_call(self): broker = ClientToolBroker() outbound = [] async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): outbound.append((frame, direction)) broker.push_frame = push_frame result = await broker.call( "open_panel", {"visible": True}, timeout_seconds=1, wait_for_response=False, ) self.assertEqual(result, {"status": "ok", "data": {"dispatched": True}}) self.assertEqual(broker._pending, {}) self.assertFalse(outbound[0][0].message["wait_for_response"]) async def test_session_wait_is_released_by_end_frame(self): broker = ClientToolBroker() async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): pass broker.push_frame = push_frame call = asyncio.create_task( broker.call( "show_message", {}, timeout_seconds=0.001, response_wait_mode="session", ) ) await asyncio.sleep(0.01) self.assertFalse(call.done()) await broker.process_frame(EndFrame(), FrameDirection.DOWNSTREAM) with self.assertRaisesRegex(ClientToolError, "会话已结束"): await call self.assertEqual(broker._pending, {}) async def test_session_wait_is_released_by_cleanup(self): broker = ClientToolBroker() async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): pass broker.push_frame = push_frame call = asyncio.create_task( broker.call( "show_message", {}, timeout_seconds=0.001, response_wait_mode="session", ) ) await asyncio.sleep(0.01) self.assertFalse(call.done()) await broker.cleanup() with self.assertRaisesRegex(ClientToolError, "通道已关闭"): await call self.assertEqual(broker._pending, {}) async def test_call_after_end_frame_fails_without_registering(self): broker = ClientToolBroker() outbound = [] async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): outbound.append(frame) broker.push_frame = push_frame await broker.process_frame(EndFrame(), FrameDirection.DOWNSTREAM) with self.assertRaisesRegex(ClientToolError, "会话已结束"): await broker.call( "show_message", {}, timeout_seconds=1, response_wait_mode="session", ) self.assertEqual(broker._pending, {}) self.assertEqual(len(outbound), 1) async def test_call_after_cleanup_fails_without_registering(self): broker = ClientToolBroker() await broker.cleanup() with self.assertRaisesRegex(ClientToolError, "通道已关闭"): await broker.call( "show_message", {}, timeout_seconds=1, response_wait_mode="session", ) self.assertEqual(broker._pending, {}) async def test_cancel_frame_closes_broker(self): broker = ClientToolBroker() async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): pass broker.push_frame = push_frame await broker.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM) with self.assertRaisesRegex(ClientToolError, "会话已取消"): await broker.call( "show_message", {}, timeout_seconds=1, response_wait_mode="session", ) async def test_stop_frame_releases_pending_but_keeps_broker_reusable(self): broker = ClientToolBroker() outbound = [] async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): outbound.append(frame) broker.push_frame = push_frame stopped_call = asyncio.create_task( broker.call( "show_message", {}, timeout_seconds=1, response_wait_mode="session", ) ) await asyncio.sleep(0) await broker.process_frame(StopFrame(), FrameDirection.DOWNSTREAM) with self.assertRaisesRegex(ClientToolError, "管线已停止"): await stopped_call self.assertEqual(broker._pending, {}) next_call = asyncio.create_task( broker.call( "show_message", {}, timeout_seconds=1, response_wait_mode="session", ) ) await asyncio.sleep(0) message = next( frame.message for frame in reversed(outbound) if isinstance(frame, OutputTransportMessageUrgentFrame) ) await broker.process_frame( InputTransportMessageFrame( message={ "type": "client-tool-result", "tool_call_id": message["tool_call_id"], "status": "ok", "data": {"action": "confirmed"}, } ), FrameDirection.DOWNSTREAM, ) self.assertEqual( await next_call, {"status": "ok", "data": {"action": "confirmed"}}, ) if __name__ == "__main__": unittest.main()