Building a Mercado Livre Reply Bot with Claude AI: 28-Second Response Time, 24/7

Mercado Livre API ──polling──► FoxReply Worker
                                      │
                               Claude API (Sonnet)
                                      │
                               ML Answer Endpoint

GET /questions/search?seller_id={seller_id}&status=UNANSWERED&limit=50
Authorization: Bearer {access_token}

import asyncio
from datetime import datetime, time

async def poll_interval() -> int:
    """Return polling interval in seconds based on time of day."""
    now = datetime.now().time()
    # Business hours: every 30s
    if time(8, 0) <= now <= time(22, 0):
        return 30
    # Night: every 5 minutes (questions are rare, SLA resets at 8am anyway)
    return 300

async def generate_reply(question: str, product: dict, tone: str = "professional") -> str:
    prompt = f"""You are a helpful store assistant for a Brazilian e-commerce seller.

Product: {product['title']}
Description: {product['description'][:500]}
Attributes: {format_attributes(product['attributes'])}

Customer question: {question}

Reply in Brazilian Portuguese, {tone} tone, max 3 sentences.
Be specific — reference the product details above.
Never say "I don't know" — if unsure, offer to check and get back."""

    response = await anthropic.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=200,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

class MLTokenManager:
    def __init__(self, client_id: str, client_secret: str, refresh_token: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self._access_token: str | None = None
        self._expires_at: float = 0

    async def get_token(self) -> str:
        # Refresh 5 minutes before expiry (not after)
        if time.time() >= self._expires_at - 300:
            await self._refresh()
        return self._access_token

    async def _refresh(self) -> None:
        resp = await httpx.post(
            "https://api.mercadolibre.com/oauth/token",
            data={
                "grant_type": "refresh_token",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "refresh_token": self.refresh_token,
            }
        )
        data = resp.json()
        self._access_token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        # IMPORTANT: ML rotates refresh tokens — always save the new one
        self.refresh_token = data["refresh_token"]
        await save_refresh_token(self.refresh_token)  # persist to DB

async def should_reply(question_id: int) -> bool:
    key = f"replied:{question_id}"
    # SET NX EX — atomic check-and-set, expire after 7 days
    result = await redis.set(key, "1", nx=True, ex=604800)
    return result is True  # True = we set it (first time), False = already exists

async def process_questions(questions: list[dict]) -> None:
    for q in questions:
        if not await should_reply(q["id"]):
            continue  # already replied or in-flight
        reply = await generate_reply(q["text"], q["item"])
        await post_reply(q["id"], reply)

class MarketplaceAdapter(ABC):
    @abstractmethod
    async def get_unanswered_questions(self) -> list[Question]:
        ...

    @abstractmethod
    async def post_reply(self, question_id: str, text: str) -> bool:
        ...

class MercadoLivreAdapter(MarketplaceAdapter):
    # ML-specific polling + OAuth
    ...

class NuvemshopAdapter(MarketplaceAdapter):
    # Nuvemshop webhook-based (they DO have webhooks)
    ...