Python SDK

The Python application SDK is the varianz package (install). Its test-side counterpart, varianz-pytest, is covered in Testing → pytest.

Runtime

from varianz import Varianz

vz = Varianz("http://localhost:50051")   # connect to a coordinator
vz = Varianz()                           # local-only mode (no coordinator)

The SDK activates only when enabled (VARIANZ_ENABLED=true); otherwise every API is a no-op and VPoints pass through.

Defining VPoints

@vz.vpoint(name="pricing/calculate")
def calculate_price(amount: float, discount: float) -> float:
    return amount * (1 - discount)

vz.register(calculate_price)

The name= keyword argument is required in practice — always set it explicitly.
Decorating is not enough: the function, class, or instance must be registered.

register() is idempotent: passing an already-registered @vz.vpoint function returns the same object, so calling it twice or discarding the result is harmless.The module-level decorator is different. from varianz import vpoint does not register, and register() on a plain function returns a new wrapper — so there you must rebind:

from varianz import vpoint

@vpoint(name="pricing/calculate")
def calculate_price(amount: float, discount: float) -> float: ...

calculate_price = vz.register(calculate_price)   # rebind: the decorator did not register

Registering an instance mutates it in place, so no rebinding is needed either way.

Registration methods

Method	Use when
`vz.register(item)`	Full control — one function, class, or instance per call
`@vz.component`	Per-class — register at definition time
`vz.discover()`	Bulk — scan the calling module and register every `@vpoint` function, `@function` context function, and class with `@vpoint` methods. This is the only method that accepts a free-standing module-level `@function`; `vz.register()` rejects one with `TypeError: … must be a @vpoint function or a class with @vpoint methods`. Call discovered VPoints through `vz.execute(name, *args)` — `discover()` does not rebind your module namespace

For classes, register the instance to preserve constructor state (vz.register(PricingService(config))); registering the class auto-instantiates via cls(). Don’t pass unbound methods.

Context functions bind to the instance registered first under a given VPoint name. Registering a second instance of the same class re-points the VPoint body to the new instance, while its @functions continue to resolve against the first. The behaviour is silent, so a stage calling ctx.* reads the earlier instance’s state without reporting anything.Create the Varianz runtime per test, alongside the service instance it registers, and the binding follows the instance you expect. This matters most in the common pytest pattern of building a fresh service for each test.

@vz.component replaces the class with a singleton proxy. It is equivalent to vz.register(cls): the decorated name refers to a proxy wrapping one auto-constructed instance, not to the class, so a later PricingService(config) raises a bare TypeError that does not mention Varianz.Use vz.register(PricingService(config)) on an instance you construct yourself whenever the constructor takes arguments or you need more than one instance.The same applies to vz.discover(): for classes it registers an auto-constructed instance. Instances you build yourself afterwards are not instrumented, and calls on them silently skip every stage — discover() does not rebind anything in your module namespace.

vz.register(...) blocks until the coordinator confirms the schema (up to 5 seconds), then proceeds regardless — Varianz fails open if the coordinator is unreachable.

Schema resolution: annotate your types

With type annotations (recommended): the schema is built at decoration time, the VPD registers eagerly, and stages work from the first request. Statically mapped types: int, float, bool, str, None, list[T], dict[K, V], Optional[T], and @dataclass types. Proto messages resolve eagerly from their DESCRIPTOR. Without annotations (lazy fallback): a VPoint defers its schema when any parameter it has to expose lacks a usable annotation. It then registers an unknown placeholder and fills the schema in from real values later. A missing return annotation is not a trigger — that registers eagerly, with an unknown return kind. opaque= is how you keep a signature eager: it excludes a slot that can never carry a schema, such as a gRPC servicer’s context, so the remaining parameters still derive. It is not a cause of deferral. Deferral costs you validation rather than delivery — stages still apply on the first call (details). Runtime-introspectable types include Pydantic models, attrs classes, and anything with __annotations__, __slots__, or vars(). Prefer annotations; where you can’t, wire varianz-scan into the build.

gRPC servicer methods

Python can instrument servicer methods directly — annotate the proto types and mark the gRPC context as opaque:

class PaymentServiceServicer(demo_pb2_grpc.PaymentServiceServicer):

    @vz.vpoint(name="payment/charge", opaque=["context"])
    def Charge(self, request: demo_pb2.ChargeRequest, context) -> demo_pb2.ChargeResponse:
        return demo_pb2.ChargeResponse(transaction_id=str(uuid.uuid4()))

vz.register(PaymentServiceServicer())

opaque=["context"] passes the ServicerContext through without exposing it to CEL. Proto field names, types, and nested message types come from the descriptor — no warmup needed. CEL overrides that construct proto structs (ChargeResponse { transaction_id: "test" }) are converted back to real proto messages automatically. Use the message’s simple name in CEL, not the package-qualified one. Set up the server interceptor once so sessions reach these handlers.

Local stages

For unit and in-process integration tests, attach a Python callback straight to a VPoint — no coordinator, no session, no CEL:

from varianz import Varianz

vz = Varianz()

