Instrument a service

This guide covers instrumenting real services: classes with many methods, gRPC handlers, and existing code you don’t want to restructure. For a single free function, the Quickstart shows the minimal form. What to instrument: you don’t need VPoints everywhere. Pick the functions tests need to control or observe — payment charges, calls to external providers, decision points (fraud checks, feature gates), and notification sends are the usual candidates.

Register a service class

When a service has several related operations, register the whole class or instance — each instrumented method becomes a VPoint.

from varianz import Varianz

vz = Varianz("http://localhost:50051")

class PricingService:
    @vz.vpoint(name="pricing/validate")
    def validate(self, amount: float) -> bool:
        return amount > 0.0

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

# Register the *instance* (preserves constructor state):
service = PricingService()
vz.register(service)

import { vpoint } from '@varianz/node';

class PricingService {
  @vpoint({ name: 'pricing/validate' })
  validate(amount: number): boolean {
    return amount > 0;
  }

  @vpoint({ name: 'pricing/calculate' })
  calculate(amount: number, discount: number): number {
    return amount * (1 - discount);
  }
}

// TC39 decorators register on construction — instantiate before use:
const service = new PricingService();

type PricingService struct {
    varianz.Embed `varianz:"pricing"`
    DB            *sql.DB   // ordinary dependency-injected fields work
}

func (s *PricingService) CalculatePrice(ctx context.Context, req PricingRequest) (PricingResponse, error) {
    return PricingResponse{Total: req.BasePrice * float64(req.Quantity)}, nil
}

func (s *PricingService) ApplyDiscount(ctx context.Context, req DiscountRequest) (DiscountResponse, error) {
    // ...
}

// Every exported method becomes a VPoint: pricing/calculate-price, pricing/apply-discount
proxy := vz.Register(&PricingService{DB: db})

public class PricingService {
    @VPoint(name = "pricing/validate")
    public boolean validate(double amount) {
        return amount > 0;
    }

    @VPoint(name = "pricing/calculate")
    public double calculate(double amount, double discount) {
        return amount * (1 - discount);
    }
}

// At startup: register VPDs with the registry.
RegistryBinding.init(registryAddr);
PricingServiceVPoints.bind(new PricingService());

class PricingService {
    @VPoint(name = "pricing/validate")
    fun validate(amount: Double): Boolean = amount > 0

    @VPoint(name = "pricing/calculate")
    fun calculate(amount: Double, discount: Double): Double =
        amount * (1 - discount)
}

RegistryBinding.init(registryAddr)
PricingServiceVPoints.bind(PricingService())

Language-specific notes:

Python — vz.register() takes one function, class, or instance per call; @vz.component registers a class at definition time; vz.discover() scans the calling module and registers everything. Registration blocks until the registry confirms the schema (up to 5s, fail-open).
Go — method names convert PascalCase → kebab-case, prefixed by the varianz:"..." struct tag (CalculatePrice → pricing/calculate-price). Call through the returned proxy (proxy.Call(ctx, "CalculatePrice", req)) or generate typed proxies with varianz-gen.
Java/Kotlin — the annotation processor generates a <ClassName>VPoints aggregate at compile time; the varianz-agent intercepts calls to the annotated methods at class-load time, so ordinary calls (service.calculate(...)) route through the pipeline with no code changes. RegistryBinding.init() is idempotent and must run before your server starts serving.
TypeScript — for plain objects instead of classes, use createVPoints(namespace, api); call through the returned instrumented proxy (or pass { patch: 'mutate' } to instrument the object in place). Calls made on the original object with the default wrap mode are not intercepted.

Migrate an existing function

Rename the original, wrap it, keep every caller working:

// Before
func CalculatePrice(ctx context.Context, req PricingRequest) (PricingResponse, error) { ... }

// After
func calculatePriceDirect(ctx context.Context, req PricingRequest) (PricingResponse, error) { ... }

var CalculatePrice = varianz.VPoint(vz, "pricing/calculate-price", calculatePriceDirect)

The same shape works in every language — in Python and TypeScript the decorator does the wrapping, and in Java/Kotlin the agent rewrites the method in place, so no rename is needed at all.

gRPC handlers

