lyzr-rag-client

Typed sync/async Python client for the Lyzr RAG /v1 API. Use it from PepGenX (Okta JWT only: the SDK mints an OAuth2 client-credentials token and sends Authorization: Bearer) and from Lyzr-hosted deployments (x-api-key). Built on httpx and pydantic.

Install

pip install lyzr-rag-client
# or
uv add lyzr-rag-client

Requires Python 3.10+. Runtime dependencies: httpx and pydantic only.

PepGenX quick start

JWT-only posture: the caller presents an Okta JWT and nothing else. RagClient.for_pepgenx mints the token (OAuth2 client credentials against the Okta authorization server), caches it until near expiry, re-mints once on a 401, and sends only Authorization: Bearer <jwt>. No API key, no team_id / project_id / user_id headers: identity is the token, entitlement is the server's job.

import os

from lyzr_rag_client import ModelSpec, RagClient, SourceInput, VectorStoreSpec

client = RagClient.for_pepgenx(
    os.environ["LYZR_RAG_BASE_URL"],  # APIM base + RAG route from onboarding
    client_id=os.environ["OIDC_CLIENT_ID"],
    client_secret=os.environ["OIDC_CLIENT_SECRET"],
    issuer=os.environ["OIDC_ISSUER"],  # token endpoint = {issuer}/v1/token
    scope="pepgenx2.0",
)

me = client.auth.whoami()
print(me.subject, me.org_id, me.roles)

kb = client.knowledge_bases.create(
    "demo",
    vector_store=VectorStoreSpec(provider="qdrant", credential_id="qdrant-prod"),
    # Or: VectorStoreSpec(provider="astra", credential_id="astra-prod")
    embedding=ModelSpec(model="text-embedding-3-small"),
)

job = client.documents.ingest(
    kb.id,
    [SourceInput(kind="text", text="PepsiCo Q3 margin notes…", title="notes")],
)
client.jobs.wait(job.id)

hits = client.query.retrieve(kb.id, "What changed in Q3?")
answer = client.query.answer(kb.id, "What changed in Q3?")
print(answer.answer, len(hits.results))

for_pepgenx takes exactly one of: client_id + client_secret with issuer (Okta) or token_url (any OIDC token endpoint), a token_provider you own, or a static token. Credentials the SDK mints are closed with the client (close() / context manager). The explicit form is equivalent and leaves the provider's lifecycle to you:

from lyzr_rag_client import BearerAuth, ClientCredentials, RagClient

tokens = ClientCredentials(
    issuer=os.environ["OIDC_ISSUER"],
    client_id=os.environ["OIDC_CLIENT_ID"],
    client_secret=os.environ["OIDC_CLIENT_SECRET"],
    scope="pepgenx2.0",
)
client = RagClient(os.environ["LYZR_RAG_BASE_URL"], BearerAuth(token_provider=tokens))

With OIDC_CLIENT_ID, OIDC_CLIENT_SECRET and OIDC_ISSUER (or OIDC_TOKEN_URL) exported, plain RagClient() resolves to the same JWT-only client.

The legacy composite header posture (x-pepgenx-apikey plus team_id / project_id / user_id) is reached only by passing auth=PepGenXAuth(...) explicitly; see docs/AUTH.md and docs/PEPGENX.md.

Lyzr-hosted quick start

import os

from lyzr_rag_client import RagClient

client = RagClient.for_lyzr(
    base_url=os.environ["LYZR_RAG_BASE_URL"],
    api_key=os.environ["LYZR_API_KEY"],
)

base_url is always explicit or taken from LYZR_RAG_BASE_URL. There is deliberately no built-in default hostname.

Environment variables

When auth=None, resolution order is OIDC_* (client id + secret with issuer or token URL) → PEPGENX_API_KEY (legacy composite, see docs/AUTH.md) → LYZR_API_KEY → LYZR_RAG_TOKEN. Whitespace-only values count as unset; only the chosen mode's class name is logged, never a credential. ClientCredentials.from_env() reads the same OIDC_* variables; pass prefix= to use a different naming scheme.

Async

from lyzr_rag_client import AsyncRagClient

async with AsyncRagClient.for_lyzr(base_url=..., api_key=...) as client:
    me = await client.auth.whoami()

AsyncRagClient mirrors RagClient; call await client.aclose() if you are not using the context manager.

Streaming answers

from lyzr_rag_client import CitationsEvent, DeltaEvent, DoneEvent, ErrorEvent

for event in client.query.stream(kb_id, "Summarize the margin notes"):
    match event:
        case CitationsEvent(citations=cites):
            ...
        case DeltaEvent(text=chunk):
            print(chunk, end="", flush=True)
        case DoneEvent(answer=final):
            print("\n", final)
        case ErrorEvent(code=code, detail=detail):
            raise RuntimeError(f"{code}: {detail}")

Jobs and uploads

• documents.ingest(...) and documents.upload(...) return 202 + a JobOut.
• An Idempotency-Key is generated automatically (pass idempotency_key= to reuse one).
• Poll with client.jobs.wait(job.id) (default 1 s interval, 600 s timeout).
• Multipart upload: client.documents.upload(kb.id, files=["./report.pdf"]).

Retries and timeouts

Defaults: 60 s total timeout, 5 s connect; 3 attempts with exponential backoff 1 s → 30 s (jitter); Retry-After is honoured when present.

Which requests retry (idempotent methods, bodies with an Idempotency-Key, transport connect failures, 429/503, and 502-504 on idempotent calls) is documented in docs/ERRORS.md.

Error handling

from lyzr_rag_client import NotFoundError

try:
    client.knowledge_bases.get(kb_id)
except NotFoundError as e:
    print(e.code, e.request_id, e.detail)

See docs/ERRORS.md for the full hierarchy and .code vocabulary.

Transaction ids

Pass transaction_id= on any resource call. The SDK sends it as the transaction_id header; the service echoes the correlation id in X-Request-Id (available as APIError.request_id).

Security

• Auth credentials are wrapped in Secret; minted/imported API-key material is redacted in model/RequestSpec repr/str. Neither appears in logs. A minted Okta token shows up only as BearerAuth(token_provider=<provider>) in repr.
• TLS verification is on by default (verify=True). The OAuth token endpoint must be https:// (loopback http:// only).
• Corporate CA: RagClient(..., verify="/path/to/ca.pem").
• HTTP proxies follow httpx trust_env (standard HTTP_PROXY / HTTPS_PROXY / NO_PROXY).

Compatibility

SDK 0.1.x targets Lyzr RAG API v1 as frozen in spec/SOURCE (OpenAPI pin: spec/openapi.json).

Development

License

MIT. See LICENSE.