"""UpServe Agent Runtime API 최소 Python client (표준 라이브러리 + httpx). 계약 원천은 https://docs.upserve.app/runtime-api/openapi.json 이다. 같은 기능의 TypeScript 판은 runtime_client.ts 이고 함수 이름은 snake_case <-> camelCase 로만 다르다. - API key 는 환경변수 UPSERVE_API_KEY 에서만 읽는다. 코드·로그에 키를 쓰지 않는다. - 생성/입력/취소/resolve/close/delete/업로드 요청에는 Idempotency-Key 를 자동으로 만들고 (uuid4), 호출자가 넘긴 키는 그대로 쓴다. 재시도에는 항상 같은 키를 다시 보낸다. - 네트워크 오류, 429, 503, retryable=true 인 5xx 만 재시도한다(Retry-After 우선, 없으면 지수 backoff + jitter). 그 밖의 4xx 는 RuntimeApiError 로 올린다(code, request_id 보존). - webhook 서명 검증은 이 파일에 다시 만들지 않는다. runtime_webhook_receiver.py 를 쓴다. 사용 예(환경변수 UPSERVE_API_KEY 필요): python runtime_client.py session "이번 달 매출을 요약해주세요." python runtime_client.py run '{"count": 3}' """ from __future__ import annotations import json import os import random import sys import time import uuid from typing import Any, Callable, Iterator from urllib.parse import urlparse import httpx DEFAULT_BASE_URL = "https://api.upserve.app" TURN_DONE = {"SUCCEEDED", "FAILED", "CANCELED", "EXPIRED"} RUN_DONE = {"SUCCEEDED", "PARTIALLY_SUCCEEDED", "FAILED", "CANCELED", "EXPIRED"} TERMINAL_EVENTS = {"turn.completed", "turn.failed", "turn.canceled", "turn.expired", "run.completed", "run.failed", "run.canceled", "run.expired"} # 종결은 아니지만 호출자 응답 없이는 더 오지 않는 이벤트. 스트림을 닫고 상태를 조회한다. WAIT_EVENTS = {"turn.action_required", "run.action_required"} MAX_RETRY_AFTER = 300.0 # 서버 Retry-After(최대 60초)를 자르지 않되 비정상 값은 막는다 LOOPBACK = {"localhost", "127.0.0.1", "::1"} class RuntimeApiError(Exception): """{error:{code,message,retryable,request_id,details}} 오류 본문을 그대로 보존한다.""" def __init__(self, status: int, code: str, message: str, retryable: bool = False, request_id: str | None = None, details: dict | None = None, retry_after: float | None = None): super().__init__(f"{status} {code}: {message} (request_id={request_id})") self.status, self.code, self.message, self.retryable = status, code, message, retryable self.request_id, self.details, self.retry_after = request_id, details or {}, retry_after class CursorExpired(RuntimeApiError): """410 EVENT_CURSOR_EXPIRED. 복구: state_url 로 현재 상태, items_url 로 보존된 결과를 읽고 필요하면 cursor 없이 events() 를 다시 연다.""" @property def state_url(self) -> str | None: return self.details.get("state_url") @property def items_url(self) -> str | None: return self.details.get("items_url") def _error_from(status: int, body: Any, headers: Any = None) -> RuntimeApiError: retry_after = None if headers is not None and headers.get("Retry-After", "").strip().isdigit(): retry_after = float(headers["Retry-After"]) detail = body.get("error") if isinstance(body, dict) and isinstance(body.get("error"), dict) else None if detail is None: # 게이트웨이/프록시가 만든 비정형 오류. 일시적 상태 코드만 재시도 대상으로 본다. return RuntimeApiError(status, "HTTP_" + str(status), "Non-API error response", retryable=status in (429, 502, 503, 504), retry_after=retry_after) cls = CursorExpired if detail.get("code") == "EVENT_CURSOR_EXPIRED" else RuntimeApiError return cls(status, detail.get("code", "UNKNOWN"), detail.get("message", ""), bool(detail.get("retryable")), detail.get("request_id"), detail.get("details"), retry_after) def parse_sse(lines: Iterator[str]) -> Iterator[dict]: """SSE 줄 -> {"id","event","data"}. 주석(": keep-alive")은 버리고 빈 줄에서 한 건을 낸다.""" frame: dict = {} for line in lines: if line == "": if frame: yield frame frame = {} elif not line.startswith(":"): name, _, value = line.partition(":") frame[name] = (frame[name] + "\n" if name in frame and name == "data" else "") + value.removeprefix(" ") class RuntimeClient: def __init__(self, base_url: str | None = None, api_key: str | None = None, *, max_retries: int = 4, base_delay: float = 0.5, max_delay: float = 30.0, timeout: float = 30.0, sleep: Callable[[float], None] = time.sleep, http: httpx.Client | None = None): self.base_url = (base_url or os.environ.get("UPSERVE_API_BASE_URL") or DEFAULT_BASE_URL).rstrip("/") key = api_key or os.environ.get("UPSERVE_API_KEY") if not key: raise RuntimeError("Set UPSERVE_API_KEY in the environment") host = urlparse(self.base_url) if host.scheme != "https" and host.hostname not in LOOPBACK: raise RuntimeError("The API key is only sent over https (or to a loopback address)") self._key = key self.max_retries, self.base_delay, self.max_delay, self._sleep = max_retries, base_delay, max_delay, sleep self._http = http or httpx.Client(base_url=self.base_url, timeout=timeout) def close(self) -> None: self._http.close() def __enter__(self) -> "RuntimeClient": return self def __exit__(self, *exc: Any) -> None: self.close() # ---- 전송 계층: 재시도·멱등 키·오류 파싱 ---- def _delay(self, attempt: int, error: RuntimeApiError | None) -> float: if error is not None and error.retry_after is not None: # 서버가 알려 준 대기는 max_delay로 자르지 않는다. 더 일찍 보내면 같은 한도에 다시 걸린다. return min(error.retry_after, MAX_RETRY_AFTER) return min(self.max_delay, self.base_delay * 2 ** attempt) * random.uniform(0.5, 1.0) def _request(self, method: str, path: str, *, params: dict | None = None, json_body: Any = None, files: dict | None = None, idempotent: bool = False, idempotency_key: str | None = None) -> httpx.Response: headers = {"Authorization": "Bearer " + self._key, "Accept": "application/json"} if idempotent: # 키는 재시도 루프 밖에서 한 번만 정한다. 재시도마다 같은 값을 보낸다. headers["Idempotency-Key"] = idempotency_key or str(uuid.uuid4()) for attempt in range(self.max_retries + 1): try: response = self._http.request(method, path, params=params, json=json_body, files=files, headers=headers) except httpx.TransportError: if attempt >= self.max_retries: raise self._sleep(self._delay(attempt, None)) continue if response.status_code < 400: return response try: body = response.json() except ValueError: body = None error = _error_from(response.status_code, body, response.headers) if error.retryable and attempt < self.max_retries: self._sleep(self._delay(attempt, error)) continue raise error raise AssertionError("unreachable") def _json(self, *args: Any, **kwargs: Any) -> Any: return self._request(*args, **kwargs).json() # ---- Deployment / Session / Turn ---- def list_deployments(self) -> list[dict]: return self._json("GET", "/v1/deployments")["items"] def get_deployment(self, deployment_id: str) -> dict: return self._json("GET", f"/v1/deployments/{deployment_id}") @staticmethod def _turn_body(text: str, metadata: dict | None, limits: dict | None, file_ids: list[str] | None) -> dict: body: dict = {"input": [{"type": "TEXT", "text": text}] + [{"type": "FILE", "file_id": f} for f in file_ids or []]} if metadata: body["metadata"] = metadata if limits: body["limits"] = limits return body def create_session(self, deployment_id: str, text: str, *, metadata: dict | None = None, limits: dict | None = None, file_ids: list[str] | None = None, idempotency_key: str | None = None) -> dict: """202 receipt: session_id, turn_id, status, status_url, events_url, idempotency_expires_at.""" return self._json("POST", f"/v1/deployments/{deployment_id}/sessions", idempotent=True, idempotency_key=idempotency_key, json_body=self._turn_body(text, metadata, limits, file_ids)) def send_turn(self, session_id: str, text: str, *, metadata: dict | None = None, limits: dict | None = None, file_ids: list[str] | None = None, idempotency_key: str | None = None) -> dict: return self._json("POST", f"/v1/sessions/{session_id}/turns", idempotent=True, idempotency_key=idempotency_key, json_body=self._turn_body(text, metadata, limits, file_ids)) def get_session(self, session_id: str) -> dict: return self._json("GET", f"/v1/sessions/{session_id}") def get_turn(self, session_id: str, turn_id: str, *, after: str | None = None, limit: int | None = None) -> dict: params = {k: v for k, v in {"after": after, "limit": limit}.items() if v is not None} return self._json("GET", f"/v1/sessions/{session_id}/turns/{turn_id}", params=params) def wait_turn(self, session_id: str, turn_id: str, *, timeout: float = 600.0, interval: float = 1.0) -> dict: """종결 상태(SUCCEEDED/FAILED/CANCELED/EXPIRED) 또는 REQUIRES_ACTION 까지 폴링한다.""" deadline = time.monotonic() + timeout while True: turn = self.get_turn(session_id, turn_id) if turn["status"] in TURN_DONE or turn["status"] == "REQUIRES_ACTION": return turn if time.monotonic() >= deadline: raise TimeoutError(f"Turn {turn_id} is still {turn['status']}") self._sleep(interval) interval = min(interval * 1.5, 5.0) def cancel_turn(self, session_id: str, turn_id: str, *, idempotency_key: str | None = None) -> dict: return self._json("POST", f"/v1/sessions/{session_id}/turns/{turn_id}/cancel", idempotent=True, idempotency_key=idempotency_key) def close_session(self, session_id: str, *, idempotency_key: str | None = None) -> dict: return self._json("POST", f"/v1/sessions/{session_id}/close", idempotent=True, idempotency_key=idempotency_key) def delete_session(self, session_id: str, *, idempotency_key: str | None = None) -> dict: return self._json("DELETE", f"/v1/sessions/{session_id}", idempotent=True, idempotency_key=idempotency_key) def resolve_action(self, session_id: str, action_id: str, answers: list[dict], *, idempotency_key: str | None = None) -> dict: """caller 가 풀 수 있는 QUESTION 만 해소한다. answers: [{"question_id","text"|"option_ids"}].""" return self._json("POST", f"/v1/sessions/{session_id}/actions/{action_id}/resolve", idempotent=True, idempotency_key=idempotency_key, json_body={"answers": answers}) # ---- SSE ---- def events(self, session_id: str, *, after: str | None = None, until_terminal: bool = True, max_reconnects: int = 5) -> Iterator[dict]: """Session 이벤트를 {"id","event","data"} 로 낸다. 끊기면 마지막 id 를 Last-Event-ID 로 재연결한다. until_terminal=True 면 첫 turn.* 종결 이벤트에서 끝낸다(이후 Turn 은 새로 호출).""" return self._events(f"/v1/sessions/{session_id}/events", after, until_terminal, max_reconnects) def run_events(self, run_id: str, *, after: str | None = None, until_terminal: bool = True, max_reconnects: int = 5) -> Iterator[dict]: return self._events(f"/v1/runs/{run_id}/events", after, until_terminal, max_reconnects) def _events(self, path: str, after: str | None, until_terminal: bool, max_reconnects: int) -> Iterator[dict]: last, failures = after, 0 while True: headers = {"Authorization": "Bearer " + self._key, "Accept": "text/event-stream"} if last: headers["Last-Event-ID"] = last got_event, retry_error = False, None try: with self._http.stream("GET", path, headers=headers, timeout=httpx.Timeout(30.0, read=60.0)) as response: if response.status_code >= 400: response.read() try: body = response.json() except ValueError: body = None raise _error_from(response.status_code, body, response.headers) for frame in parse_sse(response.iter_lines()): if frame.get("event") == "error": # 헤더를 보낸 뒤의 오류. 만료면 CursorExpired 로 복구 경로를 알려 준다. raise _error_from(410 if "EVENT_CURSOR_EXPIRED" in frame.get("data", "") else 500, json.loads(frame["data"])) got_event = True last = frame["id"] yield {"id": frame["id"], "event": frame["event"], "data": json.loads(frame["data"])} if until_terminal and frame["event"] in TERMINAL_EVENTS | WAIT_EVENTS: return except httpx.TransportError: pass # 연결 끊김: 아래에서 마지막 id 로 재연결 except RuntimeApiError as error: if not error.retryable: raise retry_error = error failures = 0 if got_event else failures + 1 if failures > max_reconnects: raise RuntimeError("Event stream kept closing without new events") self._sleep(self._delay(failures, retry_error) if not got_event else 0) # ---- 파일과 Artifact ---- def upload_file(self, content: bytes | str | os.PathLike, *, filename: str | None = None, mime_type: str = "application/octet-stream", idempotency_key: str | None = None) -> dict: """POST /v1/files (multipart, 필드 하나: file). 201 FileReceipt 의 file_id 를 입력 FILE 에 쓴다.""" if isinstance(content, (bytes, bytearray)): data, name = bytes(content), filename or "upload.bin" else: with open(content, "rb") as handle: data = handle.read() name = filename or os.path.basename(os.fspath(content)) return self._json("POST", "/v1/files", idempotent=True, idempotency_key=idempotency_key, files={"file": (name, data, mime_type)}) def get_artifact(self, artifact_id: str) -> dict: return self._json("GET", f"/v1/artifacts/{artifact_id}") def download_artifact(self, artifact_id: str, *, max_reissues: int = 2) -> bytes: """download 주소를 발급(60초 만료·일회용)받아 content 를 받는다. 만료/소진(404)이면 다시 발급한다.""" for attempt in range(max_reissues + 1): receipt = self._json("POST", f"/v1/artifacts/{artifact_id}/download") if not receipt["url"].startswith("/v1/artifacts/"): raise RuntimeError("Unexpected download URL") # 서버가 준 상대 경로만 따른다 try: return self._request("GET", receipt["url"]).content except RuntimeApiError as error: if error.status != 404 or attempt >= max_reissues: raise raise AssertionError("unreachable") # ---- Workflow Run ---- def get_workflow(self, workflow_id: str) -> dict: return self._json("GET", f"/v1/workflows/{workflow_id}") def start_run(self, workflow_id: str, published_version: int, input: dict, *, file_ids: list[str] | None = None, limits: dict | None = None, metadata: dict | None = None, idempotency_key: str | None = None) -> dict: """202 receipt: run_id, status_url, events_url, idempotency_expires_at.""" body: dict = {"published_version": published_version, "input": input} for name, value in (("file_ids", file_ids), ("limits", limits), ("metadata", metadata)): if value: body[name] = value return self._json("POST", f"/v1/workflows/{workflow_id}/runs", idempotent=True, idempotency_key=idempotency_key, json_body=body) def get_run(self, run_id: str) -> dict: return self._json("GET", f"/v1/runs/{run_id}") def get_run_action(self, run_id: str, action_id: str) -> dict: """Read the current question or approval content to display in your application's UI.""" return self._json("GET", f"/v1/runs/{run_id}/actions/{action_id}") def resolve_run_action(self, run_id: str, action_id: str, decision: dict, *, idempotency_key: str | None = None) -> dict: """Submit answers, or a human decision with content_hash and actor.subject. Never auto-approve.""" return self._json("POST", f"/v1/runs/{run_id}/actions/{action_id}/resolve", json_body=decision, idempotent=True, idempotency_key=idempotency_key) def wait_run(self, run_id: str, *, timeout: float = 3600.0, interval: float = 2.0) -> dict: deadline = time.monotonic() + timeout while True: detail = self.get_run(run_id) if detail["run"]["status"] in RUN_DONE or detail["run"]["status"] == "REQUIRES_ACTION": return detail if time.monotonic() >= deadline: raise TimeoutError(f"Run {run_id} is still {detail['run']['status']}") self._sleep(interval) interval = min(interval * 1.5, 10.0) def list_outputs(self, run_id: str, *, limit: int = 50) -> Iterator[dict]: """결과를 cursor 로 이어 읽는다(next_cursor 가 null 이 될 때까지).""" cursor = None while True: params = {"limit": limit, **({"after": cursor} if cursor else {})} page = self._json("GET", f"/v1/runs/{run_id}/outputs", params=params) yield from page["outputs"] cursor = page["next_cursor"] if not cursor: return def cancel_run(self, run_id: str, *, idempotency_key: str | None = None) -> dict: return self._json("POST", f"/v1/runs/{run_id}/cancel", idempotent=True, idempotency_key=idempotency_key) def delete_run(self, run_id: str, *, idempotency_key: str | None = None) -> dict: return self._json("DELETE", f"/v1/runs/{run_id}", idempotent=True, idempotency_key=idempotency_key) def main(argv: list[str]) -> int: with RuntimeClient() as client: if len(argv) == 3 and argv[0] == "session": receipt = client.create_session(argv[1], argv[2]) print("accepted", receipt["session_id"], receipt["turn_id"], flush=True) for event in client.events(receipt["session_id"], after=None): print(event["id"], event["event"], flush=True) turn = client.wait_turn(receipt["session_id"], receipt["turn_id"]) print(json.dumps({"status": turn["status"], "outputs": turn["outputs"], "usage": turn["usage"]})) return 0 if turn["status"] == "SUCCEEDED" else 1 if len(argv) == 4 and argv[0] == "run": receipt = client.start_run(argv[1], int(argv[2]), json.loads(argv[3])) detail = client.wait_run(receipt["run_id"]) print(json.dumps({"status": detail["run"]["status"], "outputs": list(client.list_outputs(receipt["run_id"]))})) return 0 if detail["run"]["status"] in {"SUCCEEDED", "PARTIALLY_SUCCEEDED"} else 1 print(__doc__, file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))