liberrex-copilot (Python)

Embed the Liberrex AI copilot in your product. Your backend declares tools (plain Python functions), mints short-lived session tokens for your signed-in users, and runs the tools when the copilot needs them — as that end user, under your own permission checks. Your data never leaves your system except what a tool returns.

• Python ≥ 3.9, one runtime dependency (requests).
• Implements Embedded Copilot protocol v1. Pair it with @liberrex/copilot-react in the browser.

pip install liberrex-copilot                 # extras: [pydantic] [flask] [fastapi] [django] [sqlalchemy] [cli]

Quick start

import os
from typing import Literal, Optional
from liberrex_copilot import Copilot, Context, ToolError

copilot = Copilot(secret_key=os.environ["LIBERREX_SECRET_KEY"])   # base_url: env LIBERREX_API_URL or https://ai-api.liberrex.com

copilot.vocabulary("Atlas ERP manages invoices, orders, stock… 'DSO' means days sales outstanding.")
copilot.entity("Invoice", "A customer invoice",
               fields={"number": "str", "total": "decimal", "status": "draft|sent|paid|overdue"},
               relations={"customer": "Customer"})

@copilot.tool(description="Find invoices by customer and status.", permission="invoices.read", output="table")
def search_invoices(ctx: Context, customer: Optional[str] = None,
                    status: Optional[Literal["draft", "sent", "paid", "overdue"]] = None, limit: int = 20) -> list:
    """
    Args:
        customer: Customer name or id.
        limit: Maximum rows.
    """
    user = my_app.load_user(ctx.user.id, tenant=ctx.tenant.id)     # act as the end user…
    if not user.can("invoices.read"):                                 # …with YOUR permission rules
        raise ToolError("forbidden", "You cannot see invoices.")
    return my_app.invoices.search(user, customer=customer, status=status, limit=limit)

@copilot.tool(kind="write", confirm="Create a purchase order: {quantity} × {product}", permission="orders.write")
def create_purchase_order(ctx: Context, product: str, quantity: int) -> dict:
    """Create a purchase order for a product."""
    return my_app.purchasing.create_po(ctx.user.id, product, quantity)

copilot.sync()                     # PUT /manifest — version = hash of the manifest (no-op when unchanged)
copilot.start_worker(threads=4)    # poll mode: background threads, reconnect with backoff, graceful stop

Mint a session token for the browser (never send the secret key to the browser):

@app.get("/api/copilot/token")
def copilot_token(request):
    u = request.user
    session = copilot.create_session(
        user={"id": u.id, "name": u.name, "email": u.email, "locale": "fr", "roles": u.roles,
              "attributes": {"region": u.region}},
        tenant={"id": u.org_id, "name": u.org_name},
        ttl_seconds=900)
    return {"token": session.token, "expires_at": session.expires_at}

Declaring tools

@copilot.tool(...) (or bare @copilot.tool) options:

Input schema is generated from type hints: str, int, float, bool, Decimal, date, datetime, time, UUID, Literal[...], Optional[...] / X | None, Union, list[...] / tuple / set, dict[str, X], Enum, TypedDict, dataclasses, pydantic models (v1/v2) and Annotated[X, "description"]. Parameter descriptions come from Google, NumPy or reST docstrings. Parameters with a default are optional. The first parameter annotated Context (or named ctx / context) receives the context and is not part of the schema.

Arguments are validated and coerced before your function runs ("5" → 5, ISO strings → date, enum values → Enum members, dicts → dataclasses / pydantic models). Bad arguments return invalid_arguments to the model.

What-if tools (simulates)

A read tool can declare that it simulates one of your write tools: it accepts the same (or compatible) arguments, returns the predicted impact and changes nothing. The configuration advisor surface (§3.8 of the protocol) uses the pair to propose a setting change, show its simulated impact (numbers from your simulation, never from the model) and apply it through the write tool — after the user confirms.

@copilot.tool(kind="write", permission="stock.manage", confirm="Set the reorder point of {product_id} to {reorder_point}")
def update_reorder_point(ctx: Context, product_id: str, reorder_point: int) -> dict:
    """Change a product's reorder point."""
    return my_app.stock.set_reorder_point(ctx.user.id, product_id, reorder_point)

@copilot.tool(simulates="update_reorder_point", permission="stock.read", output="kpis")
def simulate_reorder_point(ctx: Context, product_id: str, reorder_point: int) -> dict:
    """What-if: stock-outs, extra stock value and order frequency with a new reorder point (changes nothing)."""
    return my_app.stock.simulate_reorder_point(ctx.user.id, product_id, reorder_point)

