yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#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 SyncAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#7
#8	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#9
#10
#11	class WriteTextFileArgs(BaseModel):
#12	path: str = Field(description="Relative file path under the runtime workspace root.")
#13	content: str = Field(description="Exact UTF-8 text content to write.")
#14	encoding: str = Field(default=_TOOL_DEFAULTS.default_text_encoding, description="Text encoding.")
#15	overwrite: bool = Field(default=True, description="Whether to overwrite an existing file.")
#16
#17
#18	class WriteTextFileOutput(BaseModel):
#19	path: str
#20	bytes_written: int
#21	created: bool
#22
#23
#24	class ReadTextFileArgs(BaseModel):
#25	path: str = Field(description="Relative file path under the runtime workspace root.")
#26	encoding: str = Field(default=_TOOL_DEFAULTS.default_text_encoding, description="Text encoding.")
#27	max_bytes: int = Field(
#28	default=_TOOL_DEFAULTS.filesystem_read_max_bytes,
#29	ge=1,
#30	le=_TOOL_DEFAULTS.filesystem_read_hard_limit_bytes,
#31	description="Maximum bytes to read.",
#32	)
#33
#34
#35	class ReadTextFileOutput(BaseModel):
#36	path: str
#37	content: str
#38	bytes_read: int
#39	truncated: bool
#40
#41
#42	class DirectoryEntryOutput(BaseModel):
#43	name: str
#44	path: str
#45	kind: str
#46	size_bytes: int \| None
#47	modified_at: str
#48
#49
#50	class ReadDirectoryArgs(BaseModel):
#51	path: str = Field(description="Relative directory path under the runtime workspace root.")
#52	limit: int = Field(
#53	default=_TOOL_DEFAULTS.directory_entry_limit,
#54	ge=1,
#55	le=_TOOL_DEFAULTS.directory_entry_hard_limit,
#56	description="Maximum number of entries to return.",
#57	)
#58
#59
#60	class ReadDirectoryOutput(BaseModel):
#61	path: str
#62	entries: list[DirectoryEntryOutput]
#63	count: int
#64	truncated: bool
#65
#66
#67	class WriteDirectoryArgs(BaseModel):
#68	path: str = Field(description="Relative directory path under the runtime workspace root.")
#69	parents: bool = Field(default=True, description="Whether to create missing parent directories.")
#70	exist_ok: bool = Field(default=True, description="Whether an existing directory is accepted.")
#71
#72
#73	class WriteDirectoryOutput(BaseModel):
#74	path: str
#75	created: bool
#76
#77
#78	class DeleteFileArgs(BaseModel):
#79	path: str = Field(description="Relative file path under the runtime workspace root.")
#80	missing_ok: bool = Field(default=False, description="Whether a missing file should be treated as success.")
#81
#82
#83	class DeleteDirectoryArgs(BaseModel):
#84	path: str = Field(description="Relative directory path under the runtime workspace root.")
#85	recursive: bool = Field(default=False, description="Whether to delete a non-empty directory recursively.")
#86	missing_ok: bool = Field(default=False, description="Whether a missing directory should be treated as success.")
#87
#88
#89	class DeletePathOutput(BaseModel):
#90	path: str
#91	kind: str
#92	deleted: bool
#93	recursive: bool = False
#94
#95
#96	class ReadTextFileTool(SyncAgentTool[ReadTextFileArgs]):
#97	name = "read_text_file"
#98	description = (
#99	"Read UTF-8 text from a file under the runtime workspace root. "
#100	"This is a Skills/Tools Layer wrapper around the libOS filesystem primitive; "
#101	"the primitive enforces filesystem read capability, path containment, audit, and events."
#102	)
#103	args_schema = ReadTextFileArgs
#104	output_schema = ReadTextFileOutput
#105	policy = ToolPolicy(
#106	side_effects=False,
#107	idempotent=True,
#108	requires_confirmation=False,
#109	permissions={"filesystem.read"},
#110	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#111	)
#112	tags = ["filesystem", "workspace", "read"]
#113
#114	def run(self, args: ReadTextFileArgs, ctx: ToolContext) -> ReadTextFileOutput:
#115	runtime = ctx.runtime
#116	if runtime is None:
#117	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#118	cwd = runtime.process.working_directory(ctx.pid)
#119	try:
#120	result = runtime.filesystem.read_text(
#121	pid=ctx.pid,
#122	path=args.path,
#123	encoding=args.encoding,
#124	max_bytes=args.max_bytes,
#125	cwd=cwd,
#126	)
#127	except UnicodeDecodeError as exc:
#128	raise ToolExecutionError(
#129	"File could not be decoded with the requested encoding.",
#130	code=ToolErrorCode.EXECUTION_ERROR,
#131	details={"encoding": args.encoding, "error": str(exc)},
#132	) from exc
#133	return ReadTextFileOutput(
#134	path=result.path,
#135	content=result.content,
#136	bytes_read=result.bytes_read,
#137	truncated=result.truncated,
#138	)
#139
#140
#141	class ReadDirectoryTool(SyncAgentTool[ReadDirectoryArgs]):
#142	name = "read_directory"
#143	description = (
#144	"List entries in a directory under the runtime workspace root. "
#145	"The filesystem primitive enforces directory read capability, path containment, audit, and events."
#146	)
#147	args_schema = ReadDirectoryArgs
#148	output_schema = ReadDirectoryOutput
#149	policy = ToolPolicy(
#150	side_effects=False,
#151	idempotent=True,
#152	requires_confirmation=False,
#153	permissions={"filesystem.read"},
#154	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#155	)
#156	tags = ["filesystem", "workspace", "read", "directory"]
#157
#158	def run(self, args: ReadDirectoryArgs, ctx: ToolContext) -> ReadDirectoryOutput:
#159	runtime = ctx.runtime
#160	if runtime is None:
#161	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#162	cwd = runtime.process.working_directory(ctx.pid)
#163	result = runtime.filesystem.read_directory(
#164	pid=ctx.pid,
#165	path=args.path,
#166	limit=args.limit,
#167	cwd=cwd,
#168	)
#169	return ReadDirectoryOutput(
#170	path=result.path,
#171	entries=[DirectoryEntryOutput(**entry.__dict__) for entry in result.entries],
#172	count=result.count,
#173	truncated=result.truncated,
#174	)
#175
#176
#177	class WriteTextFileTool(SyncAgentTool[WriteTextFileArgs]):
#178	name = "write_text_file"
#179	description = (
#180	"Write UTF-8 text to a file under the runtime workspace root. "
#181	"This is a Skills/Tools Layer wrapper around the libOS filesystem primitive; "
#182	"the primitive enforces filesystem write capability, path containment, audit, and events."
#183	)
#184	args_schema = WriteTextFileArgs
#185	output_schema = WriteTextFileOutput
#186	policy = ToolPolicy(
#187	side_effects=True,
#188	idempotent=False,
#189	requires_confirmation=True,
#190	permissions={"filesystem.write"},
#191	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#192	)
#193	tags = ["filesystem", "workspace", "side_effect"]
#194
#195	def run(self, args: WriteTextFileArgs, ctx: ToolContext) -> WriteTextFileOutput:
#196	runtime = ctx.runtime
#197	if runtime is None:
#198	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#199	cwd = runtime.process.working_directory(ctx.pid)
#200	try:
#201	result = runtime.filesystem.write_text(
#202	pid=ctx.pid,
#203	path=args.path,
#204	text=args.content,
#205	encoding=args.encoding,
#206	overwrite=args.overwrite,
#207	cwd=cwd,
#208	)
#209	except FileExistsError as exc:
#210	raise ToolExecutionError(
#211	"File already exists and overwrite is false.",
#212	code=ToolErrorCode.EXECUTION_ERROR,
#213	details={"path": args.path},
#214	) from exc
#215	return WriteTextFileOutput(
#216	path=result.path,
#217	bytes_written=result.bytes_written,
#218	created=result.created,
#219	)
#220
#221
#222	class WriteDirectoryTool(SyncAgentTool[WriteDirectoryArgs]):
#223	name = "write_directory"
#224	description = (
#225	"Create or ensure a directory under the runtime workspace root. "
#226	"The filesystem primitive enforces directory write capability, path containment, audit, and events."
#227	)
#228	args_schema = WriteDirectoryArgs
#229	output_schema = WriteDirectoryOutput
#230	policy = ToolPolicy(
#231	side_effects=True,
#232	idempotent=False,
#233	requires_confirmation=True,
#234	permissions={"filesystem.write"},
#235	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#236	)
#237	tags = ["filesystem", "workspace", "side_effect", "directory"]
#238
#239	def run(self, args: WriteDirectoryArgs, ctx: ToolContext) -> WriteDirectoryOutput:
#240	runtime = ctx.runtime
#241	if runtime is None:
#242	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#243	cwd = runtime.process.working_directory(ctx.pid)
#244	try:
#245	result = runtime.filesystem.write_directory(
#246	pid=ctx.pid,
#247	path=args.path,
#248	parents=args.parents,
#249	exist_ok=args.exist_ok,
#250	cwd=cwd,
#251	)
#252	except FileExistsError as exc:
#253	raise ToolExecutionError(
#254	"Directory already exists and exist_ok is false.",
#255	code=ToolErrorCode.EXECUTION_ERROR,
#256	details={"path": args.path},
#257	) from exc
#258	return WriteDirectoryOutput(path=result.path, created=result.created)
#259
#260
#261	class DeleteFileTool(SyncAgentTool[DeleteFileArgs]):
#262	name = "delete_file"
#263	description = (
#264	"Delete a file under the runtime workspace root. "
#265	"The filesystem primitive enforces delete capability, path containment, audit, and events."
#266	)
#267	args_schema = DeleteFileArgs
#268	output_schema = DeletePathOutput
#269	policy = ToolPolicy(
#270	side_effects=True,
#271	idempotent=False,
#272	requires_confirmation=True,
#273	permissions={"filesystem.delete"},
#274	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#275	)
#276	tags = ["filesystem", "workspace", "side_effect", "delete"]
#277
#278	def run(self, args: DeleteFileArgs, ctx: ToolContext) -> DeletePathOutput:
#279	runtime = ctx.runtime
#280	if runtime is None:
#281	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#282	cwd = runtime.process.working_directory(ctx.pid)
#283	result = runtime.filesystem.delete_file(
#284	pid=ctx.pid,
#285	path=args.path,
#286	missing_ok=args.missing_ok,
#287	cwd=cwd,
#288	)
#289	return DeletePathOutput(
#290	path=result.path,
#291	kind=result.kind,
#292	deleted=result.deleted,
#293	recursive=result.recursive,
#294	)
#295
#296
#297	class DeleteDirectoryTool(SyncAgentTool[DeleteDirectoryArgs]):
#298	name = "delete_directory"
#299	description = (
#300	"Delete a directory under the runtime workspace root. "
#301	"The filesystem primitive enforces delete capability, path containment, audit, and events."
#302	)
#303	args_schema = DeleteDirectoryArgs
#304	output_schema = DeletePathOutput
#305	policy = ToolPolicy(
#306	side_effects=True,
#307	idempotent=False,
#308	requires_confirmation=True,
#309	permissions={"filesystem.delete"},
#310	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#311	)
#312	tags = ["filesystem", "workspace", "side_effect", "delete", "directory"]
#313
#314	def run(self, args: DeleteDirectoryArgs, ctx: ToolContext) -> DeletePathOutput:
#315	runtime = ctx.runtime
#316	if runtime is None:
#317	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#318	cwd = runtime.process.working_directory(ctx.pid)
#319	try:
#320	result = runtime.filesystem.delete_directory(
#321	pid=ctx.pid,
#322	path=args.path,
#323	recursive=args.recursive,
#324	missing_ok=args.missing_ok,
#325	cwd=cwd,
#326	)
#327	except OSError as exc:
#328	raise ToolExecutionError(
#329	"Directory could not be deleted.",
#330	code=ToolErrorCode.EXECUTION_ERROR,
#331	details={"path": args.path, "error": str(exc)},
#332	) from exc
#333	return DeletePathOutput(
#334	path=result.path,
#335	kind=result.kind,
#336	deleted=result.deleted,
#337	recursive=result.recursive,
#338	)
#339

z6MkqRzA/yingqi-z20-Agent-libOS