Commands Reference

Fresh 🌱

Quick reference for common code patterns and commands.

Installation

# Install fastapi-poe
pip install fastapi-poe

# Or with requirements.txt
pip install -r requirements.txt

Basic Imports

import fastapi_poe as fp
import os

API Key Setup

# From environment variable (recommended)
api_key = os.getenv("POE_API_KEY")

# Direct assignment (not recommended for production)
api_key = "your_api_key_here"

Synchronous Query

message = fp.ProtocolMessage(role="user", content="Hello")

for partial in fp.get_bot_response_sync(
    messages=[message],
    bot_name="GPT-5",
    api_key=api_key
):
    print(partial, end="", flush=True)

Asynchronous Query

import asyncio

async def query():
    message = fp.ProtocolMessage(role="user", content="Hello")
    async for partial in fp.get_bot_response(
        messages=[message],
        bot_name="GPT-5",
        api_key=api_key
    ):
        print(partial, end="", flush=True)

asyncio.run(query())

Upload and Send File

# Synchronous
with open("document.pdf", "rb") as f:
    attachment = fp.upload_file_sync(f, api_key=api_key)

message = fp.ProtocolMessage(
    role="user",
    content="Analyze this",
    attachments=[attachment]
)

# Asynchronous
async def upload_and_query():
    with open("document.pdf", "rb") as f:
        attachment = await fp.upload_file(f, api_key=api_key)
    
    message = fp.ProtocolMessage(
        role="user",
        content="Analyze this",
        attachments=[attachment]
    )

Custom Parameters

# Claude with thinking budget
message = fp.ProtocolMessage(
    role="user",
    content="Explain quantum computing",
    parameters={"thinking_budget": 12288}
)

# GPT with reasoning effort
message = fp.ProtocolMessage(
    role="user",
    content="Solve this problem",
    parameters={"reasoning_effort": "high"}
)

# Image generation with aspect ratio
message = fp.ProtocolMessage(
    role="user",
    content="A cat in a hat",
    parameters={"aspect_ratio": "4:3"}
)

Conversation History

messages = [
    fp.ProtocolMessage(role="user", content="Hello"),
    fp.ProtocolMessage(role="assistant", content="Hi there!"),
    fp.ProtocolMessage(role="user", content="What's the weather?")
]

for partial in fp.get_bot_response_sync(
    messages=messages,
    bot_name="GPT-5",
    api_key=api_key
):
    print(partial, end="", flush=True)

OpenAI-Compatible API

from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("POE_API_KEY"),
    base_url="https://api.poe.com/v1",
)

response = client.chat.completions.create(
    model="Claude-Opus-4.1",
    messages=[{"role": "user", "content": "Hello"}],
)

print(response.choices[0].message.content)

With Leaderboard Headers

from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("POE_API_KEY"),
    base_url="https://api.poe.com/v1",
)

response = client.chat.completions.create(
    extra_headers={
        "HTTP-Referer": "https://my-app.com",
        "X-Title": "My App",
    },
    model="Claude-Opus-4.1",
    messages=[{"role": "user", "content": "Hello"}],
)

Error Handling

try:
    for partial in fp.get_bot_response_sync(
        messages=[message],
        bot_name="GPT-5",
        api_key=api_key
    ):
        print(partial, end="", flush=True)
except Exception as e:
    if "rate limit" in str(e).lower():
        print("Rate limit exceeded. Wait 60 seconds.")
    elif "insufficient" in str(e).lower():
        print("Insufficient points in account.")
    else:
        print(f"Error: {e}")

Rate Limiting Helper

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests=500, window=60):
        self.max_requests = max_requests
        self.window = window
        self.requests = deque()
    
    async def wait_if_needed(self):
        now = time.time()
        while self.requests and self.requests[0] < now - self.window:
            self.requests.popleft()
        
        if len(self.requests) >= self.max_requests:
            sleep_time = self.window - (now - self.requests[0])
            if sleep_time > 0:
                await asyncio.sleep(sleep_time)
        
        self.requests.append(time.time())

Query Multiple Bots

import asyncio

async def query_all(bots, question, api_key):
    tasks = []
    for bot in bots:
        message = fp.ProtocolMessage(role="user", content=question)
        task = fp.get_bot_response(
            messages=[message],
            bot_name=bot,
            api_key=api_key
        )
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results

Complete Client Class

import fastapi_poe as fp
import os
from typing import Optional, List

class PoeClient:
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("POE_API_KEY")
    
    def query_sync(
        self,
        message: str,
        bot_name: str = "GPT-5",
        parameters: Optional[dict] = None
    ) -> str:
        msg = fp.ProtocolMessage(
            role="user",
            content=message,
            parameters=parameters or {}
        )
        
        response = ""
        for partial in fp.get_bot_response_sync(
            messages=[msg],
            bot_name=bot_name,
            api_key=self.api_key
        ):
            response += partial
        
        return response
    
    async def query(
        self,
        message: str,
        bot_name: str = "GPT-5",
        parameters: Optional[dict] = None
    ) -> str:
        msg = fp.ProtocolMessage(
            role="user",
            content=message,
            parameters=parameters or {}
        )
        
        response = ""
        async for partial in fp.get_bot_response(
            messages=[msg],
            bot_name=bot_name,
            api_key=self.api_key
        ):
            response += partial
        
        return response

Environment Variables

# .env file
POE_API_KEY=your_api_key_here

# Load in Python
from dotenv import load_dotenv
load_dotenv()

Common Patterns

Pattern: Streaming Response

response = ""
for partial in fp.get_bot_response_sync(...):
    response += partial
    print(partial, end="", flush=True)

Pattern: Collect Full Response

response = ""
async for partial in fp.get_bot_response(...):
    response += partial
# Use full response

Pattern: With Retry

import time

def query_with_retry(message, bot_name, api_key, max_retries=3):
    for attempt in range(max_retries):
        try:
            for partial in fp.get_bot_response_sync(
                messages=[message],
                bot_name=bot_name,
                api_key=api_key
            ):
                print(partial, end="", flush=True)
            break
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise

---

Next: Troubleshooting