Vapi × MemoAir Voice Memory

pip install "memoair-vapi>=0.1.0" "fastapi>=0.110" "uvicorn>=0.27" python-dotenv

# MemoAir
MEMOAIR_API_KEY=memoair_pk_...
MEMOAIR_PROJECT_ID=proj_...
MEMOAIR_AGENT_ID=agent_...
MEMOAIR_SEARCH_TIMEOUT_MS=250
 
# Vapi webhook secret (from the Custom Knowledge Base config in step 6).
# For local dev you can skip verification instead:
VAPI_WEBHOOK_SECRET=your-webhook-secret
# VAPI_SKIP_SIGNATURE_VERIFY=1

from __future__ import annotations
 
import json
import os
from contextlib import asynccontextmanager
 
from dotenv import load_dotenv
from fastapi import FastAPI, Request, Response
from memoair_vapi import MemoAirVapiSearch, VapiWebhookError, verify_vapi_signature
 
load_dotenv()
 
WEBHOOK_SECRET = os.getenv("VAPI_WEBHOOK_SECRET", "")
SKIP_VERIFY = os.getenv("VAPI_SKIP_SIGNATURE_VERIFY", "0").lower() in {"1", "true", "yes"}
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # One search client per process. It owns the bundled voice-runtime and
    # pulls the org .mv2 snapshot on first use; aclose() tears it down.
    app.state.search = MemoAirVapiSearch(
        api_key=os.environ["MEMOAIR_API_KEY"],
        project_id=os.environ["MEMOAIR_PROJECT_ID"],
        agent_id=os.environ["MEMOAIR_AGENT_ID"],
        search_timeout_ms=int(os.getenv("MEMOAIR_SEARCH_TIMEOUT_MS", "250")),
    )
    try:
        yield
    finally:
        await app.state.search.aclose()
 
 
app = FastAPI(lifespan=lifespan)
 
 
@app.get("/health")
async def health() -> dict:
    return {"status": "ok"}
 
 
@app.post("/vapi/webhook")
async def vapi_webhook(request: Request) -> Response:
    raw_body = await request.body()
 
    if not SKIP_VERIFY:
        sig = request.headers.get("x-vapi-signature", "")
        if not verify_vapi_signature(raw_body, sig, WEBHOOK_SECRET):
            return Response(status_code=401, content='{"error":"invalid signature"}',
                            media_type="application/json")
 
    try:
        payload = json.loads(raw_body or b"{}")
    except json.JSONDecodeError:
        return Response(status_code=400, content='{"error":"invalid json"}',
                        media_type="application/json")
 
    try:
        # Non-knowledge-base messages are acked with {}. Retrieval failures
        # degrade to {"documents": []} so a live call never breaks.
        result = await request.app.state.search.handle_request(payload)
    except VapiWebhookError:
        return Response(status_code=400, content='{"error":"malformed payload"}',
                        media_type="application/json")
 
    return Response(status_code=200, content=json.dumps(result),
                    media_type="application/json")

uvicorn memoair_vapi_server:app --port 8000
# In another terminal, expose it (Vapi's cloud must reach your webhook):
ngrok http 8000   # -> https://<subdomain>.ngrok-free.dev

# 1. Create the custom KB -> returns an id
curl -X POST https://api.vapi.ai/knowledge-base \
  -H "Authorization: Bearer $VAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "provider": "custom-knowledge-base",
    "server": {
      "url": "https://<subdomain>.ngrok-free.dev/vapi/webhook",
      "secret": "your-webhook-secret"
    }
  }'
 
# 2. Attach the returned id to your assistant (send the FULL model object)
curl -X PATCH https://api.vapi.ai/assistant/$ASSISTANT_ID \
  -H "Authorization: Bearer $VAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": {
      "provider": "openai",
      "model": "gpt-4o",
      "knowledgeBaseId": "<KB_ID_FROM_STEP_1>",
      "messages": [
        { "role": "system", "content": "Answer using the knowledge base context provided." }
      ]
    }
  }'

# Tail the server log — every turn shows POST /vapi/webhook 200
# Or smoke-test the webhook directly:
curl -X POST https://<subdomain>.ngrok-free.dev/vapi/webhook \
  -H 'Content-Type: application/json' \
  -d '{"message":{"type":"knowledge-base-request",
       "messages":[{"role":"user","content":"What is your return policy?"}]}}'
# -> {"documents": [{"content": "...", "similarity": 0.9, "uuid": "..."}]}