OpenAI Agents SDK

export SENSO_API_KEY="YOUR_SENSO_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_KEY"
pip install openai-agents requests pydantic

import os, requests
from pydantic import BaseModel
from agents import Agent, Runner, function_tool

BASE = "https://apiv2.senso.ai/api/v1"
HEADERS = {
    "X-API-Key": os.environ["SENSO_API_KEY"],
    "Content-Type": "application/json",
}

# The approval grant, in Senso terms. Leave APPROVED_CONTENT_IDS empty to let the
# agent read everything the API key's KB scope allows.
APPROVED_CONTENT_IDS: list[str] = []

class Citation(BaseModel):
    content_id: str
    title: str

class Evidence(BaseModel):
    answer: str
    citations: list[Citation]

# The ledger lives in application memory, not in the model's context. Nothing the
# model emits can add a row here — that asymmetry is what makes step 5 mean anything.
LEDGER: list[Evidence] = []

@function_tool
def senso_lookup(question: str) -> Evidence:
    """Look up a verified answer in the company knowledge base.

    Use this for any factual claim about the company, its products, pricing or
    policies. Never answer such a question from memory.
    """
    payload: dict = {"query": question, "max_results": 5}

    # require_scoped_ids is the hard edge: without it, content_ids only biases the
    # search and unapproved chunks can still come back.
    if APPROVED_CONTENT_IDS:
        payload["content_ids"] = APPROVED_CONTENT_IDS
        payload["require_scoped_ids"] = True

    resp = requests.post(f"{BASE}/org/search", headers=HEADERS, json=payload)
    resp.raise_for_status()
    body = resp.json()

    # Chunks dedupe to sources: five chunks from one document is one citation.
    seen: dict[str, str] = {}
    for chunk in body.get("results", []):
        seen.setdefault(chunk["content_id"], chunk.get("title", "Untitled"))

    evidence = Evidence(
        answer=body.get("answer", ""),
        citations=[Citation(content_id=cid, title=t) for cid, t in seen.items()],
    )
    LEDGER.append(evidence)
    return evidence

class SupportReply(BaseModel):
    reply: str
    cited_content_ids: list[str]

agent = Agent(
    name="Support agent",
    instructions=(
        "You answer customer questions about the company. "
        "Call senso_lookup before making any factual claim. "
        "If the knowledge base does not answer the question, say so plainly "
        "rather than guessing. List the content_id of every source you relied "
        "on in cited_content_ids."
    ),
    tools=[senso_lookup],
    output_type=SupportReply,
)

def verify(reply: SupportReply, ledger: list[Evidence]) -> list[str]:
    """Return the cited IDs that no tool call ever produced."""
    real = {c.content_id for e in ledger for c in e.citations}
    return [cid for cid in reply.cited_content_ids if cid not in real]

result = Runner.run_sync(agent, "What is your refund window for annual plans?")
reply = result.final_output

fabricated = verify(reply, LEDGER)
if fabricated:
    raise ValueError(f"Answer cites sources that were never retrieved: {fabricated}")
if not reply.cited_content_ids:
    raise ValueError("Answer makes claims with no source behind it.")

print(reply.reply)
for c in {c.content_id: c for e in LEDGER for c in e.citations}.values():
    print(f"  source: {c.title} ({c.content_id})")