Alright, let's cut through the fluff. You want a Poe bot that hits a public API and serves fresh data. Most tutorials give you a glorified "Hello World." I'll give you a production-grade pattern. The core concept is simple: your bot is a stateless query engine. The real challenge is handling API failures, structuring the response, and not getting rate-limited.
Here's the architecture we're building:
* Poe bot as the interface layer.
* A dedicated service class to encapsulate API logic and error handling.
* Explicit schema definition for the output.
We'll use the USGS Earthquake API as our example. It's public, reliable, and demonstrates key points.
**Step 1: The Bot Skeleton**
Your bot code needs to handle the incoming query, delegate the work, and format the reply. No API calls happen here.
```python
import asyncio
from typing import AsyncIterable
from fastapi_poe import PoeBot
from earthquake_service import EarthquakeService
class RealtimeDataBot(PoeBot):
def __init__(self):
# Initialize your service layer
self.service = EarthquakeService()
async def get_response(self, query: dict) -> AsyncIterable[str]:
# Extract the user's message
user_message = query["query"][0]["content"].strip()
# Delegate to the service layer
try:
data = await self.service.fetch_earthquake_data(user_message)
yield self.service.format_response(data)
except Exception as e:
# Never expose raw errors to the user
yield f"Error retrieving data. Please try again or rephrase your query."
```
**Step 2: The Service Layer (This is where the real work happens)**
This is the battle-tested part. Notice the retry logic, timeout, and schema validation.
```python
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential
from pydantic import BaseModel
from typing import List, Optional
# Define your output schema first. This is non-negotiable.
class Earthquake(BaseModel):
magnitude: float
location: str
time: str
depth_km: float
class EarthquakeService:
BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def _call_api(self, params: dict) -> dict:
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(self.BASE_URL, params=params) as response:
response.raise_for_status()
return await response.json()
async def fetch_earthquake_data(self, user_input: str) -> List[Earthquake]:
# Parse user input for magnitude threshold. Default logic.
try:
min_mag = float(user_input)
except ValueError:
min_mag = 2.5
params = {
"format": "geojson",
"starttime": "2024-01-01",
"minmagnitude": min_mag,
"orderby": "time",
"limit": 10
}
raw_data = await self._call_api(params)
earthquakes = []
for feature in raw_data.get('features', []):
props = feature['properties']
geometry = feature['geometry']
eq = Earthquake(
magnitude=props['mag'],
location=props['place'],
time=props['time'],
depth_km=geometry['coordinates'][2]
)
earthquakes.append(eq)
return earthquakes
def format_response(self, earthquakes: List[Earthquake]) -> str:
if not earthquakes:
return "No recent earthquakes found matching your criteria."
lines = [f"**Last {len(earthquakes)} earthquakes (M{earthquakes[0].magnitude}+):**"]
for eq in earthquakes:
lines.append(f"- M{eq.magnitude} | {eq.location} | Depth: {eq.depth_km}km")
return "n".join(lines)
```
**Key Takeaways & Pitfalls:**
* **Separation of Concerns:** The bot handles conversation, the service handles data. This lets you test and change the API logic independently.
* **Resilience:** Retries with exponential backoff are mandatory for any external dependency. Never assume the API is up.
* **Timeouts:** Always set them. A hanging API call will kill your bot's responsiveness.
* **Schema Validation:** Using Pydantic (or similar) catches API contract changes early. Don't just assume the JSON structure.
* **Rate Limiting:** This API is lenient, but for others, you must track calls per user/session. Implement a token bucket or similar in your service layer.
The biggest mistake I see? People shoving all the logic directly into the bot's `get_response` method. It becomes an unmaintainable, untestable mess. This pattern scales. Swap out `EarthquakeService` for `StockService` or `WeatherService`. The bot stays clean.
garbage in, garbage out