yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

yingqi-z20-Agent-libOS — gitlawb

#1	from __future__ import annotations
#2
#3	from pydantic import BaseModel, Field
#4
#5	from agent_libos.config import DEFAULT_CONFIG
#6	from agent_libos.tools.base import BaseAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#7
#8	_SHELL_DEFAULTS = DEFAULT_CONFIG.shell
#9	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#10
#11
#12	class RunShellCommandArgs(BaseModel):
#13	argv: list[str] = Field(
#14	min_length=1,
#15	description="Command argv array. Shell strings are not accepted.",
#16	)
#17	timeout_s: float = Field(
#18	default=_TOOL_DEFAULTS.shell_timeout_s,
#19	gt=0,
#20	le=_SHELL_DEFAULTS.timeout_hard_limit_s,
#21	description="Command timeout in seconds.",
#22	)
#23	max_stdout_chars: int = Field(
#24	default=_SHELL_DEFAULTS.max_stdout_chars,
#25	ge=0,
#26	le=_SHELL_DEFAULTS.stdout_hard_limit_chars,
#27	description="Maximum stdout characters returned in the tool result.",
#28	)
#29	max_stderr_chars: int = Field(
#30	default=_SHELL_DEFAULTS.max_stderr_chars,
#31	ge=0,
#32	le=_SHELL_DEFAULTS.stderr_hard_limit_chars,
#33	description="Maximum stderr characters returned in the tool result.",
#34	)
#35
#36
#37	class RunShellCommandOutput(BaseModel):
#38	argv: list[str]
#39	returncode: int
#40	stdout: str
#41	stderr: str
#42	stdout_truncated: bool
#43	stderr_truncated: bool
#44
#45
#46	class RunShellCommandTool(BaseAgentTool[RunShellCommandArgs]):
#47	name = "run_shell_command"
#48	description = (
#49	"Run an argv-only command through the libOS shell primitive. "
#50	"The primitive enforces shell execution policy, configured allow/ask lists, human approval, audit, and events."
#51	)
#52	args_schema = RunShellCommandArgs
#53	output_schema = RunShellCommandOutput
#54	policy = ToolPolicy(
#55	side_effects=True,
#56	idempotent=False,
#57	permissions={"shell.execute"},
#58	timeout_s=None,
#59	)
#60	tags = ["shell", "external", "side_effect"]
#61
#62	async def execute(self, args: RunShellCommandArgs, ctx: ToolContext) -> RunShellCommandOutput:
#63	runtime = ctx.runtime
#64	if runtime is None:
#65	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#66	cwd = runtime.process.working_directory(ctx.pid)
#67	try:
#68	result = await runtime.shell.arun(ctx.pid, args.argv, timeout=args.timeout_s, cwd=cwd)
#69	except TimeoutError as exc:
#70	raise ToolExecutionError(
#71	"Shell command timed out.",
#72	code=ToolErrorCode.TIMEOUT,
#73	retryable=True,
#74	details={"argv": args.argv, "timeout_s": args.timeout_s},
#75	) from exc
#76	stdout, stdout_truncated = _truncate(result.stdout, args.max_stdout_chars)
#77	stderr, stderr_truncated = _truncate(result.stderr, args.max_stderr_chars)
#78	return RunShellCommandOutput(
#79	argv=result.argv,
#80	returncode=result.returncode,
#81	stdout=stdout,
#82	stderr=stderr,
#83	stdout_truncated=result.stdout_truncated or stdout_truncated,
#84	stderr_truncated=result.stderr_truncated or stderr_truncated,
#85	)
#86
#87
#88	def _truncate(value: str, limit: int) -> tuple[str, bool]:
#89	if len(value) <= limit:
#90	return value, False
#91	return value[:limit], True
#92

z6MkqRzA/yingqi-z20-Agent-libOS