# Python integration

Run these examples on your server. They use the language runtime HTTP client, not a platform SDK. No request is sent until you run a command. Models must be configured and enabled before use.

## Setup

Runtime: **Python 3.11+**. [Download complete source](/examples/python.py).

Set your deployment base URL, including `/v1` exactly once, and a server-side API key. `https://xxxrouter.com` is the official service domain; for local testing, replace it with your deployment origin. The service is not yet deployed and integration certification is not complete.

```sh
export XXXROUTER_BASE_URL='https://xxxrouter.com/v1'
export XXXROUTER_API_KEY='YOUR_SERVER_SIDE_KEY'
```

## Run one path

`chat` reads SSE deltas and final usage; `image` receives a synchronous image; `async-image` and `video` handle HTTP 202 and query the returned task. Polling is bounded to approximately five minutes plus any in-flight read; a timeout preserves the task for later reads.

```sh
python3 python.py chat
# python3 python.py image
# python3 python.py async-image
# python3 python.py video
```

## Idempotency and failure handling

A new run creates a new idempotency key and can incur a new charge. In an application, persist that key with the exact body before submitting. Set `XXXROUTER_REQUEST_ID` only to resume the same logical request with the same model, payload, API key and source. Do not rerun with a new key after a transport failure. These clients perform one generation POST and never retry it automatically.

Record the response `X-Request-ID`. HTTP errors retain the response body for diagnosis. A task with `status=unknown` stops automatic polling and requires later read/reconciliation; it is not a successful or free request. A stream error event or missing `[DONE]` is an interruption. Missing usage is not zero cost. Stopping the connection does not cancel accepted work. Do not log API keys or signed asset URLs.

## Image inputs and result files

For image-to-image or image-to-video, pass a public HTTPS URL in `image` when the model supports reference images. The public inference API takes URLs, not file bytes. Playground can upload temporary input images or videos to private R2 storage; only signed-in, verified users can use it. Generated results are never downloaded, stored or rehosted by XXXRouter. Successful responses return the original generated-media URL unchanged. The original host controls availability, expiry and browser cross-origin behavior. Save results before expiry; querying a completed task returns the saved URL and does not renew it. Request metadata and result URLs may be retained for recovery; generated-media bytes are never retained.

## Complete example

The code is an integration example, not proof that a live model or payment channel is enabled. Do not place it in a browser bundle.

```python
"""Python 3.11+, standard library only. No automatic generation retries."""
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid

BASE = os.environ['XXXROUTER_BASE_URL'].rstrip('/')  # Include /v1 exactly once.
KEY = os.environ['XXXROUTER_API_KEY']
MODE = sys.argv[1] if len(sys.argv) > 1 else 'chat'
CASES = {
    'chat': ('chat/completions', {'model': 'deepseek-v4-flash', 'messages': [{'role': 'user', 'content': 'Explain HTTP in one sentence.'}], 'max_tokens': 256, 'stream': True, 'stream_options': {'include_usage': True}}),
    'image': ('images/generations', {'model': 'seedream-4.5', 'prompt': 'A ceramic cup in natural light', 'size': '2048x2048', 'n': 1}),
    'async-image': ('images/generations', {'model': 'nano-banana-pro', 'prompt': 'A ceramic cup in natural light', 'size': '1024x1024', 'n': 1}),
    'video': ('videos/generations', {'model': 'kling-3.0', 'prompt': 'A slow camera move across a mountain landscape', 'duration': 5, 'aspect_ratio': '16:9'}),
}

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None

opener = urllib.request.build_opener(NoRedirect)

def request(path, payload=None, idem=None):
    headers = {'Authorization': 'Bearer ' + KEY}
    if idem:
        headers['Idempotency-Key'] = idem
    data = None
    if payload is not None:
        headers['Content-Type'] = 'application/json'
        data = json.dumps(payload).encode()
    req = urllib.request.Request(BASE + '/' + path, data=data, headers=headers)
    try:
        return opener.open(req, timeout=300 if payload is not None else 30)
    except urllib.error.HTTPError as error:
        # Read the structured failure: an uncertain request may already have a hold.
        print('HTTP', error.code, 'request_id=', error.headers.get('X-Request-ID'), file=sys.stderr)
        print(error.read().decode(), file=sys.stderr)
        raise

def stream(response):
    event = []
    def consume():
        data = '\n'.join(event)
        if data == '[DONE]':
            return True
        if data:
            chunk = json.loads(data)
            if chunk.get('error'):
                raise RuntimeError(chunk['error'])
            for choice in chunk.get('choices', []):
                print(choice.get('delta', {}).get('content') or '', end='', flush=True)
            if chunk.get('usage'):
                print('\nusage:', chunk['usage'])
        return False
    for raw in response:
        line = raw.decode('utf-8').rstrip('\r\n')
        if line == '':
            if consume():
                return
            event.clear()
        elif line.startswith('data:'):
            event.append(line[5:].lstrip(' '))
    if event and consume():
        return
    raise RuntimeError('Stream interrupted. Keep request ID and idempotency key; do not create another submission.')

def finish(result):
    deadline = time.monotonic() + 300
    while result.get('task_id') and result.get('status') not in ('succeeded', 'failed', 'unknown'):
        if time.monotonic() >= deadline:
            print(result)
            raise TimeoutError('Keep task_id and query later; do not resubmit.')
        time.sleep(2)
        with request('tasks/' + result['task_id']) as response:
            result = json.load(response)
    print(json.dumps(result, indent=2))
    if result.get('status') == 'unknown':
        raise RuntimeError('Outcome unconfirmed. Retain IDs and wait for reconciliation; do not resubmit.')
    if result.get('status') == 'failed':
        raise RuntimeError('Generation failed. Inspect error and billing before a new logical request.')

if __name__ == '__main__':
    path, payload = CASES[MODE]
    idem = os.environ.get('XXXROUTER_REQUEST_ID') or str(uuid.uuid4())
    print('idempotency_key=', idem, file=sys.stderr)
    # Persist the key with this exact payload before submitting in a real application.
    with request(path, payload, idem) as response:
        print('request_id=', response.headers.get('X-Request-ID'), file=sys.stderr)
        if 'text/event-stream' in response.headers.get('Content-Type', ''):
            stream(response)
        else:
            finish(json.load(response))
```

## Related reference

[chat](/docs?page=chat&lang=en) · [images](/docs?page=images&lang=en) · [videos](/docs?page=videos&lang=en) · [tasks](/docs?page=tasks&lang=en) · [uploads](/docs?page=uploads&lang=en) · [errors](/docs?page=errors&lang=en)

[Runtime documentation](https://docs.python.org/3/library/urllib.request.html) · 2026-10-26