From 61d68a64138c087b97bc27a4463c3d1caf6b39f2 Mon Sep 17 00:00:00 2001 From: bpb02 Date: Thu, 25 Jun 2026 16:05:47 +0200 Subject: [PATCH 1/3] feat(tools): add RunPython tool for isolated Python execution Adds RunPython, a Timbal-native tool that runs LLM-authored Python in an isolated subprocess (uv PEP 723 with venv+pip fallback), with: - Code mode: the script can call exposed Timbal tools via a Unix-socket RPC bridge, plus top-level `await` support so async agent code just works. - Dependency handling: explicit deps + auto-detection from imports, with base-name dedupe so a pinned spec (e.g. `timbal @ file://...`) is not shadowed by a bare auto-detected import. - Hardening (E2B/Monty-inspired): secrets stripped from the child env by default with an `env_passthrough` allowlist, head+tail output truncation, process-group kill on timeout, and code/return-value size limits. - Headline use case validated end-to-end: install timbal inside the sandbox and run an Agent within an Agent (real Anthropic call). Includes ported competitor test cases (pydantic monty / mcp-run-python), hardening and async tests, real integration tests, and a manual validation runner. Registers RunPython in the tools package and ignores RunPython execution artifacts. Co-authored-by: Cursor --- .gitignore | 4 + python/tests/tools/test_run_python.py | 681 +++++++++++++++ .../tests/tools/validate_run_python_agent.py | 116 +++ python/timbal/tools/__init__.py | 3 + python/timbal/tools/run_python.py | 786 ++++++++++++++++++ 5 files changed, 1590 insertions(+) create mode 100644 python/tests/tools/test_run_python.py create mode 100644 python/tests/tools/validate_run_python_agent.py create mode 100644 python/timbal/tools/run_python.py diff --git a/.gitignore b/.gitignore index 8f014b1e..38607a17 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,10 @@ version.zig *.log* .DS_Store +# RunPython tool isolated execution artifacts +.timbal_run_venv/ +.timbal_run_*/ + # State savers artifacts *.jsonl !python/tests/types/files/examples/test.jsonl diff --git a/python/tests/tools/test_run_python.py b/python/tests/tools/test_run_python.py new file mode 100644 index 00000000..d972d869 --- /dev/null +++ b/python/tests/tools/test_run_python.py @@ -0,0 +1,681 @@ +"""Tests for RunPython tool. + +Cases ported from public competitor test suites: +- pydantic/monty: crates/monty-python/tests/test_basic.py, test_exceptions.py, + test_external.py, test_print.py, test_limits.py +- pydantic/mcp-run-python: tests/test_sandbox.py +""" + +from __future__ import annotations + +import os +import shutil +from pathlib import Path +from typing import Any + +import pytest +import timbal as _timbal_pkg +from timbal import Agent +from timbal.tools.run_python import RunPython, _detect_imports, _resolve_dependencies + +EXECUTOR = "venv" + + +def _repo_root() -> Path: + """Locate the repo root (dir containing pyproject.toml) from the package.""" + start = Path(_timbal_pkg.__file__).resolve() + for parent in start.parents: + if (parent / "pyproject.toml").exists(): + return parent + raise RuntimeError("Could not locate timbal repo root") + + +@pytest.fixture +def run_python() -> RunPython: + return RunPython(executor=EXECUTOR) + + +async def collect(tool: RunPython, code: str, **kwargs: Any) -> dict[str, Any]: + result = await tool(code=code, **kwargs).collect() + assert result.error is None, result.error + return result.output + + +def double(x: int) -> int: + """Double an integer.""" + return x * 2 + + +def noop() -> str: + """No-op external function.""" + return "called" + + +def take_positional(a: int, b: int, c: int) -> str: + """Accept positional args (Monty test_external_function_positional_args).""" + assert (a, b, c) == (1, 2, 3) + return "ok" + + +def take_kwargs(**kwargs: Any) -> str: + """Accept keyword args.""" + assert kwargs == {"a": 1, "b": "two"} + return "ok" + + +def take_mixed(a: int, b: int, **kwargs: Any) -> str: + """Accept positional and keyword args.""" + assert (a, b) == (1, 2) + assert kwargs == {"x": "hello", "y": True} + return "ok" + + +def fail_tool() -> None: + """Raise a ValueError like Monty external-function error tests.""" + raise ValueError("intentional error") + + +class TestRunPythonInitialization: + def test_tool_creation(self): + tool = RunPython() + assert tool.name == "run_python" + assert tool.handler is not None + assert tool.timeout == 60.0 + + def test_tool_with_exposed_tools(self): + tool = RunPython(tools=[double]) + assert "double" in tool.description + + def test_detect_imports(self): + code = "import json\nfrom pathlib import Path\nimport tabulate" + imports = _detect_imports(code) + assert "json" in imports + assert "pathlib" in imports + assert "tabulate" in imports + + def test_resolve_dependencies_auto(self): + code = "import tabulate\nimport yaml" + deps = _resolve_dependencies(code, explicit=None, default=None) + assert "tabulate" in deps + assert "pyyaml" in deps + + def test_resolve_dependencies_explicit(self): + code = "x = 1" + deps = _resolve_dependencies(code, explicit=["requests>=2.0"], default=["httpx"]) + assert "requests>=2.0" in deps + assert "httpx" in deps + + +class TestRunPythonMontyBasic: + """Ported from pydantic/monty test_basic.py.""" + + @pytest.mark.asyncio + async def test_simple_expression(self, run_python: RunPython): + out = await collect(run_python, "1 + 2") + assert out["return_value"] == 3 + + @pytest.mark.asyncio + async def test_arithmetic(self, run_python: RunPython): + out = await collect(run_python, "10 * 5 - 3") + assert out["return_value"] == 47 + + @pytest.mark.asyncio + async def test_string_concatenation(self, run_python: RunPython): + out = await collect(run_python, '"hello" + " " + "world"') + assert out["return_value"] == "hello world" + + @pytest.mark.asyncio + async def test_multiple_runs_same_code(self, run_python: RunPython): + for x, expected in [(5, 10), (10, 20), (-3, -6)]: + code = f"x = {x}\nx * 2" + out = await collect(run_python, code) + assert out["return_value"] == expected + + @pytest.mark.asyncio + async def test_multiline_code(self, run_python: RunPython): + code = "x = 1\ny = 2\nx + y" + out = await collect(run_python, code) + assert out["return_value"] == 3 + + @pytest.mark.asyncio + async def test_function_definition_and_call(self, run_python: RunPython): + code = """ +def add(a, b): + return a + b + +add(3, 4) +""" + out = await collect(run_python, code) + assert out["return_value"] == 7 + + +class TestRunPythonMontyExceptions: + """Ported from pydantic/monty test_exceptions.py.""" + + @pytest.mark.asyncio + async def test_zero_division_error(self, run_python: RunPython): + out = await collect(run_python, "1 / 0") + assert out["status"] == "error" + assert out["error"]["type"] == "ZeroDivisionError" + + @pytest.mark.asyncio + async def test_value_error(self, run_python: RunPython): + out = await collect(run_python, "raise ValueError('bad value')") + assert out["status"] == "error" + assert out["error"]["type"] == "ValueError" + assert "bad value" in out["error"]["message"] + + @pytest.mark.asyncio + async def test_type_error(self, run_python: RunPython): + out = await collect(run_python, "'string' + 1") + assert out["status"] == "error" + assert out["error"]["type"] == "TypeError" + + @pytest.mark.asyncio + async def test_index_error(self, run_python: RunPython): + out = await collect(run_python, "[1, 2, 3][10]") + assert out["status"] == "error" + assert out["error"]["type"] == "IndexError" + + @pytest.mark.asyncio + async def test_key_error(self, run_python: RunPython): + out = await collect(run_python, "{'a': 1}['b']") + assert out["status"] == "error" + assert out["error"]["type"] == "KeyError" + + @pytest.mark.asyncio + async def test_name_error(self, run_python: RunPython): + out = await collect(run_python, "undefined_variable") + assert out["status"] == "error" + assert out["error"]["type"] == "NameError" + + @pytest.mark.asyncio + async def test_assertion_error(self, run_python: RunPython): + out = await collect(run_python, "assert False") + assert out["status"] == "error" + assert out["error"]["type"] == "AssertionError" + + @pytest.mark.asyncio + async def test_assertion_error_with_message(self, run_python: RunPython): + out = await collect(run_python, "assert False, 'custom message'") + assert out["status"] == "error" + assert out["error"]["type"] == "AssertionError" + assert "custom message" in out["error"]["message"] + + @pytest.mark.asyncio + async def test_nested_traceback(self, run_python: RunPython): + """Ported from mcp-run-python test_sandbox traceback case.""" + code = """ +def foo(): + 1 / 0 + +def bar(): + foo() + +def baz(): + bar() + +baz() +""" + out = await collect(run_python, code) + assert out["status"] == "error" + assert out["error"]["type"] == "ZeroDivisionError" + assert "traceback" in out["error"] + assert "foo" in out["error"]["traceback"] + + +class TestRunPythonMontyPrint: + """Ported from pydantic/monty test_print.py and mcp-run-python print cases.""" + + @pytest.mark.asyncio + async def test_print_basic(self, run_python: RunPython): + out = await collect(run_python, 'print("hello")') + assert "hello" in out["stdout"] + assert out["return_value"] is None + + @pytest.mark.asyncio + async def test_print_multiple_lines(self, run_python: RunPython): + code = 'print("line 1")\nprint("line 2")' + out = await collect(run_python, code) + assert "line 1" in out["stdout"] + assert "line 2" in out["stdout"] + + @pytest.mark.asyncio + async def test_print_with_values(self, run_python: RunPython): + out = await collect(run_python, "print(1, 2, 3)") + assert "1 2 3" in out["stdout"] + + @pytest.mark.asyncio + async def test_print_with_sep(self, run_python: RunPython): + out = await collect(run_python, 'print(1, 2, 3, sep="-")') + assert "1-2-3" in out["stdout"] + + @pytest.mark.asyncio + async def test_print_with_end(self, run_python: RunPython): + out = await collect(run_python, 'print("hello", end="!")') + assert "hello!" in out["stdout"] + + @pytest.mark.asyncio + async def test_print_then_expression(self, run_python: RunPython): + """Ported from mcp-run-python: print output + last expression return value.""" + out = await collect(run_python, "print(1)\n1") + assert "1" in out["stdout"] + assert out["return_value"] == 1 + + +class TestRunPythonMcpSandbox: + """Ported from pydantic/mcp-run-python tests/test_sandbox.py.""" + + @pytest.mark.asyncio + async def test_return_value_success(self, run_python: RunPython): + out = await collect(run_python, "a = 1\na + 1") + assert out["return_value"] == 2 + + @pytest.mark.asyncio + async def test_return_string(self, run_python: RunPython): + out = await collect(run_python, '"foobar"') + assert out["return_value"] == "foobar" + + @pytest.mark.asyncio + async def test_inline_globals_equivalent(self, run_python: RunPython): + """mcp-run-python injects globals; we inline equivalent assignments.""" + out = await collect(run_python, "a = [1, 2, 3]\na") + assert out["return_value"] == [1, 2, 3] + + @pytest.mark.asyncio + async def test_multiple_globals(self, run_python: RunPython): + out = await collect(run_python, "a = 4\nb = 5\na + b") + assert out["return_value"] == 9 + + @pytest.mark.asyncio + async def test_return_complex_dataclass(self, run_python: RunPython): + code = """ +from dataclasses import dataclass + +@dataclass +class Foobar: + a: int + b: str + c: bytes + +f = Foobar(1, "2", b"3") +{"a": f.a, "b": f.b, "c": f.c.decode()} +""" + out = await collect(run_python, code) + assert out["return_value"] == {"a": 1, "b": "2", "c": "3"} + + @pytest.mark.asyncio + async def test_print_error_name_error(self, run_python: RunPython): + out = await collect(run_python, "print(unknown)") + assert out["status"] == "error" + assert out["error"]["type"] == "NameError" + + @pytest.mark.asyncio + async def test_multiple_sequential_runs(self, run_python: RunPython): + """Ported from mcp-run-python test_multiple_commands.""" + for expected in (1, 2, 3): + out = await collect(run_python, f"print({expected})\n{expected}") + assert str(expected) in out["stdout"] + assert out["return_value"] == expected + + @pytest.mark.asyncio + async def test_missing_package_without_deps(self, run_python: RunPython): + """Ported from mcp-run-python ModuleNotFoundError when deps not installed.""" + out = await collect(run_python, "__import__('timbal_nonexistent_module_xyz')") + assert out["status"] == "error" + assert out["error"]["type"] in {"ModuleNotFoundError", "ImportError"} + + @pytest.mark.asyncio + async def test_explicit_deps_install_package(self, run_python: RunPython): + """Explicit dependencies make third-party imports available (mcp-run-python deps param).""" + out = await collect( + run_python, + "import tabulate\ntabulate.tabulate([[1]], headers=['x'])", + dependencies=["tabulate"], + ) + assert out["status"] == "success" + + @pytest.mark.asyncio + @pytest.mark.skipif(shutil.which("uv") is None, reason="uv not available") + async def test_numpy_with_dependencies(self): + """Ported from mcp-run-python return-numpy-success.""" + tool = RunPython(executor="uv") + code = "import numpy\nnumpy.array([1, 2, 3]).tolist()" + out = await collect(tool, code, dependencies=["numpy"]) + assert out["return_value"] == [1, 2, 3] + + @pytest.mark.asyncio + async def test_python_version_available(self, run_python: RunPython): + """Ported from mcp-run-python python-version case.""" + code = "import sys\nlist(sys.version_info[:3])" + out = await collect(run_python, code) + assert isinstance(out["return_value"], list) + assert len(out["return_value"]) == 3 + + +class TestRunPythonExecution: + @pytest.mark.asyncio + async def test_timeout(self): + """Ported from pydantic/monty test_limits.py ResourceLimits(max_duration_secs).""" + tool = RunPython(timeout=1.0, executor=EXECUTOR) + out = await collect(tool, "import time\ntime.sleep(5)") + assert out["status"] == "error" + assert out["error"]["type"] == "TimeoutError" + + @pytest.mark.asyncio + @pytest.mark.skipif(shutil.which("uv") is None, reason="uv not available") + async def test_auto_dependency_install_uv(self): + """Ported from mcp-run-python auto dependency detection.""" + tool = RunPython(executor="uv") + code = """ +import tabulate +tabulate.tabulate([["a", 1], ["b", 2]], headers=["name", "value"]) +""" + out = await collect(tool, code) + assert out["status"] == "success" + assert "a" in out["return_value"] + + @pytest.mark.asyncio + async def test_explicit_dependencies_venv(self, run_python: RunPython): + code = """ +import tabulate +tabulate.tabulate([["x", 1]], headers=["k", "v"]) +""" + out = await collect(run_python, code, dependencies=["tabulate"]) + assert out["status"] == "success" + + +class TestRunPythonMontyExternal: + """Ported from pydantic/monty test_external.py (code mode / external_functions).""" + + @pytest.mark.asyncio + async def test_external_function_no_args(self): + tool = RunPython(tools=[noop], executor=EXECUTOR) + out = await collect(tool, "noop()") + assert out["return_value"] == "called" + + @pytest.mark.asyncio + async def test_external_function_positional_args(self): + tool = RunPython(tools=[take_positional], executor=EXECUTOR) + out = await collect(tool, "take_positional(1, 2, 3)") + assert out["return_value"] == "ok" + + @pytest.mark.asyncio + async def test_external_function_kwargs_only(self): + tool = RunPython(tools=[take_kwargs], executor=EXECUTOR) + out = await collect(tool, 'take_kwargs(a=1, b="two")') + assert out["return_value"] == "ok" + + @pytest.mark.asyncio + async def test_external_function_mixed_args_kwargs(self): + tool = RunPython(tools=[take_mixed], executor=EXECUTOR) + out = await collect(tool, 'take_mixed(1, 2, x="hello", y=True)') + assert out["return_value"] == "ok" + + @pytest.mark.asyncio + async def test_external_function_complex_types(self): + def take_complex(items: list[int], mapping: dict[str, str]) -> str: + assert items == [1, 2] + assert mapping == {"key": "value"} + return "ok" + + tool = RunPython(tools=[take_complex], executor=EXECUTOR) + out = await collect(tool, 'take_complex([1, 2], {"key": "value"})') + assert out["return_value"] == "ok" + + @pytest.mark.asyncio + async def test_external_function_raises_exception(self): + tool = RunPython(tools=[fail_tool], executor=EXECUTOR) + out = await collect(tool, "fail_tool()") + assert out["status"] == "error" + assert out["error"]["type"] == "ValueError" + assert "intentional error" in out["error"]["message"] + + @pytest.mark.asyncio + async def test_external_function_caught_inside_script(self): + """Ported from pydantic/monty test_external.py except-in-script cases.""" + tool = RunPython(tools=[fail_tool], executor=EXECUTOR) + code = """ +try: + fail_tool() + result = 'no error' +except ValueError: + result = 'caught' +result +""" + out = await collect(tool, code) + assert out["return_value"] == "caught" + + @pytest.mark.asyncio + async def test_undeclared_external_function_name_error(self, run_python: RunPython): + """Ported from pydantic/monty: unknown external function -> NameError.""" + out = await collect(run_python, "missing_fn()") + assert out["status"] == "error" + assert out["error"]["type"] == "NameError" + + @pytest.mark.asyncio + async def test_double_external_function(self): + tool = RunPython(tools=[double], executor=EXECUTOR) + out = await collect(tool, "double(5)") + assert out["return_value"] == 10 + + @pytest.mark.asyncio + async def test_external_function_in_loop(self): + tool = RunPython(tools=[double], executor=EXECUTOR) + code = """ +values = [double(1), double(2), double(3)] +sum(values) +""" + out = await collect(tool, code) + assert out["return_value"] == 12 + + +class TestRunPythonWithAgent: + @pytest.mark.integration + @pytest.mark.asyncio + async def test_agent_uses_run_python(self): + agent = Agent( + name="python_runner", + model="openai/gpt-4o-mini", + system_prompt=( + "You solve math by calling run_python with Python code. " + "Always use run_python for calculations. Return only the numeric result." + ), + tools=[RunPython(executor=EXECUTOR)], + ) + response = await agent(prompt="What is 17 * 23? Use run_python.").collect() + assert response.error is None + assert response.output is not None + + +class TestRunPythonHardening: + """Hardening behaviour inspired by E2B / Monty top-performer guidance.""" + + @pytest.mark.asyncio + async def test_secrets_not_leaked_to_child(self, monkeypatch: pytest.MonkeyPatch): + """Host secrets must never be visible to executed code.""" + monkeypatch.setenv("OPENAI_API_KEY", "sk-supersecret") + monkeypatch.setenv("TIMBAL_API_KEY", "tk-supersecret") + tool = RunPython(executor=EXECUTOR) + code = "import os\n[os.environ.get('OPENAI_API_KEY'), os.environ.get('TIMBAL_API_KEY')]" + out = await collect(tool, code) + assert out["return_value"] == [None, None] + + @pytest.mark.asyncio + async def test_env_passthrough_allowlist(self, monkeypatch: pytest.MonkeyPatch): + """Explicitly allowlisted variables are forwarded to the child.""" + monkeypatch.setenv("MY_PUBLIC_FLAG", "enabled") + tool = RunPython(executor=EXECUTOR, env_passthrough=["MY_PUBLIC_FLAG"]) + out = await collect(tool, "import os\nos.environ.get('MY_PUBLIC_FLAG')") + assert out["return_value"] == "enabled" + + @pytest.mark.asyncio + async def test_stdout_truncation(self): + tool = RunPython(executor=EXECUTOR, max_output_chars=200) + out = await collect(tool, "print('A' * 5000)") + assert len(out["stdout"]) < 500 + assert "characters truncated" in out["stdout"] + + @pytest.mark.asyncio + async def test_code_size_limit(self, run_python: RunPython): + oversized = "x = 1\n" + ("# pad\n" * 40000) + result = await run_python(code=oversized).collect() + out = result.output + assert out["status"] == "error" + assert out["error"]["type"] == "ValueError" + assert "maximum size" in out["error"]["message"] + + @pytest.mark.asyncio + async def test_result_size_limit(self, run_python: RunPython): + result = await run_python(code="'X' * 2_000_000").collect() + out = result.output + assert out["status"] == "error" + assert out["error"]["type"] == "ValueError" + assert "maximum size" in out["error"]["message"] + + @pytest.mark.asyncio + async def test_timeout_kills_runaway(self): + tool = RunPython(executor=EXECUTOR, timeout=2.0) + result = await tool(code="while True:\n pass").collect() + out = result.output + assert out["status"] == "error" + assert out["error"]["type"] == "TimeoutError" + + +class TestRunPythonAsync: + """Top-level await support (so agent-written async code just works).""" + + @pytest.mark.asyncio + async def test_top_level_await_return_value(self, run_python: RunPython): + code = """ +import asyncio +await asyncio.sleep(0) +total = 0 +for i in range(5): + await asyncio.sleep(0) + total += i +total +""" + out = await collect(run_python, code) + assert out["return_value"] == 10 + + @pytest.mark.asyncio + async def test_top_level_await_with_gather(self, run_python: RunPython): + code = """ +import asyncio + +async def work(n): + await asyncio.sleep(0) + return n * n + +results = await asyncio.gather(*[work(i) for i in range(4)]) +sum(results) +""" + out = await collect(run_python, code) + assert out["return_value"] == 14 + + @pytest.mark.asyncio + async def test_await_in_codemode(self): + """Async user code can still call exposed tools synchronously.""" + tool = RunPython(tools=[double], executor=EXECUTOR) + code = """ +import asyncio +await asyncio.sleep(0) +double(21) +""" + out = await collect(tool, code) + assert out["return_value"] == 42 + + +class TestTimbalInTimbal: + """Install timbal inside the sandbox and run an Agent within an Agent. + + This is the headline use case: an outer agent generates Python that spins up + a fully isolated inner Timbal agent (its own installed timbal, its own LLM + call) and returns the result. Requires ANTHROPIC_API_KEY and uv. + """ + + @pytest.mark.integration + @pytest.mark.asyncio + @pytest.mark.skipif( + not os.environ.get("ANTHROPIC_API_KEY"), reason="ANTHROPIC_API_KEY required" + ) + @pytest.mark.skipif(shutil.which("uv") is None, reason="uv required") + async def test_inner_agent_runs(self, monkeypatch: pytest.MonkeyPatch): + # hatch-vcs derives the version from git, which may be unavailable in the + # isolated build; pin a pretend version so the source build never fails. + monkeypatch.setenv("SETUPTOOLS_SCM_PRETEND_VERSION", "0.0.0") + + repo_root = _repo_root() + tool = RunPython( + dependencies=[f"timbal @ file://{repo_root}"], + env_passthrough=["ANTHROPIC_API_KEY", "SETUPTOOLS_SCM_PRETEND_VERSION"], + timeout=900, + executor="uv", + ) + + code = """ +from timbal import Agent + +inner = Agent( + name="inner_agent", + model="anthropic/claude-haiku-4-5", + tools=[], + max_tokens=64, + system_prompt="You reply with exactly one lowercase word and nothing else.", +) + +result = await inner(prompt="Reply with the single word: pong").collect() +text = result.output.collect_text().strip().lower() +{"text": text, "status": result.status.code} +""" + + result = await tool(code=code).collect() + assert result.error is None, result.error + out = result.output + assert out["status"] == "success", out + assert out["return_value"]["status"] == "success", out["return_value"] + assert "pong" in out["return_value"]["text"], out["return_value"] + + @pytest.mark.integration + @pytest.mark.asyncio + @pytest.mark.skipif( + not os.environ.get("ANTHROPIC_API_KEY"), reason="ANTHROPIC_API_KEY required" + ) + @pytest.mark.skipif(shutil.which("uv") is None, reason="uv required") + async def test_outer_agent_delegates_to_inner_agent(self, monkeypatch: pytest.MonkeyPatch): + """End-to-end: an outer Agent uses run_python to launch an inner Agent.""" + monkeypatch.setenv("SETUPTOOLS_SCM_PRETEND_VERSION", "0.0.0") + repo_root = _repo_root() + + run_python = RunPython( + dependencies=[f"timbal @ file://{repo_root}"], + env_passthrough=["ANTHROPIC_API_KEY", "SETUPTOOLS_SCM_PRETEND_VERSION"], + timeout=900, + executor="uv", + ) + + outer = Agent( + name="orchestrator", + model="anthropic/claude-haiku-4-5", + max_tokens=1024, + system_prompt=( + "You orchestrate work by writing Python for the run_python tool. " + "When asked to delegate, write code that imports `from timbal import Agent`, " + "creates an inner Agent with model 'anthropic/claude-haiku-4-5' and max_tokens=64, " + "calls `result = await inner(prompt=...).collect()`, and ends with the expression " + "`result.output.collect_text()` so the value is returned. " + "Then report the inner agent's answer to the user." + ), + tools=[run_python], + ) + + response = await outer( + prompt=( + "Delegate to an inner Timbal agent: ask it to translate the word " + "'hello' to Spanish. Use run_python to create and run that inner agent." + ) + ).collect() + assert response.error is None, response.error + assert response.output is not None + assert "hola" in response.output.collect_text().lower() diff --git a/python/tests/tools/validate_run_python_agent.py b/python/tests/tools/validate_run_python_agent.py new file mode 100644 index 00000000..d84fa879 --- /dev/null +++ b/python/tests/tools/validate_run_python_agent.py @@ -0,0 +1,116 @@ +""" +Manual validation: RunPython with a real Agent on realistic use cases. + +Run: + uv run python python/tests/tools/validate_run_python_agent.py +""" +# ruff: noqa: T201 # this is a manual runner; prints are the intended output + +from __future__ import annotations + +import asyncio +import os +from pathlib import Path + +from dotenv import load_dotenv +from timbal import Agent +from timbal.tools.run_python import RunPython + +# Load API keys from examples if present +load_dotenv(Path(__file__).resolve().parents[3] / "examples" / "mcp" / ".env") + +SALES_CSV = """product,units,price +Widget A,120,9.99 +Widget B,85,14.50 +Gadget C,200,4.25 +Widget A,30,9.99 +Gadget C,50,4.25 +Premium D,15,49.00 +Widget B,40,14.50 +""" + +PRICE_CATALOG = { + "WIDGET-A": 9.99, + "WIDGET-B": 14.50, + "GADGET-C": 4.25, + "PREMIUM-D": 49.00, +} + + +def get_unit_price(sku: str) -> float: + """Look up unit price for a SKU from the product catalog.""" + key = sku.upper().replace(" ", "-") + if key not in PRICE_CATALOG: + raise ValueError(f"Unknown SKU: {sku}") + return PRICE_CATALOG[key] + + +async def scenario_pandas_analysis() -> None: + """Use case: data analyst computes top products by revenue with pandas.""" + print("\n=== Scenario 1: Sales analysis with pandas (RunPython only) ===") + agent = Agent( + name="sales_analyst", + model="openai/gpt-4o-mini", + max_tokens=1024, + system_prompt=( + "You are a data analyst. Always use run_python to analyze data. " + "Write Python that uses pandas. Put the CSV in a triple-quoted string variable. " + "Return a concise summary of findings after running the code." + ), + tools=[RunPython(executor="auto", timeout=120)], + ) + prompt = ( + "Analyze this sales CSV and tell me the top 3 products by total revenue " + "(units * price summed per product). Include the revenue numbers.\n\n" + f"CSV:\n{SALES_CSV}" + ) + result = await agent(prompt=prompt).collect() + text = result.output.collect_text() if result.output else "" + print("Agent response:", text[:800]) + assert result.error is None, result.error + assert result.output is not None + # Ground truth: Widget A = 150*9.99=1498.5, Gadget C = 250*4.25=1062.5, Widget B = 125*14.5=1812.5 + # Top 3 by revenue: Widget B (~1812.5), Widget A (~1498.5), Gadget C (~1062.5) + lower = text.lower() + assert "widget b" in lower or "widget-b" in lower.replace(" ", "") + print("Scenario 1: PASS") + + +async def scenario_code_mode_pricing() -> None: + """Use case: code mode — script calls get_unit_price for each line item.""" + print("\n=== Scenario 2: Code mode batch pricing (RunPython + get_unit_price) ===") + agent = Agent( + name="pricing_agent", + model="openai/gpt-4o-mini", + max_tokens=1024, + system_prompt=( + "You compute order totals using run_python. " + "Inside the script you MUST call get_unit_price(sku) for each SKU — " + "do not hardcode prices. SKUs: WIDGET-A x2, GADGET-C x5, PREMIUM-D x1." + ), + tools=[RunPython(tools=[get_unit_price], executor="auto", timeout=60)], + ) + prompt = ( + "Use run_python to calculate the total order value for: " + "2x WIDGET-A, 5x GADGET-C, 1x PREMIUM-D. " + "Call get_unit_price for each SKU inside the script. Return the numeric total." + ) + result = await agent(prompt=prompt).collect() + text = result.output.collect_text() if result.output else "" + print("Agent response:", text[:500]) + assert result.error is None, result.error + expected = 2 * 9.99 + 5 * 4.25 + 1 * 49.00 # 90.23 + assert "90.23" in text.replace(",", "") or "90.2" in text + print(f"Scenario 2: PASS (expected total {expected:.2f})") + + +async def main() -> None: + if not os.getenv("OPENAI_API_KEY"): + raise SystemExit("OPENAI_API_KEY required for live agent validation") + await scenario_pandas_analysis() + await scenario_code_mode_pricing() + print("\nAll live agent scenarios passed.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/timbal/tools/__init__.py b/python/timbal/tools/__init__.py index c9aedab6..3eaa612b 100644 --- a/python/timbal/tools/__init__.py +++ b/python/timbal/tools/__init__.py @@ -510,6 +510,7 @@ ResendSendEmail, ResendUpdateEmail, ) + from .run_python import RunPython from .salesforce import ( SalesforceCreateCase, SalesforceCreateComment, @@ -1287,6 +1288,7 @@ "QuiverAIListModels", "QuiverAIVectorizeSVG", "Read", + "RunPython", "ScraperAPIAmazonProduct", "ScraperAPIAmazonSearch", "ScraperAPIAsyncScrape", @@ -2327,6 +2329,7 @@ "QuiverAIListModels": ".quiver_ai", "QuiverAIVectorizeSVG": ".quiver_ai", "Read": ".read", + "RunPython": ".run_python", "ScraperAPIAmazonProduct": ".scraperapi", "ScraperAPIAmazonSearch": ".scraperapi", "ScraperAPIAsyncScrape": ".scraperapi", diff --git a/python/timbal/tools/run_python.py b/python/timbal/tools/run_python.py new file mode 100644 index 00000000..9effe13a --- /dev/null +++ b/python/timbal/tools/run_python.py @@ -0,0 +1,786 @@ +""" +RunPython tool for executing Python scripts in isolated environments. + +Supports: +- Real CPython execution via uv (PEP 723 inline dependencies) with venv+pip fallback +- Auto-detection of dependencies from import statements +- Code mode: scripts can call back into other Timbal tools via a unix-socket RPC bridge +- Structured stdout/stderr/return_value/error capture +""" + +from __future__ import annotations + +import ast +import asyncio +import contextlib +import inspect +import json +import os +import re +import shutil +import signal +import sys +import tempfile +import traceback +from pathlib import Path +from typing import Any, Literal + +from pydantic import Field, PrivateAttr + +from ..core.tool import Tool +from ..state import get_run_context + +# Common import name -> PyPI package name mismatches. +_IMPORT_TO_PACKAGE: dict[str, str] = { + "PIL": "pillow", + "bs4": "beautifulsoup4", + "cv2": "opencv-python", + "dotenv": "python-dotenv", + "googleapiclient": "google-api-python-client", + "jwt": "pyjwt", + "sklearn": "scikit-learn", + "yaml": "pyyaml", +} + +_RESULT_FILENAME = ".timbal_run_result.json" +_RPC_ENV_VAR = "TIMBAL_RPC_SOCKET" + +# Hardening limits. +_MAX_CODE_CHARS = 200_000 +_DEFAULT_MAX_OUTPUT_CHARS = 10_000 +_MAX_RESULT_BYTES = 1_000_000 + +# Minimal set of environment variables forwarded to the child process so the +# interpreter/uv can run. Everything else (secrets like API keys, tokens) is +# stripped by default so executed code cannot exfiltrate host credentials. +# Mirrors the top-performer guidance (E2B/Monty): the sandbox receives inputs +# and returns outputs; it should never receive your secrets. +_SAFE_ENV_VARS: frozenset[str] = frozenset({ + "PATH", + "HOME", + "LANG", + "LANGUAGE", + "LC_ALL", + "LC_CTYPE", + "TZ", + "TERM", + "TMPDIR", + "TEMP", + "TMP", + # Windows essentials + "SYSTEMROOT", + "SystemRoot", + "PATHEXT", + "USERPROFILE", + "COMSPEC", + "WINDIR", + "NUMBER_OF_PROCESSORS", + "PROCESSOR_ARCHITECTURE", + # uv / Python cache + config so isolated runs stay fast without leaking secrets + "UV_CACHE_DIR", + "UV_PYTHON_INSTALL_DIR", + "XDG_CACHE_HOME", + "XDG_DATA_HOME", + "XDG_CONFIG_HOME", + "PYTHONPATH", +}) + +# Preamble injected before user code in the child process. +_CHILD_PREAMBLE = ''' +import ast as __timbal_ast +import builtins as __timbal_builtins__ +import json as __timbal_json +import os as __timbal_os +import socket as __timbal_socket +import sys as __timbal_sys +import traceback as __timbal_traceback + +__TIMBAL_RESULT_PATH = __timbal_os.environ.get("TIMBAL_RESULT_PATH", "") + + +def __timbal_call_tool__(name, args, kwargs): + sock_path = __timbal_os.environ.get({rpc_env!r}) + if not sock_path: + raise RuntimeError("Code mode is not available: RPC socket not configured") + payload = __timbal_json.dumps({{"name": name, "args": args, "kwargs": kwargs}}).encode() + with __timbal_socket.socket(__timbal_socket.AF_UNIX, __timbal_socket.SOCK_STREAM) as sock: + sock.connect(sock_path) + sock.sendall(payload + b"\\n") + chunks = [] + while True: + chunk = sock.recv(65536) + if not chunk: + break + chunks.append(chunk) + if b"\\n" in chunk: + break + line = b"".join(chunks).split(b"\\n", 1)[0] + response = __timbal_json.loads(line.decode()) + if response.get("error"): + err = response["error"] + exc_name = err.get("type", "RuntimeError") + message = err.get("message", str(err)) + exc_cls = getattr(__timbal_builtins__, exc_name, None) + if exc_cls is None or not isinstance(exc_cls, type) or not issubclass(exc_cls, BaseException): + if exc_name == "KeyError" and "Unknown tool" in message: + raise NameError(f"name '{{name}}' is not defined") + exc_cls = RuntimeError + message = f"{{exc_name}}: {{message}}" + raise exc_cls(message) + return response.get("output") + + +def __timbal_write_result__(payload): + if __TIMBAL_RESULT_PATH: + with open(__TIMBAL_RESULT_PATH, "w", encoding="utf-8") as fh: + __timbal_json.dump(payload, fh) + + +def __timbal_has_top_level_await__(tree): + class _Visitor(__timbal_ast.NodeVisitor): + def __init__(self): + self.found = False + def visit_FunctionDef(self, node): + return None + def visit_AsyncFunctionDef(self, node): + return None + def visit_Lambda(self, node): + return None + def visit_Await(self, node): + self.found = True + def visit_AsyncFor(self, node): + self.found = True + self.generic_visit(node) + def visit_AsyncWith(self, node): + self.found = True + self.generic_visit(node) + visitor = _Visitor() + visitor.visit(tree) + return visitor.found + + +def __timbal_execute_user_code__(code): + result = {{"return_value": None, "error": None}} + try: + tree = __timbal_ast.parse(code) + namespace = dict(globals()) + namespace["__name__"] = "__main__" + if __timbal_has_top_level_await__(tree): + import asyncio as __timbal_asyncio + import textwrap as __timbal_textwrap + if tree.body and isinstance(tree.body[-1], __timbal_ast.Expr): + __timbal_ret = __timbal_ast.Return(value=tree.body[-1].value) + __timbal_ast.copy_location(__timbal_ret, tree.body[-1]) + tree.body[-1] = __timbal_ret + __timbal_ast.fix_missing_locations(tree) + __timbal_wrapped = ( + "async def __timbal_async_main__():\\n" + + __timbal_textwrap.indent(__timbal_ast.unparse(tree), " ") + ) + exec(compile(__timbal_wrapped, "", "exec"), namespace) + result["return_value"] = __timbal_asyncio.run(namespace["__timbal_async_main__"]()) + elif tree.body and isinstance(tree.body[-1], __timbal_ast.Expr): + if len(tree.body) > 1: + exec( + compile(__timbal_ast.Module(body=tree.body[:-1], type_ignores=[]), "", "exec"), + namespace, + ) + last_expr = compile(__timbal_ast.Expression(body=tree.body[-1].value), "", "eval") + result["return_value"] = eval(last_expr, namespace) + else: + exec(compile(tree, "", "exec"), namespace) + except Exception as exc: + result["error"] = {{ + "type": type(exc).__name__, + "message": str(exc), + "traceback": __timbal_traceback.format_exc(), + }} + if result["error"] is None: + try: + __timbal_json.dumps(result["return_value"]) + except (TypeError, ValueError): + result["return_value"] = repr(result["return_value"]) + return result +'''.format(rpc_env=_RPC_ENV_VAR) # noqa: UP032 + + +def _build_child_env(passthrough: list[str] | None) -> dict[str, str]: + """Build a minimal environment for the child process. + + Only safe, non-secret variables are forwarded by default. Additional names + can be explicitly allowlisted via ``env_passthrough``. + """ + allowed = set(_SAFE_ENV_VARS) + if passthrough: + allowed.update(passthrough) + return {key: value for key, value in os.environ.items() if key in allowed} + + +def _truncate_output(text: str, limit: int) -> str: + """Cap output to ``limit`` chars, keeping head and tail with a drop marker.""" + if limit <= 0 or len(text) <= limit: + return text + dropped = len(text) - limit + head_len = limit // 2 + tail_len = limit - head_len + head = text[:head_len] + tail = text[-tail_len:] + return f"{head}\n... [{dropped} characters truncated] ...\n{tail}" + + +def _ensure_tool(obj: Any) -> Tool: + if isinstance(obj, Tool): + return obj + if callable(obj): + return Tool(handler=obj) + raise TypeError(f"Expected Tool or callable, got {type(obj)!r}") + + +def _detect_imports(code: str) -> list[str]: + """Extract top-level import module names from Python source.""" + try: + tree = ast.parse(code) + except SyntaxError: + return [] + + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + modules.add(alias.name.split(".")[0]) + elif isinstance(node, ast.ImportFrom) and node.module: + modules.add(node.module.split(".")[0]) + return sorted(modules) + + +def _stdlib_modules() -> set[str]: + return set(sys.stdlib_module_names) + + +def _dep_base_name(spec: str) -> str: + """Extract the normalized distribution name from a requirement spec. + + Handles forms like ``timbal``, ``timbal>=1.0``, ``timbal[extra]`` and the + PEP 508 URL form ``timbal @ file:///path``. + """ + s = spec.split(";", 1)[0].strip() + if "@" in s: + s = s.split("@", 1)[0].strip() + s = re.split(r"[\s\[<>=!~(]", s, maxsplit=1)[0].strip() + return s.replace("_", "-").lower() + + +def _resolve_dependencies( + code: str, + explicit: list[str] | None, + default: list[str] | None, +) -> list[str]: + """Merge explicit/default dependencies with auto-detected imports.""" + resolved: list[str] = [] + seen: set[str] = set() + + def add(dep: str) -> None: + dep = dep.strip() + if dep and dep not in seen: + seen.add(dep) + resolved.append(dep) + + pinned_bases: set[str] = set() + for source in (default or [], explicit or []): + for dep in source: + pinned_bases.add(_dep_base_name(dep)) + add(dep) + + stdlib = _stdlib_modules() + for module in _detect_imports(code): + if module in stdlib or module.startswith("_"): + continue + package = _IMPORT_TO_PACKAGE.get(module, module) + # Skip auto-detected packages already pinned explicitly (e.g. a local + # `timbal @ file://...` must not be shadowed by a bare `timbal`). + if _dep_base_name(package) in pinned_bases: + continue + add(package) + + return resolved + + +def _build_pep723_header(dependencies: list[str]) -> str: + if not dependencies: + return "" + lines = ["# /// script", "# dependencies = ["] + for dep in dependencies: + lines.append(f'# "{dep}",') + lines.append("# ]") + lines.append("# ///") + return "\n".join(lines) + "\n" + + +def _build_proxy_functions(tools: dict[str, Tool]) -> str: + lines: list[str] = [] + for name in tools: + if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name): + continue + lines.append( + f"def {name}(*args, **kwargs):\n" + f' return __timbal_call_tool__("{name}", list(args), kwargs)\n' + ) + return "\n".join(lines) + + +def _build_script(code: str, tools: dict[str, Tool], dependencies: list[str]) -> str: + header = _build_pep723_header(dependencies) + proxies = _build_proxy_functions(tools) + proxy_block = f"\n{proxies}\n" if proxies else "\n" + body = f''' +{proxy_block} +__timbal_result__ = __timbal_execute_user_code__({json.dumps(code)}) +__timbal_write_result__(__timbal_result__) +''' + return header + _CHILD_PREAMBLE + body + + +def _serialize_output(value: Any) -> Any: + try: + json.dumps(value) + return value + except (TypeError, ValueError): + return {"__timbal_serialized__": repr(value), "__type__": type(value).__name__} + + +class ToolDispatchError(Exception): + """Structured tool error from the code-mode RPC bridge.""" + + def __init__(self, error: dict[str, Any]) -> None: + self.error = error + super().__init__(error.get("message", str(error))) + + +async def _dispatch_exposed_tool(tool: Tool, args: list[Any], kwargs: dict[str, Any]) -> Any: + """Invoke an exposed tool from the code-mode RPC bridge.""" + sig = inspect.signature(tool.handler) + bound = sig.bind(*args, **kwargs) + var_positional = next( + (param.name for param in sig.parameters.values() if param.kind == inspect.Parameter.VAR_POSITIONAL), + None, + ) + var_keyword = next( + (param.name for param in sig.parameters.values() if param.kind == inspect.Parameter.VAR_KEYWORD), + None, + ) + positional_params = [ + param.name + for param in sig.parameters.values() + if param.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) + ] + + async def _call_handler(**call_kwargs: Any) -> Any: + if inspect.iscoroutinefunction(tool.handler): + return await tool.handler(**call_kwargs) + return await asyncio.to_thread(tool.handler, **call_kwargs) + + if var_keyword and not positional_params and not var_positional: + return await _call_handler(**bound.kwargs) + if var_positional and not positional_params and not var_keyword: + handler_args = bound.arguments.get(var_positional, ()) + if inspect.iscoroutinefunction(tool.handler): + return await tool.handler(*handler_args) + return await asyncio.to_thread(tool.handler, *handler_args) + if var_keyword and positional_params: + pos_values = {param_name: bound.arguments[param_name] for param_name in positional_params} + extra_kw = bound.arguments.get(var_keyword, {}) + return await _call_handler(**pos_values, **extra_kw) + + result = await tool(**bound.arguments).collect() + if result.error: + raise ToolDispatchError(result.error) + return result.output + + +class _ToolRpcServer: + """Async unix-socket RPC server dispatching tool calls from child process.""" + + def __init__(self, tools: dict[str, Tool]) -> None: + self._tools = tools + self._socket_path: str | None = None + self._server: asyncio.AbstractServer | None = None + + @property + def socket_path(self) -> str | None: + return self._socket_path + + async def start(self) -> None: + if not self._tools: + return + tmp = tempfile.NamedTemporaryFile(prefix="timbal-rpc-", suffix=".sock", delete=False) + tmp.close() + self._socket_path = tmp.name + Path(self._socket_path).unlink(missing_ok=True) + self._server = await asyncio.start_unix_server(self._handle_client, path=self._socket_path) + + async def stop(self) -> None: + if self._server is not None: + self._server.close() + await self._server.wait_closed() + self._server = None + if self._socket_path: + Path(self._socket_path).unlink(missing_ok=True) + self._socket_path = None + + async def _handle_client( + self, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + try: + data = await reader.readline() + if not data: + return + request = json.loads(data.decode()) + name = request.get("name") + args = request.get("args") or [] + kwargs = request.get("kwargs") or {} + tool = self._tools.get(name) + if tool is None: + response = {"output": None, "error": {"type": "KeyError", "message": f"Unknown tool: {name}"}} + else: + try: + output = await _dispatch_exposed_tool(tool, args, kwargs) + response = {"output": _serialize_output(output), "error": None} + except ToolDispatchError as exc: + response = {"output": None, "error": exc.error} + except TypeError as exc: + response = { + "output": None, + "error": {"type": "TypeError", "message": str(exc), "traceback": traceback.format_exc()}, + } + except Exception as exc: + response = { + "output": None, + "error": { + "type": type(exc).__name__, + "message": str(exc), + "traceback": traceback.format_exc(), + }, + } + writer.write(json.dumps(response).encode() + b"\n") + await writer.drain() + finally: + writer.close() + await writer.wait_closed() + + +def _kill_process_tree(process: asyncio.subprocess.Process) -> None: + """Best-effort terminate the child and any subprocesses it spawned. + + On POSIX the child is started in its own session (process group) so a + runaway ``uv run`` that forks Python is killed entirely, not just the + top-level wrapper. + """ + if os.name == "posix": + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + return + with contextlib.suppress(ProcessLookupError): + process.kill() + + +async def _run_subprocess( + cmd: list[str], + *, + cwd: Path, + env: dict[str, str], + timeout: float, +) -> tuple[int, str, str]: + popen_kwargs: dict[str, Any] = {} + if os.name == "posix": + popen_kwargs["start_new_session"] = True + + process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=str(cwd), + env=env, + **popen_kwargs, + ) + try: + stdout_bytes, stderr_bytes = await asyncio.wait_for(process.communicate(), timeout=timeout) + except TimeoutError: + _kill_process_tree(process) + with contextlib.suppress(Exception): + await process.wait() + raise + stdout = stdout_bytes.decode("utf-8", errors="replace") if stdout_bytes else "" + stderr = stderr_bytes.decode("utf-8", errors="replace") if stderr_bytes else "" + return process.returncode or 0, stdout, stderr + + +def _build_command(script_path: Path, executor: Literal["uv", "venv"], venv_python: Path | None) -> list[str]: + if executor == "uv": + uv = shutil.which("uv") + if uv: + return [uv, "run", "--no-project", str(script_path)] + if venv_python is not None: + return [str(venv_python), str(script_path)] + return [sys.executable, str(script_path)] + + +async def _ensure_venv_deps(venv_dir: Path, dependencies: list[str]) -> Path: + python_path = venv_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + if not python_path.exists(): + create = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "venv", + str(venv_dir), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + await create.wait() + if create.returncode != 0: + _, stderr = await create.communicate() + raise RuntimeError(f"Failed to create venv: {stderr.decode()}") + + if dependencies: + pip_cmd = [str(python_path), "-m", "pip", "install", "--quiet", *dependencies] + pip_proc = await asyncio.create_subprocess_exec( + *pip_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + await pip_proc.wait() + if pip_proc.returncode != 0: + _, stderr = await pip_proc.communicate() + raise RuntimeError(f"Failed to install dependencies: {stderr.decode()}") + + return python_path + + +def _tools_description(tools: dict[str, Tool]) -> str: + if not tools: + return "" + lines = ["Available functions in the script environment:"] + for name, tool in tools.items(): + try: + sig = inspect.signature(tool.handler) + lines.append(f"- {name}{sig}") + except (TypeError, ValueError): + lines.append(f"- {name}(...)") + return "\n".join(lines) + + +class RunPython(Tool): + """Execute Python code in an isolated environment with optional code-mode tool callbacks.""" + + default_dependencies: list[str] | None = Field( + default=None, + description="Default pip dependencies merged with auto-detected imports and per-call overrides.", + ) + timeout: float = Field(default=60.0, description="Maximum execution time in seconds.") + executor: Literal["uv", "venv", "auto"] = Field( + default="auto", + description="Execution backend: uv (PEP 723), venv+pip fallback, or auto (prefer uv).", + ) + exposed_tools: list[Any] = Field( + default_factory=list, + description="Tools/callables exposed to the script as functions (code mode).", + ) + max_output_chars: int = Field( + default=_DEFAULT_MAX_OUTPUT_CHARS, + description="Max chars kept per stdout/stderr channel (head+tail) to protect the context window.", + ) + env_passthrough: list[str] | None = Field( + default=None, + description=( + "Allowlist of host environment variable names forwarded to the executed code. " + "By default secrets (API keys, tokens) are NOT exposed to the sandbox." + ), + ) + + _tool_map: dict[str, Tool] = PrivateAttr(default_factory=dict) + + def model_post_init(self, __context: Any) -> None: + super().model_post_init(__context) + self._tool_map = {} + for item in self.exposed_tools: + tool = _ensure_tool(item) + self._tool_map[tool.name] = tool + + def __init__( + self, + tools: list[Any] | None = None, + dependencies: list[str] | None = None, + timeout: float = 60.0, + executor: Literal["uv", "venv", "auto"] = "auto", + max_output_chars: int = _DEFAULT_MAX_OUTPUT_CHARS, + env_passthrough: list[str] | None = None, + **kwargs: Any, + ): + tool_map: dict[str, Tool] = {} + for item in tools or []: + t = _ensure_tool(item) + tool_map[t.name] = t + + default_deps = dependencies + + desc = ( + "Execute Python code in an isolated environment. " + "Captures stdout, stderr, and the value of the last expression as return_value. " + "Dependencies are auto-detected from imports or can be specified explicitly. " + "Exposed tools are callable from the script (code mode)." + ) + tools_desc = _tools_description(tool_map) + if tools_desc: + desc = f"{desc}\n\n{tools_desc}" + + async def _run_python(code: str, dependencies: list[str] | None = None) -> dict[str, Any]: + if len(code) > _MAX_CODE_CHARS: + return { + "stdout": "", + "stderr": "", + "return_value": None, + "status": "error", + "error": { + "type": "ValueError", + "message": f"Code exceeds maximum size of {_MAX_CODE_CHARS} characters", + }, + } + + run_context = get_run_context() + base_dir = run_context.resolve_cwd() if run_context else Path.cwd() + run_dir = Path(tempfile.mkdtemp(prefix=".timbal_run_", dir=base_dir)) + + resolved_deps = _resolve_dependencies(code, dependencies, default_deps) + script_content = _build_script(code, tool_map, resolved_deps) + script_path = run_dir / "script.py" + result_path = run_dir / _RESULT_FILENAME + script_path.write_text(script_content, encoding="utf-8") + + rpc = _ToolRpcServer(tool_map) + await rpc.start() + + # Strip host secrets: only forward a safe baseline + explicit allowlist. + env = _build_child_env(env_passthrough) + env["TIMBAL_RESULT_PATH"] = str(result_path) + if rpc.socket_path: + env[_RPC_ENV_VAR] = rpc.socket_path + + venv_dir = base_dir / ".timbal_run_venv" + venv_python: Path | None = None + chosen_executor: Literal["uv", "venv"] + if executor == "auto": + chosen_executor = "uv" if shutil.which("uv") else "venv" + else: + chosen_executor = executor + + try: + if chosen_executor == "venv" or (chosen_executor == "uv" and not shutil.which("uv")): + venv_python = await _ensure_venv_deps(venv_dir, resolved_deps) + chosen_executor = "venv" + + cmd = _build_command(script_path, chosen_executor, venv_python) + try: + returncode, stdout, stderr = await _run_subprocess( + cmd, + cwd=run_dir, + env=env, + timeout=timeout, + ) + except TimeoutError: + return { + "stdout": "", + "stderr": "", + "return_value": None, + "status": "error", + "error": { + "type": "TimeoutError", + "message": f"Execution exceeded timeout of {timeout}s", + }, + } + + stdout = _truncate_output(stdout, max_output_chars) + stderr = _truncate_output(stderr, max_output_chars) + + payload: dict[str, Any] | None = None + if result_path.exists(): + if result_path.stat().st_size > _MAX_RESULT_BYTES: + return { + "stdout": stdout, + "stderr": stderr, + "return_value": None, + "status": "error", + "error": { + "type": "ValueError", + "message": ( + f"return_value exceeds maximum size of {_MAX_RESULT_BYTES} bytes; " + "write large outputs to a file instead" + ), + }, + } + try: + payload = json.loads(result_path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + payload = None + + if payload and payload.get("error"): + return { + "stdout": stdout, + "stderr": stderr, + "return_value": None, + "status": "error", + "error": payload["error"], + } + + return_value = payload.get("return_value") if payload else None + status = "success" if returncode == 0 else "error" + error = None + if returncode != 0 and not payload: + error = { + "type": "ProcessError", + "message": f"Process exited with code {returncode}", + } + if returncode != 0 and stderr and error is None: + error = {"type": "ProcessError", "message": stderr.strip()} + + return { + "stdout": stdout, + "stderr": stderr, + "return_value": return_value, + "status": status, + "error": error, + } + finally: + await rpc.stop() + shutil.rmtree(run_dir, ignore_errors=True) + + super().__init__( + name="run_python", + description=desc, + handler=_run_python, + default_dependencies=dependencies, + timeout=timeout, + executor=executor, + exposed_tools=tools or [], + max_output_chars=max_output_chars, + env_passthrough=env_passthrough, + background_mode="auto", + **kwargs, + ) + self._tool_map = tool_map + + def get_config(self) -> dict[str, Any]: + """See base class.""" + return { + **super().get_config(), + **self._annotate_config( + { + "default_dependencies": self.default_dependencies, + "timeout": self.timeout, + "executor": self.executor, + "exposed_tools": [t.name for t in self._tool_map.values()], + "max_output_chars": self.max_output_chars, + "env_passthrough": self.env_passthrough, + } + ), + } From 0a07860bac6da5a3d1546111a893fff8c7a0780a Mon Sep 17 00:00:00 2001 From: bpb02 Date: Thu, 25 Jun 2026 16:30:12 +0200 Subject: [PATCH 2/3] refactor(tools): replace RunPython codegen preamble with static runner module Move child-process logic from an embedded _CHILD_PREAMBLE string into _run_python_runner.py (stdlib-only, lintable, unit-testable). The parent now writes user_code.py and passes config via env (TIMBAL_USER_CODE_PATH, TIMBAL_EXPOSED_TOOLS) instead of interpolating code into a generated script. - uv executor uses --with per dependency (no PEP 723 header codegen) - venv executor runs the static runner after pip install - Add 8 unit tests for the runner module in isolation Co-authored-by: Cursor --- python/tests/tools/test_run_python.py | 75 +++++++++ python/timbal/tools/_run_python_runner.py | 183 +++++++++++++++++++++ python/timbal/tools/run_python.py | 185 +++------------------- 3 files changed, 279 insertions(+), 164 deletions(-) create mode 100644 python/timbal/tools/_run_python_runner.py diff --git a/python/tests/tools/test_run_python.py b/python/tests/tools/test_run_python.py index d972d869..674b8308 100644 --- a/python/tests/tools/test_run_python.py +++ b/python/tests/tools/test_run_python.py @@ -587,6 +587,81 @@ async def test_await_in_codemode(self): assert out["return_value"] == 42 +class TestRunPythonRunner: + """Unit tests for the stdlib-only child runner module (no subprocess).""" + + def test_execute_user_code_sync_last_expr(self): + from timbal.tools._run_python_runner import execute_user_code + + out = execute_user_code("1 + 2", {}) + assert out["error"] is None + assert out["return_value"] == 3 + + def test_execute_user_code_sync_exec_only(self): + from timbal.tools._run_python_runner import execute_user_code + + out = execute_user_code("x = 10", {}) + assert out["error"] is None + assert out["return_value"] is None + + def test_execute_user_code_catches_exception(self): + from timbal.tools._run_python_runner import execute_user_code + + out = execute_user_code("1 / 0", {}) + assert out["return_value"] is None + assert out["error"]["type"] == "ZeroDivisionError" + + def test_execute_user_code_non_json_serializable_return(self): + from timbal.tools._run_python_runner import execute_user_code + + out = execute_user_code("object()", {}) + assert out["error"] is None + assert isinstance(out["return_value"], str) + assert "object" in out["return_value"] + + def test_has_top_level_await_detects_await(self): + import ast + + from timbal.tools._run_python_runner import has_top_level_await + + tree = ast.parse("import asyncio\nawait asyncio.sleep(0)") + assert has_top_level_await(tree) is True + + def test_has_top_level_await_ignores_nested(self): + import ast + + from timbal.tools._run_python_runner import has_top_level_await + + tree = ast.parse("async def f():\n await asyncio.sleep(0)") + assert has_top_level_await(tree) is False + + def test_execute_user_code_top_level_await(self): + from timbal.tools._run_python_runner import execute_user_code + + code = """ +import asyncio +await asyncio.sleep(0) +sum(range(4)) +""" + out = execute_user_code(code, {}) + assert out["error"] is None + assert out["return_value"] == 6 + + def test_build_proxies_calls_call_tool(self, monkeypatch: pytest.MonkeyPatch): + from timbal.tools._run_python_runner import build_proxies + + calls: list[tuple[str, list[Any], dict[str, Any]]] = [] + + def fake_call_tool(name: str, args: list[Any], kwargs: dict[str, Any]) -> Any: + calls.append((name, args, kwargs)) + return 42 + + monkeypatch.setattr("timbal.tools._run_python_runner.call_tool", fake_call_tool) + proxies = build_proxies(["double"]) + assert proxies["double"](21) == 42 + assert calls == [("double", [21], {})] + + class TestTimbalInTimbal: """Install timbal inside the sandbox and run an Agent within an Agent. diff --git a/python/timbal/tools/_run_python_runner.py b/python/timbal/tools/_run_python_runner.py new file mode 100644 index 00000000..8f5983d4 --- /dev/null +++ b/python/timbal/tools/_run_python_runner.py @@ -0,0 +1,183 @@ +"""Stdlib-only child runner for RunPython. + +Executed as a standalone script in an isolated subprocess. The parent passes +user code and configuration via environment variables and files — no string +codegen. +""" + +from __future__ import annotations + +import ast +import asyncio +import builtins +import json +import os +import socket +import textwrap +import traceback +from collections.abc import Callable +from pathlib import Path +from typing import Any + +_RPC_ENV_VAR = "TIMBAL_RPC_SOCKET" +_RESULT_PATH_ENV = "TIMBAL_RESULT_PATH" +_USER_CODE_PATH_ENV = "TIMBAL_USER_CODE_PATH" +_EXPOSED_TOOLS_ENV = "TIMBAL_EXPOSED_TOOLS" + + +def call_tool(name: str, args: list[Any], kwargs: dict[str, Any]) -> Any: + """Call an exposed Timbal tool via the parent RPC socket.""" + sock_path = os.environ.get(_RPC_ENV_VAR) + if not sock_path: + raise RuntimeError("Code mode is not available: RPC socket not configured") + payload = json.dumps({"name": name, "args": args, "kwargs": kwargs}).encode() + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.connect(sock_path) + sock.sendall(payload + b"\n") + chunks: list[bytes] = [] + while True: + chunk = sock.recv(65536) + if not chunk: + break + chunks.append(chunk) + if b"\n" in chunk: + break + line = b"".join(chunks).split(b"\n", 1)[0] + response = json.loads(line.decode()) + if response.get("error"): + err = response["error"] + exc_name = err.get("type", "RuntimeError") + message = err.get("message", str(err)) + exc_cls = getattr(builtins, exc_name, None) + if exc_cls is None or not isinstance(exc_cls, type) or not issubclass(exc_cls, BaseException): + if exc_name == "KeyError" and "Unknown tool" in message: + raise NameError(f"name '{name}' is not defined") + exc_cls = RuntimeError + message = f"{exc_name}: {message}" + raise exc_cls(message) + return response.get("output") + + +def write_result(payload: dict[str, Any]) -> None: + """Persist execution result to the path configured by the parent.""" + result_path = os.environ.get(_RESULT_PATH_ENV, "") + if result_path: + with open(result_path, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + + +def has_top_level_await(tree: ast.AST) -> bool: + """Return True if the module tree contains await at module scope.""" + + class _Visitor(ast.NodeVisitor): + def __init__(self) -> None: + self.found = False + + def visit_FunctionDef(self, _node: ast.FunctionDef) -> None: + return None + + def visit_AsyncFunctionDef(self, _node: ast.AsyncFunctionDef) -> None: + return None + + def visit_Lambda(self, _node: ast.Lambda) -> None: + return None + + def visit_Await(self, _node: ast.Await) -> None: + self.found = True + + def visit_AsyncFor(self, node: ast.AsyncFor) -> None: + self.found = True + self.generic_visit(node) + + def visit_AsyncWith(self, node: ast.AsyncWith) -> None: + self.found = True + self.generic_visit(node) + + visitor = _Visitor() + visitor.visit(tree) + return visitor.found + + +def execute_user_code(code: str, namespace: dict[str, Any]) -> dict[str, Any]: + """Execute user code and capture return_value or structured error.""" + result: dict[str, Any] = {"return_value": None, "error": None} + try: + tree = ast.parse(code) + exec_namespace = dict(namespace) + exec_namespace["__name__"] = "__main__" + if has_top_level_await(tree): + if tree.body and isinstance(tree.body[-1], ast.Expr): + ret = ast.Return(value=tree.body[-1].value) + ast.copy_location(ret, tree.body[-1]) + tree.body[-1] = ret + ast.fix_missing_locations(tree) + wrapped = "async def __timbal_async_main__():\n" + textwrap.indent(ast.unparse(tree), " ") + exec(compile(wrapped, "", "exec"), exec_namespace) + result["return_value"] = asyncio.run(exec_namespace["__timbal_async_main__"]()) + elif tree.body and isinstance(tree.body[-1], ast.Expr): + if len(tree.body) > 1: + exec( + compile(ast.Module(body=tree.body[:-1], type_ignores=[]), "", "exec"), + exec_namespace, + ) + last_expr = compile(ast.Expression(body=tree.body[-1].value), "", "eval") + result["return_value"] = eval(last_expr, exec_namespace) + else: + exec(compile(tree, "", "exec"), exec_namespace) + except Exception as exc: + result["error"] = { + "type": type(exc).__name__, + "message": str(exc), + "traceback": traceback.format_exc(), + } + if result["error"] is None: + try: + json.dumps(result["return_value"]) + except (TypeError, ValueError): + result["return_value"] = repr(result["return_value"]) + return result + + +def build_proxies(tool_names: list[str]) -> dict[str, Callable[..., Any]]: + """Build callable proxies for exposed tools (code mode).""" + proxies: dict[str, Callable[..., Any]] = {} + for name in tool_names: + def _make_proxy(tool_name: str) -> Callable[..., Any]: + def proxy(*args: Any, **kwargs: Any) -> Any: + return call_tool(tool_name, list(args), kwargs) + + proxy.__name__ = tool_name + return proxy + + proxies[name] = _make_proxy(name) + return proxies + + +def main() -> None: + """Entry point: read config from env, execute user code, write result.""" + user_code_path = os.environ.get(_USER_CODE_PATH_ENV, "") + if not user_code_path: + write_result( + { + "return_value": None, + "error": { + "type": "RuntimeError", + "message": f"{_USER_CODE_PATH_ENV} is not set", + }, + } + ) + return + + code = Path(user_code_path).read_text(encoding="utf-8") + exposed_raw = os.environ.get(_EXPOSED_TOOLS_ENV, "[]") + try: + tool_names = json.loads(exposed_raw) + except json.JSONDecodeError: + tool_names = [] + + namespace: dict[str, Any] = {"__name__": "__main__", **build_proxies(tool_names)} + write_result(execute_user_code(code, namespace)) + + +if __name__ == "__main__": + main() diff --git a/python/timbal/tools/run_python.py b/python/timbal/tools/run_python.py index 9effe13a..c9826d4a 100644 --- a/python/timbal/tools/run_python.py +++ b/python/timbal/tools/run_python.py @@ -2,7 +2,7 @@ RunPython tool for executing Python scripts in isolated environments. Supports: -- Real CPython execution via uv (PEP 723 inline dependencies) with venv+pip fallback +- Real CPython execution via uv (--with dependencies) with venv+pip fallback - Auto-detection of dependencies from import statements - Code mode: scripts can call back into other Timbal tools via a unix-socket RPC bridge - Structured stdout/stderr/return_value/error capture @@ -43,7 +43,11 @@ } _RESULT_FILENAME = ".timbal_run_result.json" +_USER_CODE_FILENAME = "user_code.py" _RPC_ENV_VAR = "TIMBAL_RPC_SOCKET" +_USER_CODE_PATH_ENV = "TIMBAL_USER_CODE_PATH" +_EXPOSED_TOOLS_ENV = "TIMBAL_EXPOSED_TOOLS" +_RUNNER_PATH = Path(__file__).parent / "_run_python_runner.py" # Hardening limits. _MAX_CODE_CHARS = 200_000 @@ -85,125 +89,6 @@ "PYTHONPATH", }) -# Preamble injected before user code in the child process. -_CHILD_PREAMBLE = ''' -import ast as __timbal_ast -import builtins as __timbal_builtins__ -import json as __timbal_json -import os as __timbal_os -import socket as __timbal_socket -import sys as __timbal_sys -import traceback as __timbal_traceback - -__TIMBAL_RESULT_PATH = __timbal_os.environ.get("TIMBAL_RESULT_PATH", "") - - -def __timbal_call_tool__(name, args, kwargs): - sock_path = __timbal_os.environ.get({rpc_env!r}) - if not sock_path: - raise RuntimeError("Code mode is not available: RPC socket not configured") - payload = __timbal_json.dumps({{"name": name, "args": args, "kwargs": kwargs}}).encode() - with __timbal_socket.socket(__timbal_socket.AF_UNIX, __timbal_socket.SOCK_STREAM) as sock: - sock.connect(sock_path) - sock.sendall(payload + b"\\n") - chunks = [] - while True: - chunk = sock.recv(65536) - if not chunk: - break - chunks.append(chunk) - if b"\\n" in chunk: - break - line = b"".join(chunks).split(b"\\n", 1)[0] - response = __timbal_json.loads(line.decode()) - if response.get("error"): - err = response["error"] - exc_name = err.get("type", "RuntimeError") - message = err.get("message", str(err)) - exc_cls = getattr(__timbal_builtins__, exc_name, None) - if exc_cls is None or not isinstance(exc_cls, type) or not issubclass(exc_cls, BaseException): - if exc_name == "KeyError" and "Unknown tool" in message: - raise NameError(f"name '{{name}}' is not defined") - exc_cls = RuntimeError - message = f"{{exc_name}}: {{message}}" - raise exc_cls(message) - return response.get("output") - - -def __timbal_write_result__(payload): - if __TIMBAL_RESULT_PATH: - with open(__TIMBAL_RESULT_PATH, "w", encoding="utf-8") as fh: - __timbal_json.dump(payload, fh) - - -def __timbal_has_top_level_await__(tree): - class _Visitor(__timbal_ast.NodeVisitor): - def __init__(self): - self.found = False - def visit_FunctionDef(self, node): - return None - def visit_AsyncFunctionDef(self, node): - return None - def visit_Lambda(self, node): - return None - def visit_Await(self, node): - self.found = True - def visit_AsyncFor(self, node): - self.found = True - self.generic_visit(node) - def visit_AsyncWith(self, node): - self.found = True - self.generic_visit(node) - visitor = _Visitor() - visitor.visit(tree) - return visitor.found - - -def __timbal_execute_user_code__(code): - result = {{"return_value": None, "error": None}} - try: - tree = __timbal_ast.parse(code) - namespace = dict(globals()) - namespace["__name__"] = "__main__" - if __timbal_has_top_level_await__(tree): - import asyncio as __timbal_asyncio - import textwrap as __timbal_textwrap - if tree.body and isinstance(tree.body[-1], __timbal_ast.Expr): - __timbal_ret = __timbal_ast.Return(value=tree.body[-1].value) - __timbal_ast.copy_location(__timbal_ret, tree.body[-1]) - tree.body[-1] = __timbal_ret - __timbal_ast.fix_missing_locations(tree) - __timbal_wrapped = ( - "async def __timbal_async_main__():\\n" - + __timbal_textwrap.indent(__timbal_ast.unparse(tree), " ") - ) - exec(compile(__timbal_wrapped, "", "exec"), namespace) - result["return_value"] = __timbal_asyncio.run(namespace["__timbal_async_main__"]()) - elif tree.body and isinstance(tree.body[-1], __timbal_ast.Expr): - if len(tree.body) > 1: - exec( - compile(__timbal_ast.Module(body=tree.body[:-1], type_ignores=[]), "", "exec"), - namespace, - ) - last_expr = compile(__timbal_ast.Expression(body=tree.body[-1].value), "", "eval") - result["return_value"] = eval(last_expr, namespace) - else: - exec(compile(tree, "", "exec"), namespace) - except Exception as exc: - result["error"] = {{ - "type": type(exc).__name__, - "message": str(exc), - "traceback": __timbal_traceback.format_exc(), - }} - if result["error"] is None: - try: - __timbal_json.dumps(result["return_value"]) - except (TypeError, ValueError): - result["return_value"] = repr(result["return_value"]) - return result -'''.format(rpc_env=_RPC_ENV_VAR) # noqa: UP032 - - def _build_child_env(passthrough: list[str] | None) -> dict[str, str]: """Build a minimal environment for the child process. @@ -305,41 +190,6 @@ def add(dep: str) -> None: return resolved -def _build_pep723_header(dependencies: list[str]) -> str: - if not dependencies: - return "" - lines = ["# /// script", "# dependencies = ["] - for dep in dependencies: - lines.append(f'# "{dep}",') - lines.append("# ]") - lines.append("# ///") - return "\n".join(lines) + "\n" - - -def _build_proxy_functions(tools: dict[str, Tool]) -> str: - lines: list[str] = [] - for name in tools: - if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name): - continue - lines.append( - f"def {name}(*args, **kwargs):\n" - f' return __timbal_call_tool__("{name}", list(args), kwargs)\n' - ) - return "\n".join(lines) - - -def _build_script(code: str, tools: dict[str, Tool], dependencies: list[str]) -> str: - header = _build_pep723_header(dependencies) - proxies = _build_proxy_functions(tools) - proxy_block = f"\n{proxies}\n" if proxies else "\n" - body = f''' -{proxy_block} -__timbal_result__ = __timbal_execute_user_code__({json.dumps(code)}) -__timbal_write_result__(__timbal_result__) -''' - return header + _CHILD_PREAMBLE + body - - def _serialize_output(value: Any) -> Any: try: json.dumps(value) @@ -516,14 +366,20 @@ async def _run_subprocess( return process.returncode or 0, stdout, stderr -def _build_command(script_path: Path, executor: Literal["uv", "venv"], venv_python: Path | None) -> list[str]: +def _build_command( + dependencies: list[str], + executor: Literal["uv", "venv"], + venv_python: Path | None, +) -> list[str]: + runner = str(_RUNNER_PATH) if executor == "uv": uv = shutil.which("uv") if uv: - return [uv, "run", "--no-project", str(script_path)] + cmd = [uv, "run", "--no-project", *(f"--with={dep}" for dep in dependencies), runner] + return cmd if venv_python is not None: - return [str(venv_python), str(script_path)] - return [sys.executable, str(script_path)] + return [str(venv_python), runner] + return [sys.executable, runner] async def _ensure_venv_deps(venv_dir: Path, dependencies: list[str]) -> Path: @@ -580,7 +436,7 @@ class RunPython(Tool): timeout: float = Field(default=60.0, description="Maximum execution time in seconds.") executor: Literal["uv", "venv", "auto"] = Field( default="auto", - description="Execution backend: uv (PEP 723), venv+pip fallback, or auto (prefer uv).", + description="Execution backend: uv (--with deps), venv+pip fallback, or auto (prefer uv).", ) exposed_tools: list[Any] = Field( default_factory=list, @@ -652,10 +508,9 @@ async def _run_python(code: str, dependencies: list[str] | None = None) -> dict[ run_dir = Path(tempfile.mkdtemp(prefix=".timbal_run_", dir=base_dir)) resolved_deps = _resolve_dependencies(code, dependencies, default_deps) - script_content = _build_script(code, tool_map, resolved_deps) - script_path = run_dir / "script.py" + user_code_path = run_dir / _USER_CODE_FILENAME result_path = run_dir / _RESULT_FILENAME - script_path.write_text(script_content, encoding="utf-8") + user_code_path.write_text(code, encoding="utf-8") rpc = _ToolRpcServer(tool_map) await rpc.start() @@ -663,6 +518,8 @@ async def _run_python(code: str, dependencies: list[str] | None = None) -> dict[ # Strip host secrets: only forward a safe baseline + explicit allowlist. env = _build_child_env(env_passthrough) env["TIMBAL_RESULT_PATH"] = str(result_path) + env[_USER_CODE_PATH_ENV] = str(user_code_path) + env[_EXPOSED_TOOLS_ENV] = json.dumps(list(tool_map.keys())) if rpc.socket_path: env[_RPC_ENV_VAR] = rpc.socket_path @@ -679,7 +536,7 @@ async def _run_python(code: str, dependencies: list[str] | None = None) -> dict[ venv_python = await _ensure_venv_deps(venv_dir, resolved_deps) chosen_executor = "venv" - cmd = _build_command(script_path, chosen_executor, venv_python) + cmd = _build_command(resolved_deps, chosen_executor, venv_python) try: returncode, stdout, stderr = await _run_subprocess( cmd, From 17930512d51f2dfdcb8424925a0309c694fa0509 Mon Sep 17 00:00:00 2001 From: bpb02 Date: Fri, 26 Jun 2026 09:31:52 +0200 Subject: [PATCH 3/3] Fix RunPython RPC bridge on Windows. Use a localhost TCP server when asyncio.start_unix_server is unavailable (Windows CI). The child runner connects via tcp://host:port or unix path. Add a regression test that simulates the Windows transport. Co-authored-by: Cursor --- python/tests/tools/test_run_python.py | 15 ++++++++++++ python/timbal/tools/_run_python_runner.py | 18 ++++++++++++-- python/timbal/tools/run_python.py | 29 ++++++++++++++++------- 3 files changed, 51 insertions(+), 11 deletions(-) diff --git a/python/tests/tools/test_run_python.py b/python/tests/tools/test_run_python.py index 674b8308..e54f9c44 100644 --- a/python/tests/tools/test_run_python.py +++ b/python/tests/tools/test_run_python.py @@ -662,6 +662,21 @@ def fake_call_tool(name: str, args: list[Any], kwargs: dict[str, Any]) -> Any: assert calls == [("double", [21], {})] +class TestRunPythonRpcTransport: + @pytest.mark.asyncio + async def test_code_mode_uses_tcp_when_unix_server_unavailable(self, monkeypatch): + """Regression: Windows lacks asyncio.start_unix_server.""" + import asyncio + + if not hasattr(asyncio, "start_unix_server"): + pytest.skip("platform already lacks unix server") + + monkeypatch.delattr(asyncio, "start_unix_server", raising=False) + tool = RunPython(tools=[noop], executor=EXECUTOR) + out = await collect(tool, "noop()") + assert out["return_value"] == "called" + + class TestTimbalInTimbal: """Install timbal inside the sandbox and run an Agent within an Agent. diff --git a/python/timbal/tools/_run_python_runner.py b/python/timbal/tools/_run_python_runner.py index 8f5983d4..15d1d6b5 100644 --- a/python/timbal/tools/_run_python_runner.py +++ b/python/timbal/tools/_run_python_runner.py @@ -25,14 +25,28 @@ _EXPOSED_TOOLS_ENV = "TIMBAL_EXPOSED_TOOLS" +def _open_rpc_connection(endpoint: str) -> socket.socket: + """Connect to the parent RPC server (unix path or ``tcp://host:port``).""" + if endpoint.startswith("tcp://"): + hostport = endpoint.removeprefix("tcp://") + host, port_str = hostport.rsplit(":", 1) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((host, int(port_str))) + return sock + if not hasattr(socket, "AF_UNIX"): + raise RuntimeError(f"Unsupported RPC endpoint on this platform: {endpoint!r}") + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(endpoint) + return sock + + def call_tool(name: str, args: list[Any], kwargs: dict[str, Any]) -> Any: """Call an exposed Timbal tool via the parent RPC socket.""" sock_path = os.environ.get(_RPC_ENV_VAR) if not sock_path: raise RuntimeError("Code mode is not available: RPC socket not configured") payload = json.dumps({"name": name, "args": args, "kwargs": kwargs}).encode() - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: - sock.connect(sock_path) + with _open_rpc_connection(sock_path) as sock: sock.sendall(payload + b"\n") chunks: list[bytes] = [] while True: diff --git a/python/timbal/tools/run_python.py b/python/timbal/tools/run_python.py index c9826d4a..199ee796 100644 --- a/python/timbal/tools/run_python.py +++ b/python/timbal/tools/run_python.py @@ -4,7 +4,7 @@ Supports: - Real CPython execution via uv (--with dependencies) with venv+pip fallback - Auto-detection of dependencies from import statements -- Code mode: scripts can call back into other Timbal tools via a unix-socket RPC bridge +- Code mode: scripts can call back into other Timbal tools via a localhost RPC bridge - Structured stdout/stderr/return_value/error capture """ @@ -248,7 +248,11 @@ async def _call_handler(**call_kwargs: Any) -> Any: class _ToolRpcServer: - """Async unix-socket RPC server dispatching tool calls from child process.""" + """Async RPC server dispatching tool calls from the child process. + + Uses a unix domain socket on POSIX and a localhost TCP socket on Windows + (``asyncio.start_unix_server`` is not available there). + """ def __init__(self, tools: dict[str, Tool]) -> None: self._tools = tools @@ -262,20 +266,27 @@ def socket_path(self) -> str | None: async def start(self) -> None: if not self._tools: return - tmp = tempfile.NamedTemporaryFile(prefix="timbal-rpc-", suffix=".sock", delete=False) - tmp.close() - self._socket_path = tmp.name - Path(self._socket_path).unlink(missing_ok=True) - self._server = await asyncio.start_unix_server(self._handle_client, path=self._socket_path) + if hasattr(asyncio, "start_unix_server"): + tmp = tempfile.NamedTemporaryFile(prefix="timbal-rpc-", suffix=".sock", delete=False) + tmp.close() + self._socket_path = tmp.name + Path(self._socket_path).unlink(missing_ok=True) + self._server = await asyncio.start_unix_server(self._handle_client, path=self._socket_path) + return + + self._server = await asyncio.start_server(self._handle_client, "127.0.0.1", 0) + sock = self._server.sockets[0] + host, port = sock.getsockname()[:2] + self._socket_path = f"tcp://{host}:{port}" async def stop(self) -> None: if self._server is not None: self._server.close() await self._server.wait_closed() self._server = None - if self._socket_path: + if self._socket_path and not self._socket_path.startswith("tcp://"): Path(self._socket_path).unlink(missing_ok=True) - self._socket_path = None + self._socket_path = None async def _handle_client( self,