Bolna

pip install fastapi uvicorn httpx geniffy

uv add fastapi uvicorn httpx geniffy

import hmac
import os

import httpx
from fastapi import FastAPI, HTTPException, Request
from geniffy import AsyncGeniffy

geniffy = AsyncGeniffy()                          # reads GENIFFY_API_KEY
bolna = httpx.AsyncClient(base_url="https://api.bolna.ai",
                          headers={"Authorization": f"Bearer {os.environ['BOLNA_API_KEY']}"})
TOKEN = os.environ["BOLNA_TOKEN"]
app = FastAPI()


def space_for(number: str | None) -> str:
    """The caller, by the digits of their number. If you know callers as customers, use your customer id, and
    the same memory serves their calls, chats and emails."""
    digits = "".join(ch for ch in number or "" if ch.isdigit())
    if not digits:                                # never fall back to one shared memory for unknown callers
        raise HTTPException(400, "This call has no caller number.")
    return f"user_{digits}"


def check(request: Request) -> None:
    """Bolna sends your token as a bearer token."""
    given = request.headers.get("Authorization", "").removeprefix("Bearer ")
    if not hmac.compare_digest(given.encode(), TOKEN.encode()):
        raise HTTPException(401)


async def known_about(number: str | None) -> str:
    brief = await geniffy.space(space_for(number)).brief(limit=12)
    return "\n".join(f"- {m['text']}" for m in brief["memories"]) or (
        "Nothing is known about this caller yet. If they mention an earlier call, say so rather than guessing.")


@app.get("/bolna/caller")
async def caller(request: Request, contact_number: str = ""):
    """Bolna asks this before it answers an incoming call; what it returns fills {{memory}} in the prompt."""
    check(request)
    return {"memory": await known_about(contact_number)}


async def place_call(number: str) -> str:
    """A call out that starts knowing the person it calls."""
    r = await bolna.post("/call", json={"agent_id": os.environ["BOLNA_AGENT_ID"], "recipient_phone_number": number,
                                        "user_data": {"memory": await known_about(number)}})
    r.raise_for_status()
    return r.json()["execution_id"]


@app.post("/bolna/calls")
async def calls(request: Request):
    """Each update on a call. The completed one carries the transcript; Bolna can send it more than once, so
    it is saved under the call's own id, and a call sent again is the same source."""
    check(request)
    call = await request.json()
    if call.get("status") != "completed":
        return {}
    turns: list[dict] = []
    for line in (call.get("transcript") or "").splitlines():
        role, colon, said = line.partition(":")
        if colon and role.strip() in ("assistant", "user"):
            turns.append({"role": role.strip(), "content": said.strip()})
        elif turns and line.strip():              # a line that carries on what was being said
            turns[-1]["content"] += " " + line.strip()
    turns = [t for t in turns if t["content"]]
    if turns:                                     # user_number: the caller, or the person called
        await geniffy.space(space_for(call.get("user_number"))).memories.add(
            messages=turns, title="Call", external_id=f"bolna-{call['id']}")
    return {}

You are a friendly phone assistant. Keep answers short.

<memory>
{{memory}}
</memory>