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	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#9
#10
#11	class GetCurrentTimeArgs(BaseModel):
#12	timezone: str = Field(
#13	default=_TOOL_DEFAULTS.clock_timezone,
#14	description="IANA timezone name, for example UTC or Asia/Shanghai.",
#15	)
#16
#17
#18	class GetCurrentTimeOutput(BaseModel):
#19	iso8601: str
#20	unix_seconds: float
#21	timezone: str
#22
#23
#24	class SleepArgs(BaseModel):
#25	seconds: float = Field(
#26	ge=0,
#27	le=_TOOL_DEFAULTS.max_sleep_seconds,
#28	description=f"Seconds to sleep. Maximum is {_TOOL_DEFAULTS.max_sleep_seconds:g} seconds.",
#29	)
#30
#31
#32	class SleepOutput(BaseModel):
#33	requested_seconds: float
#34	elapsed_seconds: float
#35
#36
#37	class GetCurrentTimeTool(BaseAgentTool[GetCurrentTimeArgs]):
#38	name = "get_current_time"
#39	description = "Return the current wall-clock time from the libOS clock primitive."
#40	args_schema = GetCurrentTimeArgs
#41	output_schema = GetCurrentTimeOutput
#42	policy = ToolPolicy(side_effects=False, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#43	tags = ["clock", "time"]
#44
#45	async def execute(self, args: GetCurrentTimeArgs, ctx: ToolContext) -> GetCurrentTimeOutput:
#46	runtime = ctx.runtime
#47	if runtime is None:
#48	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#49	result = runtime.clock.now(ctx.pid, tz=args.timezone)
#50	return GetCurrentTimeOutput(
#51	iso8601=result.iso8601,
#52	unix_seconds=result.unix_seconds,
#53	timezone=result.timezone,
#54	)
#55
#56
#57	class SleepTool(BaseAgentTool[SleepArgs]):
#58	name = "sleep"
#59	description = "Sleep for a bounded duration using the libOS clock primitive."
#60	args_schema = SleepArgs
#61	output_schema = SleepOutput
#62	policy = ToolPolicy(side_effects=False, idempotent=False, timeout_s=_TOOL_DEFAULTS.sleep_tool_timeout_s)
#63	tags = ["clock", "time", "scheduler"]
#64
#65	async def execute(self, args: SleepArgs, ctx: ToolContext) -> SleepOutput:
#66	runtime = ctx.runtime
#67	if runtime is None:
#68	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#69	result = await runtime.clock.asleep(ctx.pid, seconds=args.seconds)
#70	return SleepOutput(
#71	requested_seconds=result.requested_seconds,
#72	elapsed_seconds=result.elapsed_seconds,
#73	)
#74

z6MkqRzA/yingqi-z20-Agent-libOS