Workflow 001: Setup External Application
Fresh 🌱Workflow ID: WF-001
Version: 1.0
Status: Active
Overview
This workflow guides you through setting up a complete external application that uses the Poe API. This includes project structure, error handling, rate limiting, and best practices.
Prerequisites
- [x] API key obtained (SOP 001)
- [x] fastapi-poe installed (SOP 002)
- Python 3.7+ installed
- Text editor or IDE
Workflow Steps
┌─────────────────────────────────────────────────────────────┐
│ EXTERNAL APPLICATION SETUP │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. Project Structure ──► 2. Configuration │
│ │ │ │
│ └──► 3. Error Handling ──┘ │
│ │ │ │
│ └──► 4. Rate Limiting ──┘ │
│ │ │ │
│ └──► 5. Testing ─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘Step 1: Create Project Structure
Create a new directory and files:
bash
mkdir my-poe-app
cd my-poe-appCreate the following structure:
my-poe-app/
├── .env # API key (add to .gitignore!)
├── .gitignore
├── requirements.txt
├── config.py
├── poe_client.py # Main API client
└── main.py # Application entry pointStep 2: Setup Configuration
.env file:
bash
POE_API_KEY=your_api_key_here.gitignore:
.env
__pycache__/
*.pyc
venv/requirements.txt:
fastapi-poe>=0.1.0
python-dotenv>=1.0.0config.py:
python
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
POE_API_KEY = os.getenv("POE_API_KEY")
if not POE_API_KEY:
raise ValueError("POE_API_KEY not found in environment variables")
DEFAULT_BOT = "GPT-5"
RATE_LIMIT_PER_MINUTE = 500Step 3: Create API Client
poe_client.py:
python
import fastapi_poe as fp
import asyncio
from typing import List, Optional
from config import Config
class PoeClient:
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or Config.POE_API_KEY
self.default_bot = Config.DEFAULT_BOT
async def query(
self,
message: str,
bot_name: Optional[str] = None,
parameters: Optional[dict] = None
) -> str:
"""Query a Poe bot asynchronously."""
bot_name = bot_name or self.default_bot
protocol_message = fp.ProtocolMessage(
role="user",
content=message,
parameters=parameters or {}
)
response = ""
async for partial in fp.get_bot_response(
messages=[protocol_message],
bot_name=bot_name,
api_key=self.api_key
):
response += partial
return response
def query_sync(
self,
message: str,
bot_name: Optional[str] = None,
parameters: Optional[dict] = None
) -> str:
"""Query a Poe bot synchronously."""
bot_name = bot_name or self.default_bot
protocol_message = fp.ProtocolMessage(
role="user",
content=message,
parameters=parameters or {}
)
response = ""
for partial in fp.get_bot_response_sync(
messages=[protocol_message],
bot_name=bot_name,
api_key=self.api_key
):
response += partial
return response
async def query_with_history(
self,
messages: List[fp.ProtocolMessage],
bot_name: Optional[str] = None
) -> str:
"""Query with conversation history."""
bot_name = bot_name or self.default_bot
response = ""
async for partial in fp.get_bot_response(
messages=messages,
bot_name=bot_name,
api_key=self.api_key
):
response += partial
return responseStep 4: Add Error Handling
Update poe_client.py with error handling:
python
import fastapi_poe as fp
import asyncio
from typing import List, Optional
from config import Config
class PoeError(Exception):
"""Base exception for Poe API errors."""
pass
class PoeRateLimitError(PoeError):
"""Rate limit exceeded."""
pass
class PoeInsufficientPointsError(PoeError):
"""Insufficient points in account."""
pass
class PoeClient:
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or Config.POE_API_KEY
self.default_bot = Config.DEFAULT_BOT
def _handle_error(self, error: Exception):
"""Handle API errors."""
error_msg = str(error).lower()
if "rate limit" in error_msg:
raise PoeRateLimitError("Rate limit exceeded. Wait 60 seconds.")
elif "insufficient" in error_msg or "points" in error_msg:
raise PoeInsufficientPointsError("Insufficient points in account.")
else:
raise PoeError(f"API error: {error}")
async def query(
self,
message: str,
bot_name: Optional[str] = None,
parameters: Optional[dict] = None
) -> str:
"""Query a Poe bot asynchronously with error handling."""
try:
bot_name = bot_name or self.default_bot
protocol_message = fp.ProtocolMessage(
role="user",
content=message,
parameters=parameters or {}
)
response = ""
async for partial in fp.get_bot_response(
messages=[protocol_message],
bot_name=bot_name,
api_key=self.api_key
):
response += partial
return response
except Exception as e:
self._handle_error(e)Step 5: Create Main Application
main.py:
python
import asyncio
from poe_client import PoeClient, PoeError, PoeRateLimitError
async def main():
client = PoeClient()
try:
print("Querying Poe API...")
response = await client.query("What is Python?")
print(f"\nResponse:\n{response}")
except PoeRateLimitError as e:
print(f"Rate limit error: {e}")
print("Waiting 60 seconds...")
await asyncio.sleep(60)
except PoeError as e:
print(f"Error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
if __name__ == "__main__":
asyncio.run(main())Step 6: Install and Test
bash
# Install dependencies
pip install -r requirements.txt
# Run the application
python main.pyVerification Checklist
- [ ] Project structure created
- [ ] Configuration files set up
- [ ] API client implemented
- [ ] Error handling added
- [ ] Application runs without errors
- [ ] API responses received successfully
Advanced: Rate Limiting
Add rate limiting to prevent exceeding 500 requests/minute:
python
import time
from collections import deque
class RateLimiter:
def __init__(self, max_requests: int = 500, window: int = 60):
self.max_requests = max_requests
self.window = window
self.requests = deque()
async def wait_if_needed(self):
"""Wait if rate limit would be exceeded."""
now = time.time()
# Remove old requests outside window
while self.requests and self.requests[0] < now - self.window:
self.requests.popleft()
# If at limit, wait
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())Troubleshooting
Problem: Import errors
Solution:
- Verify all dependencies installed:
pip install -r requirements.txt - Check Python path
- Use virtual environment
Problem: API key not found
Solution:
- Verify
.envfile exists - Check
.envis in.gitignore - Verify
python-dotenvinstalled
Problem: Rate limit errors
Solution:
- Implement rate limiting (see above)
- Reduce request frequency
- Use async with delays
Next Steps
Next: 002: Query Multiple Bots