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.models.exceptions import ValidationError as LibOSValidationError
#7	from agent_libos.tools.base import SyncAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#8
#9	_IMAGE_DEFAULTS = DEFAULT_CONFIG.image
#10	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#11
#12
#13	class LoadImageFromYamlArgs(BaseModel):
#14	path: str = Field(description="Workspace-relative YAML file path containing one AgentImage manifest.")
#15	encoding: str = Field(default=_TOOL_DEFAULTS.default_text_encoding, description="Text encoding.")
#16	max_bytes: int = Field(
#17	default=_IMAGE_DEFAULTS.yaml_max_bytes,
#18	ge=1,
#19	le=_IMAGE_DEFAULTS.yaml_hard_limit_bytes,
#20	description="Maximum YAML bytes to read.",
#21	)
#22	replace: bool = Field(default=False, description="Whether an existing image with the same id may be replaced.")
#23
#24
#25	class LoadImageFromYamlOutput(BaseModel):
#26	image_id: str
#27	name: str
#28	version: str
#29	source_path: str
#30	replaced: bool
#31	default_tools: list[str]
#32	required_capabilities_count: int
#33
#34
#35	class LoadImageFromYamlTool(SyncAgentTool[LoadImageFromYamlArgs]):
#36	name = "load_image_from_yaml"
#37	description = (
#38	"Read an AgentImage registration manifest from a workspace YAML file and register it with the runtime. "
#39	"The filesystem primitive enforces file read authority; the image registry primitive enforces image write authority."
#40	)
#41	args_schema = LoadImageFromYamlArgs
#42	output_schema = LoadImageFromYamlOutput
#43	policy = ToolPolicy(
#44	side_effects=True,
#45	idempotent=False,
#46	permissions={"filesystem.read", "image.write"},
#47	timeout_s=_TOOL_DEFAULTS.standard_timeout_s,
#48	)
#49	tags = ["image", "registry", "yaml", "side_effect"]
#50
#51	def run(self, args: LoadImageFromYamlArgs, ctx: ToolContext) -> LoadImageFromYamlOutput:
#52	runtime = ctx.runtime
#53	if runtime is None:
#54	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#55	cwd = runtime.process.working_directory(ctx.pid)
#56	try:
#57	file_result = runtime.filesystem.read_text(
#58	pid=ctx.pid,
#59	path=args.path,
#60	encoding=args.encoding,
#61	max_bytes=args.max_bytes,
#62	cwd=cwd,
#63	)
#64	except UnicodeDecodeError as exc:
#65	raise ToolExecutionError(
#66	"Image YAML file could not be decoded with the requested encoding.",
#67	code=ToolErrorCode.EXECUTION_ERROR,
#68	details={"encoding": args.encoding, "path": args.path},
#69	) from exc
#70	if file_result.truncated:
#71	raise ToolExecutionError(
#72	"Image YAML exceeded max_bytes; no image was registered.",
#73	code=ToolErrorCode.VALIDATION_ERROR,
#74	details={"path": file_result.path, "bytes_read": file_result.bytes_read, "max_bytes": args.max_bytes},
#75	)
#76	try:
#77	result = runtime.image_registry.register_from_yaml_text(
#78	file_result.content,
#79	actor=ctx.pid,
#80	replace=args.replace,
#81	require_capability=True,
#82	source=file_result.path,
#83	)
#84	except LibOSValidationError as exc:
#85	raise ToolExecutionError(
#86	str(exc),
#87	code=ToolErrorCode.VALIDATION_ERROR,
#88	details={"path": file_result.path},
#89	) from exc
#90	image = result.image
#91	return LoadImageFromYamlOutput(
#92	image_id=image.image_id,
#93	name=image.name,
#94	version=image.version,
#95	source_path=file_result.path,
#96	replaced=result.replaced,
#97	default_tools=list(image.default_tools),
#98	required_capabilities_count=len(image.required_capabilities),
#99	)
#100

z6MkqRzA/yingqi-z20-Agent-libOS