yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	from typing import Any
#4
#5	from pydantic import BaseModel, Field
#6
#7	from agent_libos.config import DEFAULT_CONFIG
#8	from agent_libos.models import ObjectMetadata, ObjectType
#9	from agent_libos.tools.base import SyncAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#10
#11	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#12
#13
#14	class CreateObjectFromFileArgs(BaseModel):
#15	name: str = Field(description="Namespace-local Object Memory name to create.")
#16	namespace: str \| None = Field(default=None, description="Object Memory namespace. Defaults to this process namespace.")
#17	path: str = Field(description="Workspace-relative file path to import.")
#18	encoding: str = Field(default=_TOOL_DEFAULTS.default_text_encoding, description="Text encoding.")
#19	max_bytes: int = Field(
#20	default=_TOOL_DEFAULTS.object_file_max_bytes,
#21	ge=1,
#22	le=_TOOL_DEFAULTS.object_file_hard_limit_bytes,
#23	description="Maximum bytes to import.",
#24	)
#25	allow_truncated: bool = Field(default=False, description="Whether to create the object if the file is truncated.")
#26	object_type: str = Field(default=ObjectType.ARTIFACT.value, description="ObjectType for the created object.")
#27
#28
#29	class CreateObjectFromFileOutput(BaseModel):
#30	oid: str
#31	namespace: str
#32	name: str
#33	type: str
#34	source_path: str
#35	bytes_read: int
#36	truncated: bool
#37
#38
#39	class WriteObjectToFileArgs(BaseModel):
#40	name: str = Field(description="Namespace-local Object Memory name to resolve and write.")
#41	namespace: str \| None = Field(default=None, description="Object Memory namespace. Defaults to this process namespace.")
#42	path: str = Field(description="Workspace-relative output file path.")
#43	encoding: str = Field(default=_TOOL_DEFAULTS.default_text_encoding, description="Text encoding.")
#44	overwrite: bool = Field(default=True, description="Whether to overwrite an existing file.")
#45
#46
#47	class WriteObjectToFileOutput(BaseModel):
#48	oid: str
#49	namespace: str
#50	name: str
#51	path: str
#52	bytes_written: int
#53	created: bool
#54
#55
#56	class CreateObjectFromFileTool(SyncAgentTool[CreateObjectFromFileArgs]):
#57	name = "create_object_from_file"
#58	description = (
#59	"Create a named Object Memory object whose payload is the text content of a workspace file. "
#60	"The file content is stored inside Object Memory but is not returned in the tool result."
#61	)
#62	args_schema = CreateObjectFromFileArgs
#63	output_schema = CreateObjectFromFileOutput
#64	policy = ToolPolicy(
#65	side_effects=True,
#66	idempotent=False,
#67	permissions={"filesystem.read", "object.write"},
#68	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#69	)
#70	tags = ["memory", "filesystem", "object"]
#71
#72	def run(self, args: CreateObjectFromFileArgs, ctx: ToolContext) -> CreateObjectFromFileOutput:
#73	runtime = ctx.runtime
#74	if runtime is None:
#75	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#76	cwd = runtime.process.working_directory(ctx.pid)
#77	try:
#78	result = runtime.filesystem.read_text(
#79	pid=ctx.pid,
#80	path=args.path,
#81	encoding=args.encoding,
#82	max_bytes=args.max_bytes,
#83	cwd=cwd,
#84	)
#85	except UnicodeDecodeError as exc:
#86	raise ToolExecutionError(
#87	"File could not be decoded with the requested encoding.",
#88	code=ToolErrorCode.EXECUTION_ERROR,
#89	details={"encoding": args.encoding, "path": args.path},
#90	) from exc
#91	if result.truncated and not args.allow_truncated:
#92	raise ToolExecutionError(
#93	"File exceeded max_bytes; no object was created.",
#94	code=ToolErrorCode.EXECUTION_ERROR,
#95	details={"path": result.path, "bytes_read": result.bytes_read, "max_bytes": args.max_bytes},
#96	)
#97	# The content moves into Object Memory, but the tool result exposes only
#98	# metadata so a process can copy files without seeing the bytes.
#99	payload = {
#100	"kind": "workspace_text_file",
#101	"source_path": result.path,
#102	"encoding": args.encoding,
#103	"content": result.content,
#104	"bytes_read": result.bytes_read,
#105	"truncated": result.truncated,
#106	}
#107	handle = runtime.memory.create_object(
#108	pid=ctx.pid,
#109	object_type=ObjectType(args.object_type),
#110	payload=payload,
#111	metadata=ObjectMetadata(
#112	title=args.name,
#113	tags=["file_object", "workspace_file"],
#114	mime_type="text/plain",
#115	token_estimate=0,
#116	),
#117	immutable=True,
#118	name=args.name,
#119	namespace=args.namespace,
#120	)
#121	obj = runtime.memory.get_object(ctx.pid, handle)
#122	return CreateObjectFromFileOutput(
#123	oid=handle.oid,
#124	namespace=obj.namespace,
#125	name=obj.name,
#126	type=args.object_type,
#127	source_path=result.path,
#128	bytes_read=result.bytes_read,
#129	truncated=result.truncated,
#130	)
#131
#132
#133	class WriteObjectToFileTool(SyncAgentTool[WriteObjectToFileArgs]):
#134	name = "write_object_to_file"
#135	description = (
#136	"Resolve a named Object Memory object and write its text content to a workspace file. "
#137	"The object content is not returned in the tool result."
#138	)
#139	args_schema = WriteObjectToFileArgs
#140	output_schema = WriteObjectToFileOutput
#141	policy = ToolPolicy(
#142	side_effects=True,
#143	idempotent=False,
#144	permissions={"filesystem.write", "object.read"},
#145	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#146	)
#147	tags = ["memory", "filesystem", "object", "side_effect"]
#148
#149	def run(self, args: WriteObjectToFileArgs, ctx: ToolContext) -> WriteObjectToFileOutput:
#150	runtime = ctx.runtime
#151	if runtime is None:
#152	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#153	obj = runtime.memory.get_object_by_name(ctx.pid, args.name, namespace=args.namespace)
#154	text = self._extract_text(obj.payload)
#155	# The object payload is handed directly to the filesystem primitive; the
#156	# process-visible result below still omits the concrete content.
#157	try:
#158	result = runtime.filesystem.write_text(
#159	pid=ctx.pid,
#160	path=args.path,
#161	text=text,
#162	encoding=args.encoding,
#163	overwrite=args.overwrite,
#164	cwd=runtime.process.working_directory(ctx.pid),
#165	)
#166	except FileExistsError as exc:
#167	raise ToolExecutionError(
#168	"File already exists and overwrite is false.",
#169	code=ToolErrorCode.EXECUTION_ERROR,
#170	details={"path": args.path},
#171	) from exc
#172	return WriteObjectToFileOutput(
#173	oid=obj.oid,
#174	namespace=obj.namespace,
#175	name=obj.name,
#176	path=result.path,
#177	bytes_written=result.bytes_written,
#178	created=result.created,
#179	)
#180
#181	def _extract_text(self, payload: Any) -> str:
#182	if isinstance(payload, str):
#183	return payload
#184	if isinstance(payload, dict) and isinstance(payload.get("content"), str):
#185	return payload["content"]
#186	raise ToolExecutionError(
#187	"Object payload does not contain text content.",
#188	code=ToolErrorCode.EXECUTION_ERROR,
#189	details={"expected": "string payload or dict content string"},
#190	)
#191

z6MkqRzA/yingqi-z20-Agent-libOS