repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #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 | _RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime |
| #9 | _TOOL_DEFAULTS = DEFAULT_CONFIG.tools |
| #10 | |
| #11 | |
| #12 | class RequestPermissionArgs(BaseModel): |
| #13 | resource: str = Field(description="Capability resource to request, such as filesystem:workspace:path.txt.") |
| #14 | rights: list[str] = Field(description="Capability rights to request, such as ['write'].") |
| #15 | reason: str = Field(description="Brief reason shown to the human.") |
| #16 | human: str = Field(default=_RUNTIME_DEFAULTS.default_human, description="Human recipient name.") |
| #17 | |
| #18 | |
| #19 | class RequestPermissionOutput(BaseModel): |
| #20 | request_id: str |
| #21 | resource: str |
| #22 | rights: list[str] |
| #23 | status: str |
| #24 | |
| #25 | |
| #26 | class RequestPermissionTool(SyncAgentTool[RequestPermissionArgs]): |
| #27 | name = "request_permission" |
| #28 | description = ( |
| #29 | "Ask the human to set a permission policy for a libOS capability resource. " |
| #30 | "The human can always allow, always deny, or require per-use approval." |
| #31 | ) |
| #32 | args_schema = RequestPermissionArgs |
| #33 | output_schema = RequestPermissionOutput |
| #34 | policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s) |
| #35 | tags = ["permission", "human", "capability"] |
| #36 | |
| #37 | def run(self, args: RequestPermissionArgs, ctx: ToolContext) -> RequestPermissionOutput: |
| #38 | runtime = ctx.runtime |
| #39 | if runtime is None: |
| #40 | raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR) |
| #41 | if not args.rights: |
| #42 | raise ToolExecutionError( |
| #43 | "At least one capability right is required.", |
| #44 | code=ToolErrorCode.VALIDATION_ERROR, |
| #45 | ) |
| #46 | request_id = runtime.human.request_permission( |
| #47 | pid=ctx.pid, |
| #48 | human=args.human, |
| #49 | resource=args.resource, |
| #50 | rights=args.rights, |
| #51 | reason=args.reason, |
| #52 | ) |
| #53 | return RequestPermissionOutput( |
| #54 | request_id=request_id, |
| #55 | resource=args.resource, |
| #56 | rights=args.rights, |
| #57 | status="pending", |
| #58 | ) |
| #59 |