后台响应与任务查询

export API_BASE_URL="https://gateway.example.com/v1"
export API_KEY="YOUR_API_KEY"
export API_RESPONSE_ID="resp_YOUR_EXISTING_RESPONSE_ID"
.venv/bin/python examples/response-status.py --max-attempts 10 --interval 2

import argparse
import math
import os
import time
from urllib.parse import quote
from openai import OpenAI, APIConnectionError, APIStatusError

parser = argparse.ArgumentParser(description="Poll an existing task-plugin response; never submit a new task.")
parser.add_argument("--max-attempts", type=int, default=10)
parser.add_argument("--interval", type=float, default=2.0)
args = parser.parse_args()
if not 1 <= args.max_attempts <= 60 or not math.isfinite(args.interval) or not 0 <= args.interval <= 60:
    parser.error("Use 1..60 attempts and a finite interval of 0..60 seconds.")
response_id = os.environ["API_RESPONSE_ID"]
if not response_id.startswith("resp_") or len(response_id) <= 5:
    raise SystemExit("API_RESPONSE_ID must be an existing task-plugin response ID.")

with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=30.0, max_retries=0) as client:
    try:
        for attempt in range(args.max_attempts):
            result = client.get(f"/responses/{quote(response_id, safe='')}", cast_to=dict)
            if (not isinstance(result, dict) or result.get("id") != response_id
                    or result.get("object") != "response"):
                raise ValueError("Invalid response identity.")
            status = result.get("status")
            if result.get("error") or status in ("failed", "incomplete", "cancelled"):
                raise ValueError("Response did not complete.")
            if status == "completed":
                output = result.get("output")
                if not isinstance(output, list) or not output or any(not isinstance(item, dict) for item in output):
                    raise ValueError("Missing output.")
                # Lifecycle completion is not a text, image, or safety quality check.
                print(f"completed: {len(output)} output item(s); validate content for your application.")
                break
            if status not in ("queued", "in_progress"):
                raise ValueError("Unknown response status.")
            if attempt + 1 == args.max_attempts:
                raise SystemExit("Polling limit reached. Keep the response ID; the task may still be running.")
            time.sleep(args.interval)
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). Keep the response ID; do not resubmit.")
    except APIConnectionError:
        raise SystemExit("Connection failed. Keep the response ID; do not resubmit.")
    except (ValueError, TypeError, AttributeError):
        raise SystemExit("Response is invalid or not complete. Keep the response ID; do not resubmit.")