PyAI + ContextDB
Check customer memory before a PyAI agent books or changes an account.
PyAI Omni runs the realtime voice agent. Put the action boundary in a backend tool endpoint that authenticates the call, checks current host state, evaluates memory policy, executes the action, and reports the result.
evaluate_action before bookings and account changes, then
close the decision with report_execution.
recall_for_action is a compatibility retrieval helper, not
host authorization. Examples use the public Cloud client against
ContextDB Cloud. The local engine
uses pycontextdb. This independent guide has no vendor
partnership or endorsement.
Runnable starter
Run the PyAI Omni starter.
The starter connects the official PyAI Twilio bridge to ContextDB recall. Keep action checks in the backend tool pattern below.
npx degit atomsai/contextdb-clients/starters contextdb-starters cd contextdb-starters/pyai-omni npm ci cp .env.example .env # Add your PyAI, Twilio, OpenAI, and ContextDB server keys. set -a; source .env; set +a npm start
Inspect the starter and locked package versions → Try a memory scenario in Playground →
The tool endpoint
Use one backend endpoint before every booking or account change.
Configure the PyAI tool to send the intended action, not the ContextDB
partition. Your backend resolves the customer from a verified call,
checks the actor and current account state, then handles
act, ask, and abstain
explicitly.
# backend endpoint your PyAI agent tool calls
from contextdb_cloud_client import CloudClient
from fastapi import HTTPException, Request
@app.post("/pyai/tools/gate-action")
async def gate_action(req: Request):
tool_call = await authenticate_pyai_tool_call(req)
caller = await identities.customer_for_call(tool_call.call_id)
intent = validate_account_action(tool_call.arguments["intent"])
account = await accounts.get(caller)
if not await accounts.can_apply(
actor_id=tool_call.actor_id, account=account, intent=intent
):
raise HTTPException(status_code=403, detail="not authorized")
async with CloudClient(BASE_URL, api_key=KEY) as cdb:
decision = await cdb.evaluate_action(caller, intent)
if decision.outcome == "act":
try:
result = await accounts.apply(
caller, intent, expected_version=account.version
)
except CurrentStateChanged:
await cdb.report_execution(
caller, decision.decision_id, "account.apply", "failed",
idempotency_key=f"pyai-receipt-{decision.decision_id}",
error_code="current_state_changed",
)
return {"decision": "abstain", "say": "The account changed, so I did not apply that."}
await cdb.report_execution(
caller, decision.decision_id, "account.apply", "succeeded",
idempotency_key=f"pyai-receipt-{decision.decision_id}",
external_ref=result.ref,
)
return {"decision": "act", "reference": result.ref}
if decision.outcome == "ask":
await cdb.report_execution(
caller, decision.decision_id, "account.apply", "skipped",
idempotency_key=f"pyai-receipt-{decision.decision_id}",
)
return {"decision": "ask", "say": "Please confirm the account change."}
if decision.outcome == "abstain":
await cdb.report_execution(
caller, decision.decision_id, "account.apply", "skipped",
idempotency_key=f"pyai-receipt-{decision.decision_id}",
)
return {"decision": "abstain", "say": "I cannot make that account change."}
raise RuntimeError("unknown ContextDB action outcome")
After the call
The completion handler saves what the customer said.
When PyAI reports the call finished, authenticate the event and bind
every extracted fact to an attributed transcript turn. Derive
provenance from that turn. Reject a fact whose speaker cannot be
established instead of labeling every extraction
user_stated.
SOURCE_BY_SPEAKER = {
"user": "user_stated",
"agent": "agent_inferred",
"third_party": "third_party",
}
@app.post("/pyai/on-call-complete")
async def on_call_complete(req: Request):
call = await authenticate_pyai_completion(req)
caller = await identities.customer_for_call(call.id)
async with CloudClient(BASE_URL, api_key=KEY) as cdb:
for fact in call.extracted_facts:
turn = require_attributed_turn(call.transcript, fact.turn_id)
source = SOURCE_BY_SPEAKER.get(turn.speaker)
if source is None:
raise HTTPException(
status_code=422,
detail="fact provenance requires a known speaker",
)
await cdb.remember(
caller,
fact.content,
source=source,
confidence=fact.confidence,
idempotency_key=f"pyai-{call.id}-fact-{fact.id}",
)
return {"ok": True}
PyAI call order
Add memory at four points in a PyAI call.
| Moment | PyAI surface | ContextDB call |
|---|---|---|
| Call connects | Agent context / greeting variables | recall → caller snapshot |
| Booking, refund, account change | Agent tool → your backend | evaluate_action → branch, then report_execution |
| Caller confirms | Agent tool → your backend | Host authenticates and retains the end-user attestation, then confirm records the scoped memory update |
| Call completes | Completion handler | remember with source and confidence |
PyAI implementation
Why the PyAI action gate stays in your backend.
Why gate in my backend instead of prompting the agent to be careful?
Prompts drift and models comply with confident-sounding transcripts.
evaluate_action records a policy result outside the
model. Your backend still authenticates the actor, authorizes the
operation against current state, and reports what it did.
Does memory work across voice and chat channels?
Yes, if your authenticated identity mapping resolves the same human to
the same user_id across channels. Do not let a model or
client choose that partition.
What stays in PyAI and what goes to ContextDB?
Recordings, transcripts, and call analytics stay in PyAI. ContextDB holds the extracted action-relevant facts, their provenance, and the decision plus the host-reported execution receipt.
Start from the PyAI Omni starter.
The linked starter covers conversational recall and sourced writes. The
authenticated, host-owned evaluate_action and
report_execution boundary is shown on this page for you to
add.