WhatsApp Assistant

from fastapi import FastAPI, Request, Response
from agent import agent
import os
import uvicorn
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()

VERIFY_TOKEN = os.getenv("WHATSAPP_VERIFY_TOKEN")

@app.get("/")
async def verify(request: Request):
    token = request.query_params.get("hub.verify_token")
    challenge = request.query_params.get("hub.challenge")
    if token == VERIFY_TOKEN and challenge:
        return Response(content=challenge, media_type="text/plain")
    return Response(content="Verification failed", status_code=403)

@app.post("/")
async def webhook(request: Request):
    _webhook = await request.json()
    await agent(_webhook=_webhook).collect()
    return {"status": "ok"}

def _maybe_start_ngrok(port: int) -> None:
    enable = os.getenv("ENABLE_NGORK", "false")
    if enable != "true":
        return
    try:
        from pyngrok import ngrok
        public_url = ngrok.connect(port, "http")
        print(f"ngrok public url: {public_url}")
    except Exception as e:
        print(f"Failed to start ngrok: {e}")

if __name__ == "__main__":
    host = os.getenv("HOST", "0.0.0.0")
    try:
        port = int(os.getenv("PORT", "4343"))
    except Exception:
        port = 4343
    _maybe_start_ngrok(port)
    uvicorn.run("app:app", host=host, port=port, log_level="info")

import os
import json
from timbal import Agent
from timbal.state import get_run_context
from whatsapp_tools import send_whatsapp_message

def _value(payload: dict) -> dict | None:
    try:
        return payload["entry"][0]["changes"][0]["value"]
    except Exception:
        return None

def _prompt_from_message(msg: dict) -> str:
    t = msg.get("type")
    if t == "text":
        return (msg.get("text") or {}).get("body", "").strip()
    if t == "image":
        mid = (msg.get("image") or {}).get("id")
        cap = (msg.get("image") or {}).get("caption", "")
        return f"[image:{mid}] {cap}".strip()
    if t == "audio":
        mid = (msg.get("audio") or {}).get("id")
        return f"[audio:{mid}]"
    if t == "document":
        fn = (msg.get("document") or {}).get("filename", "document")
        return f"[document:{fn}]"
    return f"[{t or 'unknown'}]"

def _append_jsonl(path: str, rec: dict) -> None:
    try:
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception:
        pass

def _recent_history(path: str, user_phone: str, limit: int = 10) -> list[str]:
    out = []
    try:
        if not os.path.exists(path):
            return []
        with open(path, "r", encoding="utf-8") as f:
            rows = [json.loads(l) for l in f if l.strip()]
            for r in reversed(rows):
                if r.get("user_phone") == user_phone:
                    try:
                        payload = json.loads(r.get("message") or "{}")
                        out.append(f"{r.get('direction')}: {payload.get('content','')}")
                    except Exception:
                        continue
                    if len(out) >= limit:
                        break
        return list(reversed(out))
    except Exception:
        return []

def pre_hook():
    """Build professional prompt from WhatsApp metadata, message, and optional history."""
    span = get_run_context().current_span()
    payload = span.input.get("_webhook")
    if not payload:
        bail("No payload found in webhook")

    value = payload["entry"][0]["changes"][0]["value"]
    if not value:
        bail("No value found in webhook")

    messages = value.get("messages", [])
    if not messages:
        bail("No messages found in webhook")

    contacts = value.get("contacts", [])
    profile = contacts[0].get("profile", {})
    name = profile.get("name", "")
    msg = messages[0]
    from_number = msg.get("from")
    message_id = msg.get("id")
    phone_number_id = value.get("metadata").get("phone_number_id")
    prompt = _prompt_from_message(msg)
    msg_type = msg.get("type")

    rec = {
        "id": msg.get("id"),
        "user_phone": from_number,
        "direction": "inbound",
        "message_type": msg_type,
        "message": json.dumps({"type": msg_type, "content": prompt}, ensure_ascii=False),
        "timestamp": msg.get("timestamp"),
        "conversation_id": f"{phone_number_id}_{from_number}",
        "user_name": name,
    }
    _append_jsonl("whatsapp_messages.jsonl", rec)

    span.input["prompt"] = _recent_history("whatsapp_messages.jsonl", user_phone=from_number, limit=10)
    span.input["whatsapp_from_number"] = from_number
    span.input["name"] = name

def post_hook():
    """Send the agent response via WhatsApp and persist outbound to JSONL."""
    span = get_run_context().current_span()
    from_number = span.input.get("whatsapp_from_number")
    if not from_number:
        return
    try:
        response_text = (span.output.collect_text() or "").strip()
    except Exception:
        return
    if not response_text:
        return
    try:
        send_whatsapp_message(to=from_number, message=response_text)
    except Exception:
        pass
    _append_jsonl("whatsapp_messages.jsonl", {
        "id": f"outbound_{from_number}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
        "user_phone": from_number,
        "direction": "outbound",
        "message_type": "agent",
        "message": json.dumps({"type": "text", "content": response_text}, ensure_ascii=False),
        "timestamp": datetime.now().isoformat(),
        "conversation_id": "",
        "user_name": span.input.get("name", ""),
    })

agent = Agent(
    name="WhatsAppAgent",
    description="A minimal WhatsApp agent for Timbal",
    model="openai/gpt-4.1-mini",
    pre_hook=pre_hook,
    post_hook=post_hook,
)