Don’t instrument gRPC handler methods that take StreamObserver or servicer plumbing directly — the schema would be polluted with transport types. Extract the business logic into a plain-typed method and instrument that:

public class PaymentServiceImpl extends PaymentServiceGrpc.PaymentServiceImplBase {

    @VPoint(name = "payment/charge")
    ChargeResult chargeCard(ChargeInput input) {
        return new ChargeResult(UUID.randomUUID().toString());
    }

    @Override
    public void charge(ChargeRequest request, StreamObserver<ChargeResponse> obs) {
        ChargeResult result = chargeCard(ChargeInput.from(request));
        obs.onNext(result.toProto());
        obs.onCompleted();
    }
}

type PaymentServer struct {
    pb.UnimplementedPaymentServiceServer
    varianz.Embed `varianz:"payment"`
}

// Plain-typed VPoint method — business logic only.
func (s *PaymentServer) ChargeVZ(ctx context.Context, in ChargeInput) (ChargeOutput, error) {
    return ChargeOutput{TxnID: uuid.NewString()}, nil
}

// gRPC handler converts protos and delegates.
func (s *PaymentServer) Charge(ctx context.Context, req *pb.ChargeRequest) (*pb.ChargeResponse, error) {
    out, err := proxy.Call(ctx, "ChargeVZ", chargeInputFrom(req))
    if err != nil {
        return nil, err
    }
    return out.(ChargeOutput).ToProto(), nil
}

# Python can instrument servicer methods directly — proto types resolve
# eagerly from their DESCRIPTOR. 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())

In CEL expressions, use the proto message’s simple name (ChargeResponse, not hipstershop.ChargeResponse). In Go, CEL struct literals handle nested proto messages directly — Money { currency_code: "USD", units: 42 } inside a parent literal works without flattening. Remember to add the session interceptor to your server — without it, test sessions never reach these VPoints.

Field names in schemas

Schemas use each language’s natural field naming, so CEL expressions match what your API already looks like:

Go — precedence: varianz:"name" tag → protobuf tag → json tag → snake_case of the Go name. Codegen types (oapi-codegen, sqlc, protoc) work automatically via their json tags. varianz:"-" excludes a field; varianz:"note,opaque" passes it through without exposing it to CEL.
Python — annotation-derived; dataclasses, Pydantic models, attrs classes, and proto messages are introspected. opaque=[...] marks parameters that pass through uninspected.
Java/Kotlin — fields discovered from getters or public fields, converted camelCase → snake_case for CEL (basePrice → base_price). Kotlin val maps to read-only, var to read-write.
TypeScript — the scanner derives fields from your TS types; @field({ kind: ... }) overrides individual kinds.

Context functions

Register helper functions your CEL expressions can call as ctx.<name>(...) — useful for domain logic tests shouldn’t have to re-implement:

vz.Function("applyTax", func(price, rate float64) float64 {
    return price * (1.0 + rate)
}, varianz.WithFunctionParams("price", "rate"))

from varianz import vpoint, function

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

    # Available as ctx.applyTax(...) to VPoints in this class.
    # Type annotations are required on every parameter and the return.
    @function(name="applyTax")
    def apply_tax(self, price: float, rate: float) -> float:
        return price * (1.0 + rate)

Then in a stage: ctx.applyTax(invoke(), 0.1). Functions can be global (available to every VPoint) or scoped to a profile.

Verify it worked

Start the service with VARIANZ_ENABLED=true and a registry endpoint, and watch the startup log for the single [varianz] ENABLED (...) line. Then confirm registration from the test side: insert a stage with await_sync_or_fail and check it reports one confirmed subscriber, or query the catalog with the list_vpoints MCP tool. A sync result of “0 of 0 subscribers” means the service isn’t connected — see How tests work.

Getting started

Core concepts

Guides

SDK reference

Testing

Reference

Instrument a service

Register a service class

Migrate an existing function

gRPC handlers

Field names in schemas

Context functions

Verify it worked

​Register a service class

​Migrate an existing function

​gRPC handlers

​Field names in schemas

​Context functions

​Verify it worked

Register a service class

Migrate an existing function

gRPC handlers

Field names in schemas

Context functions

Verify it worked