diff --git a/examples/tir/tests/test_tir.py b/examples/tir/tests/test_tir.py index d9365240a2..d6c7492afa 100644 --- a/examples/tir/tests/test_tir.py +++ b/examples/tir/tests/test_tir.py @@ -59,6 +59,27 @@ async def test_tool_manager(): await calc_manager.acleanup() +@pytest.mark.asyncio +async def test_tool_manager_error_handling(): + """Test tool manager error handling with real execution""" + python_manager = ToolManager(timeout=10, enabled_tools="python", debug_mode=False) + + try: + # Test ZeroDivisionError + zero_result, zero_status = await python_manager.aexecute_tool_call( + "```python\n1 / 0\n```" + ) + print(f"Zero division result: {zero_result}, status: {zero_status}") + assert zero_status == ToolCallStatus.ERROR + assert ( + "division by zero" in zero_result.lower() + or "zerodivisionerror" in zero_result.lower() + ) + + finally: + await python_manager.acleanup() + + @pytest.mark.asyncio async def test_tir_workflow(): """Test TIR workflow initialization.""" diff --git a/examples/tir/tools/python_tool.py b/examples/tir/tools/python_tool.py index 1847ad9dbd..7852ab14ae 100644 --- a/examples/tir/tools/python_tool.py +++ b/examples/tir/tools/python_tool.py @@ -256,9 +256,14 @@ def execute(self, parameters: dict[str, Any]) -> tuple[str, ToolCallStatus]: try: # Directly call apply to avoid using ProcessPool in async environment - result = self.python_executor.apply(code) - logger.debug(f"Python execution completed: {str(result)[:100]}...") - return str(result), ToolCallStatus.SUCCESS + res, report = self.python_executor.apply(code) + + if report != "Done": + logger.error(f"Error in Python execution: {report}") + return f"Error: {report}", ToolCallStatus.ERROR + + logger.debug(f"Python execution completed: {str(res)[:100]}...") + return str(res), ToolCallStatus.SUCCESS except Exception as e: logger.error(f"Python execution error: {e}") return f"Error: {str(e)}", ToolCallStatus.ERROR