Qefro Python Backend SDK

qefro-backend is the Python framework for building Qefro Business Tool handlers and Business Flows. Register your tools and flows, serve the signed Qefro webhook, and let the Qefro Runtime orchestrate everything — including multi-step flows and customer verification.

Wire-compatible with the JavaScript and Rust SDKs: same signed protocol, same capabilities.list schema, same 8 flow step types.

Install

pip install qefro-backend

Zero runtime dependencies — it uses only the Python standard library.

Quick start

from qefro_backend import Qefro

app = Qefro(signing_secret="dev-secret")  # or QEFRO_SIGNING_SECRET from env

@app.tool("get_orders", auth="required", lookup={"required": ["email"]})
async def get_orders(ctx):
    customer = ctx.customer.require()
    return [{"orderId": "ord_1", "customerId": customer["id"]}]

app.run(8088)  # POST http://0.0.0.0:8088/qefro

Set the same signing secret in Admin Console → Business Tools → SDK Connections, then Sync Tools. Handlers may be sync or async.

Customer authorization

Register a provider to resolve and verify customers. auth="required" tools authorize automatically before the handler runs; a challenge outcome pauses the tool and asks the customer for a code, then resumes.

from qefro_backend import Qefro, CustomerProvider

app = Qefro("dev-secret")

class Customers(CustomerProvider):
    async def lookup(self, ctx):
        return {"id": ctx.identity.get("phone", "demo")}

    async def authorize(self, ctx):
        if not ctx.response:
            return ctx.auth.sms_otp(ctx.customer["id"], "Enter the code we texted you.")
        if ctx.response.strip() != "123456":
            return ctx.auth.sms_otp(ctx.customer["id"], "Wrong code, try again.")
        return ctx.auth.success(ctx.customer, {"type": "bearer_token", "access_token": "demo"})

app.customer(Customers())

Customer Hub (optional)

When QEFRO_CUSTOMER_HUB_ENABLED=true, tools can call Hub via platform.customer on tool.invoke (or QEFRO_CUSTOMER_HUB_URL + service token). Hub is optional — defaults keep existing apps working (ENABLED=false, OPTIONAL=true). Soft-skip returns None / no-ops when Hub is off or unreachable; set QEFRO_CUSTOMER_HUB_OPTIONAL=false to hard-fail.

@app.tool("create_reservation", auth="none")
async def create_reservation(ctx):
    customer = await ctx.customer.resolve({"whatsapp_number": ctx.identity.get("phone")})
    # Hub properties: ctx.customer.id, .phone_number, .whatsapp_number, .display_name
    await ctx.timeline.append({
        "event_type": "reservation.created",
        "payload": {"code": "R-1001"},
    })
    await ctx.membership.attach({"solution_id": "restaurant-pro"})
    await ctx.consent.grant({"purpose": "marketing"})
    return {"customer_id": customer["id"] if customer else None}

ctx.storage talks to managed storage (ADR-002) the same way as the JS SDK — never call Mongo or storage-service URLs yourself.

Customer Hub Person and Business Events

ctx.person queues Customer Hub mutations (note, tag, assign, activity, merge) for the Qefro runtime. Declare events with app.business_event(...) and publish them with ctx.emit(...) — delivery is outbox → ingest, not CRM automation.

from qefro_backend import Qefro, on_person_created

app = Qefro("dev-secret")
app.business_event({"event_type": "quotation.created", "version": 1})

@app.tool("create_quotation", auth="none")
async def create_quotation(ctx):
    await ctx.person.tag("vip")
    ctx.emit({"event_type": "quotation.created", "event_id": "Q-1001", "data": {"amount": 125000}})
    return {"id": "Q-1001"}

app.flow({"id": "welcome", "trigger": on_person_created()}).message("hi", "Hi").complete("done")

Business Flows

Flows describe how your Business Tools are orchestrated. They are metadata only — the SDK advertises them through capabilities.list and the Qefro Runtime discovers, validates, versions, and executes them. Nothing runs inside the SDK.

(
    app.flow({
        "id": "order_lookup",              # immutable identity — renaming `name` never creates a new flow
        "name": "Order Lookup",
        "description": "Look up customer orders",
        "category": "crm",
        "tags": ["customer", "orders"],
        "intent": ["track order", "where is my order"],
        "inputs": ["email"],
        "outputs": ["get_orders"],
    })
    .ask("email", field="email", prompt="Please enter your email.")
    .tool("orders", tool_ref="get_orders")
    .complete("done", message="Here are your recent orders.")
)

Every step needs a unique id; tool steps reference an existing Business Tool by tool_ref. Step builders: .ask() .tool() .challenge() .upload() .condition() .delay() .approval() .complete() .message() .tag() .assign() .activity(). A duplicate/empty flow or step id raises FlowError. See examples/order-approval for a full ask → tool → condition → approval → OTP-authenticated tool → complete flow.

Protocol

Requests are HMAC-SHA256 signed (X-Qefro-Signature / X-Qefro-Timestamp, payload v1:<timestamp>:<body>). Responses include X-Qefro-Protocol, X-Qefro-SDK, and X-Qefro-Version.

Examples

export QEFRO_SIGNING_SECRET=dev-secret
python examples/basic/server.py            # ask -> tool -> complete
python examples/order-approval/server.py   # condition + approval + OTP-authenticated tool

Docs

• Register SDK Business Tools
• Define Business Flows
• Run Business Flows

Development

pip install -e ".[dev]"
pytest

License

MIT