"""Python 3.10+ Intelligence example; standard library, preview by default. client.py reads TASKVEGAS_CLIENT_ID and TASKVEGAS_CLIENT_SECRET from securely supplied environment variables. Actual values must never be committed. The approved brief can reach configured inference processors and use free credits. Explicit ``--run`` executes only ready, free, read-only child tools. Secrets, tokens, briefs, envelopes, and results are never printed or persisted. Socket timeouts and elapsed checks are bounded, but urllib cannot guarantee hard wall-clock cancellation. An uncertain execution is never retried; use the plan-status tool to investigate. """ from __future__ import annotations import json import re import sys import time from collections.abc import Mapping, Sequence from datetime import datetime, timezone from urllib.error import HTTPError, URLError from urllib.request import Request from uuid import UUID, uuid4 from client import ClientError, TaskVegasClient, status_error MAX_RESPONSE_BYTES = 256 * 1024 MAX_REQUEST_BYTES = 32 * 1024 PLANNING_TIMEOUT_SECONDS = 20 RUN_TIMEOUT_SECONDS = 50 TOOL_IDS = frozenset({ "taskvegas_plan_task", "taskvegas_run_plan", "taskvegas_get_plan_status", "taskvegas_get_intelligence_usage", }) PLAN_STATUSES = frozenset({ "ready", "needs-setup", "clarification", "recommendation", "host-action", }) RUN_STATUSES = frozenset({ "completed", "partial", "stopped", "failed", "running", "unknown", }) DEFAULT_TASK = ( "Find reviewed design guidance for an issue-triage dashboard with keyboard " "access and readable status labels." ) JsonObject = dict[str, object] def canonical_uuid(value: str) -> str: """Reject invalid identifiers before constructing any network request.""" try: if not isinstance(value, str) or str(UUID(value)) != value: raise ValueError except (ValueError, AttributeError) as error: raise ClientError("Use a canonical UUID identifier.") from error return value def encode_input(tool_input: JsonObject, key: str) -> bytes: """Bound outbound JSON before authentication or network activity.""" try: payload = json.dumps( {"input": tool_input, "idempotencyKey": key}, allow_nan=False ).encode("utf-8") except (ValueError, TypeError): raise ClientError("Use valid bounded JSON input.") from None if len(payload) > MAX_REQUEST_BYTES: raise ClientError("Intelligence input exceeded the client size limit.") return payload class TaskVegasIntelligenceClient(TaskVegasClient): """Use exact Intelligence grants, existing child permissions, and consent.""" def _json( self, request: Request, maximum: int = MAX_RESPONSE_BYTES, timeout_seconds: int = PLANNING_TIMEOUT_SECONDS, ) -> JsonObject: deadline = time.monotonic() + timeout_seconds try: with self.opener.open(request, timeout=timeout_seconds) as response: if response.headers.get_content_type() != "application/json": raise ClientError("TaskVegas returned an unexpected response type.") parts: list[bytes] = [] size = 0 while True: if time.monotonic() > deadline: raise ClientError( "TaskVegas exceeded the client elapsed-time limit." ) chunk = response.read1(min(65536, maximum + 1 - size)) if time.monotonic() > deadline: raise ClientError( "TaskVegas exceeded the client elapsed-time limit." ) if not chunk: break size += len(chunk) if size > maximum: raise ClientError( "TaskVegas exceeded the client response-size limit." ) parts.append(chunk) value = json.loads(b"".join(parts).decode("utf-8")) if not isinstance(value, dict): raise ClientError("TaskVegas returned an unexpected JSON shape.") return value except HTTPError as error: error.close() raise status_error(error.code) from None except (URLError, TimeoutError, OSError): raise ClientError( "TaskVegas could not be reached within its timeout. An uncertain " "execution was not retried; check its plan status." ) from None except (ValueError, UnicodeError): raise ClientError("TaskVegas returned invalid JSON.") from None def execute( self, tool_id: str, tool_input: JsonObject, idempotency_key: str | None = None, ) -> JsonObject: """Submit a fixed tool once; never repeat an uncertain inference or run.""" if ( not isinstance(tool_id, str) or tool_id not in TOOL_IDS or not isinstance(tool_input, dict) ): raise ClientError( "This example supports only its fixed Intelligence tools." ) key = canonical_uuid( str(uuid4()) if idempotency_key is None else idempotency_key ) payload = encode_input(tool_input, key) request = Request( f"{self.origin}/api/v1/tools/{tool_id}/execute", data=payload, headers={ "Authorization": "Bearer " + self.token(), "Content-Type": "application/json", "Accept": "application/json", }, method="POST", ) try: value = self._json( request, timeout_seconds=( RUN_TIMEOUT_SECONDS if tool_id == "taskvegas_run_plan" else PLANNING_TIMEOUT_SECONDS ), ) except ClientError as error: if error.status == 401: self.access = None raise result = value.get("result") if not isinstance(result, dict): raise ClientError("TaskVegas returned an incomplete execution response.") return result def _discovery(self) -> dict[str, JsonObject]: value = self.request("/api/v1/tools") tools = value.get("tools") if value.get("resource") != self.resource or not isinstance(tools, list): raise ClientError("Tool discovery returned the wrong resource or shape.") return { item["name"]: item for item in tools if isinstance(item, dict) and isinstance(item.get("name"), str) } def preview(self, task: str = DEFAULT_TASK) -> JsonObject: """Approve processing this short summary; execute no child tools.""" if not isinstance(task, str) or not task.strip() or len(task) > 500: raise ClientError( "Use an approved summary of 1–500 characters without secrets." ) if "taskvegas_plan_task" not in self._discovery(): raise ClientError( "Grant taskvegas_plan_task and approve Intelligence " "brief processing first." ) return self.execute("taskvegas_plan_task", { "task": task.strip(), "context": { "source": "user-approved-summary", "constraints": {"budget": "free", "readOnly": True}, }, "approvedBrief": True, }) def run_free_plan(self, plan: Mapping[str, object]) -> JsonObject: """Run only an explicit, unexpired free plan; server revalidates authority.""" if not isinstance(plan, Mapping): raise ClientError("Use a ready, unexpired plan returned by TaskVegas.") steps, envelope = plan.get("steps"), plan.get("envelope") try: expiry = datetime.fromisoformat( str(plan.get("expiresAt", "")).replace("Z", "+00:00") ) except ValueError: raise ClientError( "Use a ready, unexpired plan returned by TaskVegas." ) from None if ( plan.get("status") != "ready" or not isinstance(envelope, str) or not 20 <= len(envelope) <= 24000 or re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_-]{43}", envelope) is None or not isinstance(steps, list) or not 1 <= len(steps) <= 5 or expiry.tzinfo is None or expiry <= datetime.now(timezone.utc) ): raise ClientError("Use a ready, unexpired plan returned by TaskVegas.") tools = self._discovery() if "taskvegas_run_plan" not in tools: raise ClientError( "Grant taskvegas_run_plan before running an approved plan." ) for step in steps: tool_id = step.get("toolId") if isinstance(step, dict) else None tool = tools.get(tool_id) if isinstance(tool_id, str) else None if ( not isinstance(step, dict) or step.get("paid") is not False or not tool or tool.get("readOnly") is not True or tool.get("paid") is not False or tool.get("agentAllowed") is not True or tool_id in TOOL_IDS ): raise ClientError( "This example runs only exactly granted, free, " "read-only child tools." ) return self.execute("taskvegas_run_plan", {"envelope": envelope}) def plan_status(self, plan_id: str) -> JsonObject: """Retrieve sanitized status without replaying an execution.""" return self.execute( "taskvegas_get_plan_status", {"id": canonical_uuid(plan_id)} ) def usage(self) -> JsonObject: """Read the shared free service allowance; no payment is taken.""" return self.execute("taskvegas_get_intelligence_usage", {}) def main(arguments: Sequence[str] | None = None) -> int: """Print summaries only; never print provider text, plans, or credentials.""" args = list(sys.argv[1:] if arguments is None else arguments) try: if args not in ([], ["--run"], ["--usage"]) and not ( len(args) == 2 and args[0] == "--status" ): raise ClientError("Use no argument, --run, --usage, or --status PLAN_UUID.") client = TaskVegasIntelligenceClient() if args == ["--usage"]: client.usage() print("Free intelligence-credit usage retrieved. No payment taken.") return 0 if args and args[0] == "--status": result = client.plan_status(args[1]) else: value = client.preview() plan = value.get("plan") if not isinstance(plan, dict) or not isinstance(plan.get("steps"), list): raise ClientError( "No complete plan was returned. No child tools were run." ) status = plan.get("status") if not isinstance(status, str) or status not in PLAN_STATUSES: status = "review required" print(f"Plan preview: {status}; {len(plan['steps'])} step(s).") if not args: return 0 result = client.run_free_plan(plan) status = result.get("status") if not isinstance(status, str) or status not in RUN_STATUSES: status = "check plan status" print(f"Execution response: {status}.") return 0 except ClientError as error: print(str(error), file=sys.stderr) except Exception: print( "Intelligence client failed. Check configuration and service status. " "No execution was retried.", file=sys.stderr, ) return 1 if __name__ == "__main__": sys.exit(main())