Conseqa for Python

Your agent calls a provider and gets a reply, or does not. Conseqa records what the call was meant to do before it is made, then reads the system of record afterwards to establish what is actually true: verified, failed, or honestly unknown.

This package is the application side for Python programs. It has no dependencies.

pip install conseqa

It needs Python 3.9 or newer, and a Conseqa runner, version 0.1.8 or newer, to talk to. The runner and the tool that builds contracts are Node.js programs; the guide at https://www.conseqa.dev/docs sets them up, and npx @conseqa/cli quickstart gives you both running in two minutes.

Protect a call

This protects the refund that the quickstart sets up. create_refund stands for the call your program already makes.

import os

from conseqa import Conseqa, ambiguous, load_contract_package, rejected


def what_an_error_proves(error):
    # A refusal with a 4xx status proves the refund was not made.
    # Anything else, a dropped connection above all, proves nothing either way.
    status = getattr(error, "status", None)
    if isinstance(status, int) and 400 <= status < 500:
        return rejected(f"http_{status}")
    return ambiguous("no_reply")


package = load_contract_package("dist/conseqa")
refund_processed = package.contract(
    "quickstart.refund.processed",
    1,
    # What identifies the thing that was made, taken from the service's reply.
    bind_result=lambda reply: {"refundId": reply["id"]} if reply.get("id") else None,
    classify_exception=what_an_error_proves,
)
conseqa = Conseqa(
    package,
    runner_url="http://127.0.0.1:4319",
    sidecar_token=os.environ.get("CONSEQA_SIDECAR_TOKEN"),
)

intent = {"paymentId": "pay_1001", "amountMinor": 2500, "currency": "USD"}
reply = conseqa.protect(
    refund_processed,
    action_key="ticket-8042-refund",
    intent=intent,
    # The call you already make, given the reference to send with it.
    execute=lambda call: create_refund({**intent, "reference": call.provider_correlation_key}),
)

protect returns what your call returned and raises what it raised. In between it tells the runner what was intended, that the call is starting, and how it ended. The runner then reads your system with its own read-only credential and decides the verdict, which appears in the dashboard.

• action_key is one stable key for the business operation, such as the ticket or order it belongs to. A key that was already used raises DuplicateActionError and the call is not made.
• With the default mode="required", the call is not made unless the intent was recorded first; a runner that cannot be reached raises ProtectionUnavailableError. With mode="best_effort" the call goes ahead anyway and shows as an unprotected gap.
• intent is what the call is meant to do, in the shape of the contract's intent schema.

The contract

A contract package is built once with conseqa contracts build and installed in the runner. Its runner half, which finds and reads the record, runs inside the runner and stays as it is. Its application half is three small decisions, and a Python program makes them in Python, as arguments to package.contract:

load_contract_package hashes the package's files again and compares the result with its manifest, so a program never registers an intent against a build other than the one on disk.

Connectors

For a provider Conseqa has a connector for, the application half is already written. conseqa.connectors gives it in Python, and helps build the request so that the key goes where the runner will look for it. Build the contract package from the JavaScript connector as usual; its runner half reads the provider.

import requests
from conseqa.connectors import cashfree_refund, decorate_cashfree_refund

refund = cashfree_refund(package)
intent = {"orderId": "order_8042", "amountMinor": 249950, "currency": "INR"}


def create_refund(call):
    request = decorate_cashfree_refund(
        intent, call.provider_correlation_key, {"refund_note": "Damaged on arrival"}
    )
    response = requests.post(
        "https://api.cashfree.com/pg" + request["path"],
        json=request["body"],
        headers={**request["headers"], **cashfree_credentials},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()


reply = conseqa.protect(
    refund, action_key="ticket-8042-refund", intent=intent, execute=create_refund
)

Each connector's exception rules read the shapes Python clients raise: http_status on stripe-python's errors, response.status_code on those of requests and httpx. Raise conseqa.connectors.NotDispatchedError only when your code can prove a request never left the program; anything else is treated as possibly done, and the runner looks.

asyncio

reply = await conseqa.protect_async(
    refund_processed,
    action_key="ticket-8042-refund",
    intent=intent,
    execute=create_refund_async,
)

execute may be a coroutine function. The requests to the runner are made off the event loop. A call that is cancelled part way is reported as ambiguous, because it is.

Traces and coverage

If your program uses OpenTelemetry, each protected call gets a span in the trace that is active, marked so that the dashboard can show the action inside the agent run that made it and count coverage. Nothing needs configuring: installing opentelemetry-api is enough, and pip install "conseqa[tracing]" does that. To count a run that was not protected, wrap it:

conseqa.observe("refund.create", lambda: create_refund(refund))

What it raises

Everything your own call raises is raised to you unchanged.

What this version does not do

There is no spool: when the runner cannot be reached, a required action stops and a best-effort one runs unprotected. The JavaScript SDK can hold intents in an encrypted folder shared with the runner; this one cannot yet.

Apache-2.0 licensed. Questions: https://www.conseqa.dev/docs.