mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
Default sandbox exec commands to Bash (#764)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
This commit is contained in:
@@ -209,6 +209,13 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
|
||||
invoke_tool = tool.on_invoke_tool
|
||||
|
||||
async def invoke(ctx: Any, raw_input: str) -> Any:
|
||||
try:
|
||||
parsed = json.loads(raw_input)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
parsed = None
|
||||
if isinstance(parsed, dict) and "shell" not in parsed:
|
||||
parsed["shell"] = "bash"
|
||||
raw_input = json.dumps(parsed)
|
||||
try:
|
||||
return await invoke_tool(ctx, raw_input)
|
||||
except ValidationError as exc:
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
"""Tests for the shell tool adapters in the agent factory."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from agents.tool import FunctionTool
|
||||
|
||||
from strix.agents import factory
|
||||
|
||||
|
||||
def _capturing_exec_tool(captured: dict[str, str]) -> FunctionTool:
|
||||
async def invoke(_ctx: Any, raw_input: str) -> str:
|
||||
captured["raw_input"] = raw_input
|
||||
return "ok"
|
||||
|
||||
return FunctionTool(
|
||||
name="exec_command",
|
||||
description="test tool",
|
||||
params_json_schema={"type": "object", "properties": {}},
|
||||
on_invoke_tool=invoke,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wrap_exec_command_defaults_shell_to_bash() -> None:
|
||||
captured: dict[str, str] = {}
|
||||
wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
|
||||
|
||||
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "source /tmp/env"}))
|
||||
|
||||
assert result == "ok"
|
||||
assert json.loads(captured["raw_input"]) == {
|
||||
"cmd": "source /tmp/env",
|
||||
"shell": "bash",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("shell", ["/bin/zsh", ""])
|
||||
async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
|
||||
captured: dict[str, str] = {}
|
||||
wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
|
||||
|
||||
await wrapped.on_invoke_tool(
|
||||
cast("Any", None), json.dumps({"cmd": "echo test", "shell": shell})
|
||||
)
|
||||
|
||||
assert json.loads(captured["raw_input"])["shell"] == shell
|
||||
Reference in New Issue
Block a user