Poe Api Universal Generator

__pycache__/
*.pyc
.venv/
outputs/
*.log

MIT License

Copyright (c) 2026 Samuel F.

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

[project]
name = "poe-api-universal-generator-skill"
version = "2.0.0"
description = "Universal Poe API skill scripts for llm, image, video, and audio request flows"
requires-python = ">=3.11"
dependencies = [
  "requests>=2.31.0",
  "openai>=1.30.0",
]

[tool.uv]
package = false

#!/usr/bin/env python3
"""Backward-compatible image wrapper for the universal Poe orchestrator."""

from __future__ import annotations

import argparse
import json

from poe_request_orchestrator import main as orchestrator_main


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate image via Poe nano-banana-2")
    parser.add_argument("--prompt", required=True, help="Image generation prompt")
    parser.add_argument("--out", default="nano_banana_response.json", help="Output JSON file path")
    parser.add_argument("--aspect-ratio", dest="aspect_ratio", default=None, help="Optional aspect ratio, e.g. 4:3")
    parser.add_argument("--seed", type=int, default=None, help="Optional seed")
    parser.add_argument("--retries", type=int, default=4, help="Retry attempts for retryable errors")
    parser.add_argument("--timeout", type=int, default=90, help="Request timeout seconds")
    parser.add_argument("--dry-run", action="store_true", help="Build payload without sending network request")
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    extra = {}
    if args.aspect_ratio:
        extra["aspect_ratio"] = args.aspect_ratio
    if args.seed is not None:
        extra["seed"] = args.seed

    argv = [
        "--mode",
        "image",
        "--model",
        "nano-banana-2",
        "--prompt",
        args.prompt,
        "--out",
        args.out,
        "--retries",
        str(args.retries),
        "--timeout",
        str(args.timeout),
    ]
    if extra:
        argv.extend(["--params", json.dumps(extra, ensure_ascii=False)])
    if args.dry_run:
        argv.append("--dry-run")

    return orchestrator_main(argv)


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Universal Poe API request helper for llm/image/video/audio modes.

This script builds and submits Poe API requests with shared retry behavior
and writes reproducible response records to disk.
"""

from __future__ import annotations

import argparse
import json
import os
import random
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import requests

BASE_URL = "https://api.poe.com"
RETRYABLE_STATUS = {429, 503, 529}
SUPPORTED_MODES = {"llm", "image", "video", "audio"}


@dataclass(frozen=True)
class Route:
    mode: str
    endpoint: str


def resolve_route(mode: str) -> Route:
    key = mode.strip().lower()
    if key == "llm":
        return Route(mode="llm", endpoint="/v1/chat/completions")
    if key == "image":
        return Route(mode="image", endpoint="/v1/images")
    if key == "video":
        return Route(mode="video", endpoint="/v1/videos")
    if key == "audio":
        return Route(mode="audio", endpoint="/v1/audio")
    raise ValueError(f"Unsupported mode: {mode}")


def normalize_stream(mode: str, requested: bool) -> bool:
    # Poe docs recommend stream=False for media bots.
    if mode.lower() in {"image", "video", "audio"}:
        return False
    return requested


def build_payload(
    *,
    mode: str,
    model: str,
    prompt: str,
    params: dict[str, Any] | None = None,
    stream: bool = False,
) -> dict[str, Any]:
    route = resolve_route(mode)
    payload: dict[str, Any]

    if route.mode == "llm":
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "stream": normalize_stream(route.mode, stream),
        }
    else:
        payload = {
            "model": model,
            "prompt": prompt,
        }

    if params:
        payload["extra_body"] = params

    return payload


def post_with_retries(
    *,
    url: str,
    headers: dict[str, str],
    payload: dict[str, Any],
    timeout: int,
    retries: int,
) -> requests.Response:
    attempt = 0
    while True:
        attempt += 1
        response = requests.post(url, headers=headers, json=payload, timeout=timeout)
        if response.status_code not in RETRYABLE_STATUS:
            return response
        if attempt > retries:
            return response

        retry_after = response.headers.get("Retry-After")
        if retry_after and retry_after.isdigit():
            sleep_s = max(1.0, float(retry_after))
        else:
            base = min(8.0, 0.25 * (2 ** (attempt - 1)))
            sleep_s = base + random.uniform(0.0, 0.25)
        time.sleep(sleep_s)


def parse_params(raw: str | None) -> dict[str, Any] | None:
    if not raw:
        return None
    parsed = json.loads(raw)
    if not isinstance(parsed, dict):
        raise ValueError("--params must be a JSON object")
    return parsed


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Universal Poe API request orchestrator")
    parser.add_argument("--mode", required=True, choices=sorted(SUPPORTED_MODES), help="Request mode")
    parser.add_argument("--model", required=True, help="Poe model or bot id")
    parser.add_argument("--prompt", required=True, help="User prompt")
    parser.add_argument("--params", default=None, help="Optional JSON object for extra_body")
    parser.add_argument("--stream", action="store_true", help="Enable streaming for llm mode")
    parser.add_argument("--timeout", type=int, default=90, help="Request timeout seconds")
    parser.add_argument("--retries", type=int, default=4, help="Retry attempts for retryable statuses")
    parser.add_argument("--out", default="poe_response.json", help="Output JSON file path")
    parser.add_argument("--dry-run", action="store_true", help="Build and save payload without sending network request")
    return parser.parse_args(argv)


def save_record(path: Path, record: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(record, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    route = resolve_route(args.mode)
    params = parse_params(args.params)
    payload = build_payload(mode=args.mode, model=args.model, prompt=args.prompt, params=params, stream=args.stream)

    output_path = Path(args.out)

    if args.dry_run:
        save_record(
            output_path,
            {
                "mode": args.mode,
                "endpoint": route.endpoint,
                "payload": payload,
                "dry_run": True,
            },
        )
        print(f"Dry-run payload saved to {output_path}")
        return 0

    api_key = os.getenv("POE_API_KEY")
    if not api_key:
        raise SystemExit("Missing POE_API_KEY environment variable")

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    url = f"{BASE_URL}{route.endpoint}"
    response = post_with_retries(url=url, headers=headers, payload=payload, timeout=args.timeout, retries=args.retries)

    try:
        body: Any = response.json()
    except Exception:
        body = {"raw": response.text}

    record = {
        "mode": args.mode,
        "endpoint": route.endpoint,
        "status_code": response.status_code,
        "model": args.model,
        "payload": payload,
        "response": body,
        "request_id": response.headers.get("x-request-id"),
    }
    save_record(output_path, record)

    if response.ok:
        print(f"Saved response to {output_path}")
        return 0

    print(f"Request failed: status={response.status_code}, saved response to {output_path}")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())

# Managed primarily via uv/pyproject.toml; kept for compatibility export.
requests>=2.31.0
openai>=1.30.0