@vz.vpoint(name="pricing/calculate")
def calculate(amount: float, discount_pct: float) -> float:
    return amount * (1 - discount_pct / 100)

handle = vz.attach_stage(
    "pricing/calculate", "PRE", "half-price",
    lambda ctx: ctx.args.amount * 0.5,
)

assert calculate(80.0, 10.0) == 40.0

handle.set_enabled(False)          # park it
assert calculate(80.0, 10.0) == 72.0
handle.remove()                    # permanent

attach_stage(target, anchor, name, stage_fn, enabled=True). target is a VPoint lookup name or the @vpoint-decorated function itself; anchor is a varianz.Anchor ("VALIDATE" | "PRE" | "MID" | "POST" | "AUDIT", checked eagerly, so a typo raises at attach time even while the SDK is disabled). attach_cel_stage(target, anchor, name, expression, enabled=True) takes the same shape with a CEL string instead of a callback, and the same methods are available on the decorated wrapper itself (calculate.attach_stage(anchor, name, fn)). The callback contract:

ctx.args exposes the VPoint’s named arguments — the caller’s live objects, with Python identity preserved. ctx.self is the registered instance for a class-based VPoint, None for a plain function.
Return varianz.CONTINUE to fall through to the next stage or the original function; return anything else, including None, to override the result.
The override is converted against the VPoint’s declared return type — a mistyped one raises TypeError rather than returning corrupted data. Struct returns are checked against the declared class, not just scalars.
An exception raised inside the callback reaches the caller unchanged, with its original type and traceback. That’s the way to inject failures.
async def callbacks are rejected: stages are synchronous, and an un-awaited coroutine would silently do nothing.

Handles and scoping

attach_stage and attach_cel_stage return a StageHandle with set_enabled(bool), is_enabled(), and remove(). Operating on a stage that’s already gone raises rather than silently doing nothing. Use it as a context manager to scope a stage to a block — the analogue of Node’s using:

with vz.attach_stage("pricing/calculate", "PRE", "half-price", half_price):
    assert calculate(80.0, 10.0) == 40.0
assert calculate(80.0, 10.0) == 72.0   # removed on exit

Pass enabled=False to attach a stage dormant. When the SDK is disabled, attaching returns an inert handle with the same methods, so test code doesn’t need to branch — but argument validation still runs, so a bad anchor fails identically either way.

Typed helpers

Running python -m varianz.buildscan during your build emits varianz_stages.py alongside the descriptor file, with one attach_<vpoint>_stage helper per VPoint. The helper types ctx.args field by field and types the override’s return, so a wrong field name or return type is a type error in your editor rather than a runtime one:

python -m varianz.buildscan src/myservice -o generated/varianz_descriptors.py
# also writes generated/varianz_stages.py (override with --stages-output)

from varianz_stages import attach_pricing_calculate_stage

attach_pricing_calculate_stage(vz, "PRE", "half-price", lambda ctx: ctx.args.amount * 0.5)

The generated module is native-free, so importing it is safe even when Varianz is disabled. The scanner refuses to overwrite a file that doesn’t carry its generated-file header, so pointing --stages-output at hand-written code fails the build instead of eating it.

Local stages apply to every call in the process, not just calls carrying a session ID. Use them for behavior owned by the process under test; use CEL stages when you need session isolation, cross-service delivery, or samples.

Session API

Propagation is interceptor-driven; these are the primitives interceptors use:

Varianz.set_session(session_id)    # bind session to the current context (returns a contextvars token)
Varianz.get_current_session()      # read it (for outbound header injection)
Varianz.reset_session(token)       # restore the previous value

The session lives in a contextvar, so it follows asyncio tasks automatically.

Context functions

from varianz import vpoint, function

class PricingService:
    @vpoint(name="pricing/calculate")
    def calculate(self, amount: float) -> float: ...

    @function(name="applyTax")           # ctx.applyTax(...) in CEL
    def apply_tax(self, price: float, rate: float) -> float:
        return price * (1.0 + rate)

Context functions require type annotations on every parameter and the return — they are never lazily resolved. Functions with no profile are available to all VPoints; class-scoped functions are always available to that class’s VPoints.

Platform notes

Wheels only (no sdist): CPython 3.10–3.15 and PyPy 3.11 on macOS and Linux (glibc and musl). In Docker, install inside the Linux image — or run tests from the host against the containerized service, since only the coordinator connection matters.

Getting started

Core concepts

Guides

SDK reference

Testing

Reference

Runtime

Defining VPoints

Registration methods

Schema resolution: annotate your types

gRPC servicer methods

Local stages

Handles and scoping

Typed helpers

Session API

Context functions

Platform notes

​Runtime

​Defining VPoints

​Registration methods

​Schema resolution: annotate your types

​gRPC servicer methods

​Local stages

​Handles and scoping

​Typed helpers

​Session API

​Context functions

​Platform notes

Runtime

Defining VPoints

Registration methods

Schema resolution: annotate your types

gRPC servicer methods

Local stages

Handles and scoping

Typed helpers

Session API

Context functions

Platform notes