Checks: simulates is only allowed on a read tool (ManifestError at registration otherwise); validate() / sync() fail when the target is unknown or is not a write tool, and warn when the simulation does not accept one of the write tool's required arguments.

Context

ctx.user (id, name, email, locale, roles, attributes), ctx.tenant (id, name), ctx.session_id, ctx.call_id, ctx.locale, ctx.deadline, ctx.has_role("a", "b"). These are exactly the values your backend passed to create_session. Context.for_testing(user_id=…, roles=[…]) helps unit-test tools.

Results

Return anything JSON-serialisable: Decimal, date/datetime, Enum, UUID, dataclasses, pydantic models, namedtuples and SQLAlchemy rows are converted. Results over 256 KB are truncated (the longest list is cut and a note is added) so the call never fails for size.

• raise ToolError("not_found", "No invoice F-9") → {ok: false, error: {code, message}} (message ≤ 500 chars, shown to the model — keep it user-safe).
• Any other exception → {ok: false, error: {code: "internal_error"}}; the traceback is logged locally, never sent.
• Display hints for the UI (otherwise output decides):

return copilot.display.table(rows, columns=[("number", "Invoice"), ("total", "Total", "currency")], title="Overdue")
return copilot.display.kpis([("Overdue", 3), {"label": "Amount", "value": 1250.5, "unit": "EUR"}], data=summary)
return copilot.display.text("**Done.**")

Column formats: text, number, currency, percent, date, datetime.

Run a tool locally exactly as Liberrex would: copilot.run_tool("search_invoices", {"customer": "Acme"}, user={"id": "u1", "roles": ["invoices.read"]}, tenant={"id": "t1"}).

The bridge: poll (default) or webhook

Poll mode works behind NAT/firewalls (no inbound endpoint): worker = copilot.start_worker(threads=4) starts a long-polling thread plus a pool of tool threads. It reconnects with exponential backoff (≤ 30 s), skips calls whose deadline passed, retries posting results, and stops gracefully with copilot.stop_worker() (running calls finish and their results are posted). copilot.start_worker(block=True) runs until Ctrl-C / SIGTERM — handy for a dedicated worker process. worker.stats and worker.last_error expose counters. For cron-style use: copilot.poll_once().

Run one worker per process you want polling; any number of processes can poll (each call is claimed exactly once).

Webhook mode (choose it in the Liberrex console, set the webhook secret lxwh_…):

copilot = Copilot(secret_key=..., webhook_secret=os.environ["LIBERREX_WEBHOOK_SECRET"])

# Flask
from liberrex_copilot.flask import blueprint
app.register_blueprint(blueprint(copilot), url_prefix="/liberrex")          # POST /liberrex/webhook
# FastAPI
from liberrex_copilot.fastapi import router
app.include_router(router(copilot), prefix="/liberrex")
# Django
from liberrex_copilot.django import view
urlpatterns = [path("liberrex/webhook", view(copilot))]
# Anything else
status, body = copilot.handle_webhook(raw_body_bytes, headers)

Requests are verified (X-Liberrex-Signature: v1=<HMAC-SHA256> over timestamp + "." + raw body, constant-time compare) and rejected when the timestamp is more than 300 s away.

Other API

Logging uses logging.getLogger("liberrex_copilot"). Secrets are never logged (reprs mask the key) and tool arguments / results are never logged at INFO.

CLI

export LIBERREX_SECRET_KEY=lxsk_test_…  LIBERREX_API_URL=http://localhost:5001

liberrex-copilot draft --sqlalchemy myapp.models:Base -o copilot.yaml      # entities + list/get read tools
liberrex-copilot draft --django --settings mysite.settings -o copilot.yaml
liberrex-copilot draft --openapi openapi.json -o copilot.yaml              # GET → read tools, others → write tools
liberrex-copilot validate copilot.yaml
liberrex-copilot push copilot.yaml           # disabled tools are pushed with enabled:false (--only-enabled to omit)
liberrex-copilot push --app myapp.copilot:copilot      # push the manifest of a Copilot object
liberrex-copilot sessions create --user-id u1 --roles invoices.read,orders.write --tenant-id acme --json
liberrex-copilot manifest                    # show what Liberrex stores
liberrex-copilot ping

draft never reads rows — only model shapes. Sensitive-looking columns (password, token, secret, hash…) are skipped, and every proposed tool is enabled: false until a human reviews it. YAML needs PyYAML ([cli] extra); without it the CLI reads and writes JSON.

Development

cd sdk/python
python3 -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest            # fake Liberrex server on 127.0.0.1 — no network

See examples/embedded-erp/backend for a complete FastAPI app (Atlas ERP) using this SDK.