Python SDK
Synchronous client, server-side OAuth and storage adapters.
Backend OAuth
create_authorization returns URL and pending. Save pending in the authenticated session and pass the full callback to exchange. Close the HTTPX client when finished. The Python client is synchronous: in an ASGI app run it outside the event loop or in a worker.
from pura_llm import create_authorization, PuraOAuthClient
options = dict(api_base="https://ai.puradigital.it/v1", client_id="YOUR_CLIENT_ID")
url, pending = create_authorization(
**options, redirect_uri="https://your-app.example/pura/callback", locale="it")
oauth = PuraOAuthClient(**options, client_secret=server_client_secret)
try:
tokens = oauth.exchange(callback_url, pending)
user_token_store.set(tokens)
finally:
oauth.close()Chat and usage
Create a client for the current grant with a persistent store. capabilities(), chat(), usage() and transcribe() use the same connection and subscription. Save rotated tokens with set; get returns None for a missing connection.
from pura_llm import PuraInferenceClient
pura = PuraInferenceClient(**options, store=user_token_store,
client_secret=server_client_secret)
try:
capabilities = pura.capabilities()
answer = pura.chat(messages=[
{"role": "user", "content": "Hello"}])
usage = pura.usage()
finally:
pura.close()Streaming
stream_chat is a context manager returning native SSE lines, including the [DONE] terminator. Interpret data: lines as events and close the context even when interrupted.
with pura.stream_chat(messages=messages) as lines:
for line in lines:
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
consume_json_event(data)Adapters and locks
TokenStore offers get() and set(tokens). Across processes implement with_refresh_lock() as a shared context manager for the grant, with database access in the same context. PuraError.status and .code expose stable errors; a revoked family requires fresh authorization.