---
name: api-media-responses
description: Patterns for building APIs that return media content (images, files, binaries). Covers FastAPI/Flask response types, Pillow image generation, multi-file returns, and media streaming.
trigger:
  - user asks to "create API" + "return image" / "return file" / "return binary"
  - user asks to "generate image" in an API endpoint
  - task involves serving generated media (PNG, JPEG, PDF, ZIP) from a web endpoint
  - user mentions FastAPI + Pillow / image generation
---

# API Media Responses

Quick patterns for APIs that generate and return media content (images, files, binaries).

## Core Pattern: FastAPI Return Image

```python
from fastapi import Response
from PIL import Image
import io

@app.get("/endpoint")
async def return_image():
    # Generate image with Pillow
    img = Image.new('RGB', (512, 512), (255, 100, 100))
    
    # Save to bytes
    img_byte_arr = io.BytesIO()
    img.save(img_byte_arr, format='PNG')
    img_byte_arr.seek(0)
    
    # Return as response
    return Response(
        content=img_byte_arr.getvalue(),
        media_type="image/png",
        headers={"X-Custom-Header": "value"}
    )
```

## Multiple Images Strategies

**Option 1: Index-based endpoints** (recommended for simple cases)
```python
@app.get("/images/{index}")
async def get_image(index: int):
    colors = [(255,0,0), (0,255,0), (0,0,255)]
    if 1 <= index <= len(colors):
        img = Image.new('RGB', (512, 512), colors[index-1])
        # ... return Response
```

**Option 2: Base64 JSON** (for programmatic clients)
```python
import base64
img_base64 = base64.b64encode(img_bytes).decode('utf-8')
return {"count": 3, "images": [{"index": 1, "data": img_base64}]}
```

**Option 3: Zip file** (for batch downloads)
```python
import zipfile
from fastapi.responses import StreamingResponse

zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w') as zf:
    for i, img in enumerate(images):
        zf.writestr(f"image_{i}.png", img.getvalue())
zip_buffer.seek(0)
return StreamingResponse(zip_buffer, media_type='application/zip')
```

## Service-to-Service Media Transfer

When one API service needs to call another service that generates media (images, files), you must ensure format consistency between producer and consumer.

**Option A: Binary Stream (Producer → Consumer direct pass-through)**
```python
# Producer: AI service returns binary
@app.post("/inference")
async def inference():
    image_bytes = generate_image()
    return StreamingResponse(io.BytesIO(image_bytes), media_type="image/png")

# Consumer: Backend calls producer and forwards
async with httpx.AsyncClient() as client:
    response = await client.get(f"{AI_SERVICE_URL}/inference")
    if response.status_code == 200:
        # Forward binary directly
        return Response(content=response.content, media_type="image/png")
```

**Option B: Base64 JSON (Recommended for programmatic clients)**
```python
# Producer: AI service returns JSON with base64
@app.post("/inference")
async def inference():
    image_bytes = generate_image()
    image_base64 = base64.b64encode(image_bytes).decode('utf-8')
    return {"image_base64": image_base64, "metadata": {...}}

# Consumer: Backend parses JSON and extracts base64
async with httpx.AsyncClient() as client:
    response = await client.post(f"{AI_SERVICE_URL}/inference", params={...})
    if response.status_code == 200:
        data = response.json()
        image_base64 = data["image_base64"]
        # Use image_base64 directly or decode
```

**Pitfall: Format mismatch**
- If Producer returns binary but Consumer expects JSON (or vice versa), the call will fail
- Example error: Consumer tries `response.json()` but gets binary → `JSONDecodeError`
- Fix: Ensure both sides agree on format; if using Stability AI or similar external API that returns base64, convert to JSON at the producer side

**Stability AI Integration Pattern**
```python
# Stability AI returns base64 in JSON: {"artifacts": [{"base64": "..."}]}
response = await client.post(
    "https://api.stability.ai/v1/generation/stable-diffusion-xl-1024-v1-0/text-to-image",
    headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
    json={"text_prompts": [{"text": prompt, "weight": 1}], "cfg_scale": 7, ...},
    timeout=60.0
)
data = response.json()
image_base64 = data["artifacts"][0]["base64"]
image_bytes = base64.b64decode(image_base64)
# Now either return binary or re-encode to base64 JSON
```

**Docker Compose Service Discovery**
```yaml
# docker-compose.yml
services:
  backend:
    environment:
      - AI_SERVICE_URL=http://ai-service:8001
    depends_on:
      - ai-service
  ai-service:
    # ...
```
Backend uses `http://ai-service:8001` (container name) for inter-service calls.

## Environment Setup Pitfalls

**Pitfall: No pip in environment**
```python
# Download get-pip.py and bootstrap
import urllib.request
urllib.request.urlretrieve("https://bootstrap.pypa.io/get-pip.py", "/tmp/get-pip.py")
subprocess.run([sys.executable, "/tmp/get-pip.py"])
```

**Pitfall: Port already in use**
```python
# Check port before starting
result = subprocess.run(["lsof", "-i", ":8000"], capture_output=True)
if result.stdout:
    print("Port occupied, use different port")
    # Switch to 8001, 8002, etc.
```

## Testing Media Endpoints

```bash
# Save response to file and verify
curl -o output.png -w "%{http_code}" http://localhost:8000/endpoint
file output.png  # Should show "PNG image data"

# Check headers
curl -I http://localhost:8000/endpoint
```

## FastAPI with Pillow Dependencies

```bash
pip install fastapi uvicorn pillow
```

Minimal working server:
```python
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```