Calling components¶

Components call each other through the component client, never by calling methods directly. The target of a call — an entity instance, a workflow instance, an agent session — may be on another node of the service's cluster, and the client routes the call there. What the client hides is the routing. What it does not hide is that a call can fail: a refusal arrives as a CommandError with an error code, never as a default value.

Endpoints, workflow steps, consumers, timed actions and agents all have a client. An entity's command handler does not call other components: it decides from its own state and returns an effect. See Designing a service for where cross-component logic belongs.

Addressing a component¶

The client is addressed first by kind and instance, then by handler:

ScalaJavaKotlinPythonTypeScriptRustGo

val cart = componentClient.forEventSourcedEntity(EntityId("cart-1"))

cart.call(ShoppingCartEntity.addItem).invoke(LineItem("p1", "Widget", 2))   // Done
cart.call(ShoppingCartEntity.getCart).invoke()                              // ShoppingCart

public static CartState addAndRead(ComponentClient client) {
  var cart = client.forEventSourcedEntity("cart-1");

  cart.call(ShoppingCart::addItem).invoke(new LineItem("p1", "Widget", 2)); // Done
  return cart.call(ShoppingCart::getCart).invoke(); // CartState
}

val cart = componentClient.forEventSourcedEntity("cart-1")
cart.call(ShoppingCart::addItem).invokeSuspending(LineItem("p1", "Widget", 2))
val held: Cart = cart.call(ShoppingCart::getCart).invokeSuspending()

cart = client.for_event_sourced_entity("shopping-cart", "c1")

await cart.call("add-item").invoke(LineItem("p1", "Pen", 2))            # Done
state = await cart.call("get-cart").invoke(reply=ShoppingCart)          # decoded as ShoppingCart

const cart = client.of(ShoppingCartEntity, "cart-1")

await cart.call(ShoppingCartEntity.handlers.addItem).invoke({ productId: "p1", name: "Widget", quantity: 2 })
await cart.call(ShoppingCartEntity.handlers.getCart).invoke()          // ShoppingCart

/// `/carts/{cartId}`, `/total`, `/items`, `/items/{productId}` and `/checkout`, and `DELETE
/// /carts/{cartId}`, open to anyone.
pub struct CartApi;

impl CartApi {
    fn get_cart(request: &Request) -> Result<Cart, HttpProblem> {
        let cart_id = request.path("cartId");
        Ok(request
            .client()
            .invoke(ShoppingCart, cart_id, "get-cart", ())?)
    }

    fn total(request: &Request) -> Result<i32, HttpProblem> {
        let cart_id = request.path("cartId");
        Ok(request
            .client()
            .invoke(ShoppingCart, cart_id, "total-quantity", ())?)
    }

    fn add_item(request: &Request, item: LineItem) -> Result<Done, HttpProblem> {
        let cart_id = request.path("cartId");
        Ok(request
            .client()
            .invoke(ShoppingCart, cart_id, "add-item", item)?)
    }

    fn remove_item(request: &Request) -> Result<Done, HttpProblem> {
        let (cart_id, product_id) = (request.path("cartId"), request.path("productId"));
        let removed = request.client().invoke(
            ShoppingCart,
            cart_id,
            "remove-item",
            product_id.to_string(),
        )?;
        Ok(removed)
    }

    fn checkout(request: &Request, (): ()) -> Result<Cart, HttpProblem> {
        let cart_id = request.path("cartId");
        Ok(request
            .client()
            .invoke(ShoppingCart, cart_id, "checkout", ())?)
    }

    fn discard(request: &Request) -> Result<Done, HttpProblem> {
        let cart_id = request.path("cartId");
        Ok(request
            .client()
            .invoke(ShoppingCart, cart_id, "discard", ())?)
    }
}

// A component is addressed by its id and instance, then by its handler's wire name; the reply is
// decoded as the type the caller names.
ankka.Post("/carts/{cartId}", func(ctx context.Context, req *ankka.Request, item cart.LineItem) (cart.Cart, error) {
    cartID := req.Param("cartId")
    if _, err := ankka.Invoke[ankka.Done](ctx, "shopping-cart", cartID, "add-item", item); err != nil {
        return cart.Cart{}, err
    }
    return ankka.Invoke[cart.Cart](ctx, "shopping-cart", cartID, "get-cart", ankka.Unit{})
}),

In Scala and TypeScript the handler is the typed value the component declared, so the argument and reply types are checked by the compiler and the component's id travels with the handler. Python names the component by its id and the handler by its wire name, and every call is awaited.

Addresses	Scala	Python	TypeScript
an event sourced entity instance	`forEventSourcedEntity(entityId)`	`for_event_sourced_entity(component_id, entity_id)`	`of(Entity, entityId)`
a key value entity instance	`forKeyValueEntity(entityId)`	`for_key_value_entity(component_id, entity_id)`	`of(Entity, entityId)`
a workflow instance	`forWorkflow(workflowId)`; `lifecycle(companion)` for the engine's state; `awaitEnd(companion, timeout)` and `call(command).thenAwaitEnd(timeout)` to wait for its end	`for_workflow(component_id, workflow_id)`; `await_end(timeout, reply=…)` and `call(name).then_await_end(timeout)`	`of(Workflow, workflowId)`; `awaitEnd(timeoutMillis)` and `call(handler).thenAwaitEnd(timeoutMillis)`
an agent session	`forAgent(sessionId)`, with `import com.thinkmorestupidless.ankka.agent.*`	`for_agent(component_id, session_id)`	`of(Agent, sessionId)`
a view's rows	the view client; see Views	`views.get(view_id, key, Row)`, `views.all(view_id, Row)`	`views.get(viewId, key, Row)`, `views.all(viewId, Row)`
timers	the timer scheduler; see Timers	`timers.schedule(...)`, `timers.cancel(timer_id)`	`timers.schedule(...)`, `timers.cancel(timerId)`

A streaming handler is reached with stream(handler)(input) in Scala and .call(name).stream(input) in Python and TypeScript.

The instance is identified by the id. An instance that has never been written to exists already, holding its empty state, so reading one is not an error.

Where the client comes from depends on what is calling:

Caller	Scala	Python and TypeScript
An HTTP endpoint	`clients.componentClient`, from the factory given to `HttpServer.of`	the endpoint's own `client`
A workflow	`context.componentClient`, from the `WorkflowContext`	`self.context.client` / `this.client`
A consumer	its context's `componentClient`	`self.client` / `this.client`
A timed action	`context.componentClient`, from the `TimedActionContext`	`self.client` / `this.client`
An agent	`componentClient`, inherited from `Agent`	`self.client` / `this.client`
A test	`testKit.componentClient`, from `AnkkaTestKit`	the test kit's own client

In Python and TypeScript the client talks to the sidecar, which routes the call through the cluster exactly as a Scala call is routed.

Blocking is free¶

invoke blocks until the reply arrives. That is the recommended style, because every place a call can be made from — endpoints, workflow steps, consumers, timed actions, agent loops — runs on a virtual thread, and a blocked virtual thread releases the operating-system thread it was using. Sequential code stays readable and costs nothing to wait.

invokeAsync returns a Future instead, for fanning out several calls at once. ComponentClient.await(future, timeout) collects one, parking the virtual thread in the same way:

ScalaJavaKotlinPythonTypeScriptGo

val pending = cartIds.map(id =>
  componentClient.forEventSourcedEntity(EntityId(id)).call(ShoppingCartEntity.totalQuantity).invokeAsync()
)
val totals = pending.map(ComponentClient.await(_, 10.seconds))

public static List<Integer> totals(ComponentClient client, List<String> cartIds) {
  var pending =
      cartIds.stream()
          .map(id -> client.forEventSourcedEntity(id).call(ShoppingCart::getCart).invokeAsync())
          .toList();
  return pending.stream()
      .map(answer -> answer.toCompletableFuture().orTimeout(10, TimeUnit.SECONDS).join())
      .map(cart -> cart.items().stream().mapToInt(LineItem::quantity).sum())
      .toList();
}

val totals: List<Int> = coroutineScope {
    cartIds
        .map { id -> async { componentClient.forEventSourcedEntity(id).call(ShoppingCart::totalQuantity).invokeSuspending() } }
        .awaitAll()
}

async def totals_of(client: ComponentClient, cart_ids: list[str]) -> list[int]:
    pending = [client.for_event_sourced_entity("shopping-cart", cart_id).call("total-quantity").invoke(reply=int) for cart_id in cart_ids]
    return list(await asyncio.gather(*pending))

/** The totals of several carts, asked for at once: the whole takes as long as the slowest. */
totals: post("/totals", s.list(s.string), s.list(s.int), (ep: ShoppingCartEndpoint, _req, cartIds) =>
  Promise.all(cartIds.map((id) => ep.cart(id).call(ShoppingCartEntity.handlers.totalQuantity).invoke())),
),

// One goroutine per call; the whole takes as long as the slowest. Each goroutine's ctx is the
// handler's, so every call joins the request's trace.
ankka.Post("/totals", func(ctx context.Context, _ *ankka.Request, cartIDs []string) (CartTotals, error) {
    var wg sync.WaitGroup
    var mu sync.Mutex
    totals, errs := CartTotals{}, []error{}
    for _, id := range cartIDs {
        wg.Add(1)
        go func() {
            defer wg.Done()
            total, err := ankka.Invoke[int](ctx, "shopping-cart", id, "total-quantity", ankka.Unit{})
            mu.Lock()
            defer mu.Unlock()
            totals[id], errs = total, append(errs, err)
        }()
    }
    wg.Wait()
    return totals, errors.Join(errs...)
}),

The calls run concurrently, and the whole takes as long as the slowest. The multi-agent orchestration guide uses the same pattern to consult several agents at once.

Refusals and failures¶

A component that refuses a command returns an error effect with a code. The caller receives it as a CommandError carrying that code and message, whether the component ran on this node or another:

ScalaJavaKotlinPythonTypeScriptRustGo

try cart.call(ShoppingCartEntity.addItem).invoke(LineItem("p1", "Widget", 0))
catch case e: CommandError if e.code == ErrorCode.BadRequest => ...

public static int addNothing(ComponentClient client) {
  try {
    client
        .forEventSourcedEntity("cart-1")
        .call(ShoppingCart::addItem)
        .invoke(new LineItem("p1", "Widget", 0));
    return 200;
  } catch (CommandException e) {
    // 400: the entity refused the command, and nothing was persisted.
    return e.problem().status();
  }
}

try {
    cart.call(ShoppingCart::addItem).invokeSuspending(LineItem("p1", "Widget", 0))
    fail("a quantity of zero was accepted")
} catch (e: CommandException) {
    assertEquals(400, e.problem().status())
}

async def add_widget(cart: Calls) -> Done | str:
    try:
        return await cart.call("add-item").invoke(LineItem("p1", "Widget", 0), reply=Done)
    except CommandError as refused:
        if refused.error.code == ErrorCode.BAD_REQUEST:
            return refused.error.message
        raise

/** Adds an item, answering a refused quantity as text rather than as the 400 it would otherwise be. */
tryAdd: post("/{cartId}/items/try", LineItem, s.string, async (ep: ShoppingCartEndpoint, req, item) => {
  try {
    await ep.cart(req.params.cartId).call(ShoppingCartEntity.handlers.addItem).invoke(item)
    return "added"
  } catch (e) {
    if (e instanceof CommandError && e.code === ErrorCode.BadRequest) return `refused: ${e.message}`
    throw e
  }
}),

/// A refused command is an `Err` carrying the entity's code and message, whichever node the
/// entity ran on; nothing was persisted. Here one code is handled and the rest passed on.
fn add_item_or_explain(request: &Request, item: LineItem) -> Result<String, HttpProblem> {
    let cart_id = request.path("cartId");
    let added: Result<Done, CommandError> =
        request
            .client()
            .invoke(ShoppingCart, cart_id, "add-item", item);
    match added {
        Ok(Done) => Ok("added".to_string()),
        Err(e) if e.code == ErrorCode::BadRequest => Ok(format!("not added: {}", e.message)),
        Err(e) => Err(e.into()),
    }
}

// A refused command is a *RefusalError carrying the entity's code; nothing was persisted. Left
// to propagate, the sidecar answers the status for its code; handled, it is read like this.
ankka.Post("/carts/{cartId}/try", func(ctx context.Context, req *ankka.Request, item cart.LineItem) (string, error) {
    _, err := ankka.Invoke[ankka.Done](ctx, "shopping-cart", req.Param("cartId"), "add-item", item)
    var refusal *ankka.RefusalError
    if errors.As(err, &refusal) && refusal.Code == ankka.BadRequest {
        return "refused: " + refusal.Message, nil
    }
    if err != nil {
        return "", err
    }
    return "added", nil
}),

Nothing is persisted by a refused command. In an endpoint, let a CommandError propagate: the server answers with the status for its code, so a Conflict in an entity is a 409 at the edge with no mapping in between. See Error codes.

Timeouts¶

A blocking invoke waits for ankka.ask-timeout, 10 seconds by default, and then fails with a CommandError whose code is Timeout. Change it in the service's application.conf:

ankka.ask-timeout = 30s

A timeout says the reply did not arrive in time, not that the command did not happen. A command that timed out may still have been applied. When a caller retries after a timeout, the command must be safe to receive twice.

A wait for a workflow's end does not use ankka.ask-timeout. Every wait takes a timeout of its own from the caller, and the runtime asks again for as long as that allows; see Workflows.

Agent calls usually take longer than an entity call. A workflow step that calls an agent should await with an explicit timeout, as ComponentClient.await(answer, 90.seconds) does, and declare a step timeout to match; see Workflows.

Metadata¶

withMetadata(metadata) on an invocation attaches metadata to the call, which the handler reads from its command context. The runtime uses metadata to carry a request's trace from component to component.

Views¶

Views are queried through the view client, not the component client, because a view is not addressed by an instance id: the point of a view is to be queried by attributes rather than by key. An endpoint gets it as clients.viewClient:

ScalaJavaKotlinPythonTypeScriptRustGo

val rows = clients.viewClient.forView(CartRows)

@Get("/checkouts")
public List<CheckoutRows.Row> checkouts() {
  var rows = viewClient().forView(CheckoutRows.class);
  return rows.all();
}

val rows = viewClient().forView(CartRows::class.java)

async def every_cart(client: ComponentClient) -> list[CartRow]:
    rows: list[CartRow] = await client.views.all("cart-rows", CartRow)
    return rows

rows: get("/rows", s.list(CartRow), (ep: ShoppingCartEndpoint) => ep.client.views.all(CartRows.componentId, CartRow)),

fn row(request: &Request) -> Result<CartRow, HttpProblem> {
    let cart_id = request.path("cartId");
    let rows: Vec<CartRow> = request
        .client()
        .query(CartRows, "by-id", cart_id.to_string())?;
    rows.into_iter()
        .next()
        .ok_or_else(|| HttpProblem::new(404, format!("no row for cart '{cart_id}'")))
}

// A view is queried by its id and a declared query, not by an instance: every row of the cart
// rows view, decoded as the row type.
ankka.Get("/rows", func(ctx context.Context, _ *ankka.Request) ([]cart.Row, error) {
    return ankka.QueryView[cart.Row](ctx, "cart-rows", ankka.All())
}),

forView takes the view's companion, so the row type is carried into the queries. The queries themselves are described in Views.

The request context does not follow work to another thread¶

A handler's request — query parameters, headers, the principal — and its trace belong to the handler's own thread. A call made from that thread with invoke or invokeAsync carries the trace, because the call is issued there. Work handed to another thread, such as a Future callback or a thread pool, cannot see the request, and calls made from it appear in the console as unattributed time. Read what you need first, and issue calls from the handler's thread.

Encoding a call¶

In Python, invoke(input, reply=Type) encodes the input with its type's default codec and decodes the reply as reply; without reply the call expects Done, and codec= and reply_codec= override either codec. TypeScript takes both shapes from the handler, so a call needs no annotation. In Scala both come from the handler's given serializers. A refusal is a CommandError in every SDK, carrying the ErrorCode the handler chose.

Pass the request's metadata on¶

A call made from an endpoint should be a child of the request in the console, not a trace of its own. Scala gets that from the request context on the handler's own thread; Python and TypeScript carry it by passing the request's metadata to the client:

ScalaJavaKotlinPythonTypeScriptRustGo

post("/{cartId}/checkouts") { (cartId: String, mode: String) =>
  // The request's context is already on this thread, so the call joins its trace.
  client.forWorkflow(EntityId(cartId)).call(CheckoutWorkflow.start).invoke(mode)
}

/** The body is how long the checkout waits for a confirmation, in milliseconds. */
@Post("/{cartId}/checkouts")
public Done startCheckout(String cartId, @Body String confirmWithin) {
  // The request's context is already on this thread, so the call joins its trace.
  return componentClient()
      .forWorkflow(cartId)
      .call(Checkout::start)
      .invoke(new Checkout.Start(cartId, Long.parseLong(confirmWithin)));
}

post("/carts/{cartId}/checkouts") {
    // The request's context is already on this thread, so the call joins its trace.
    componentClient().forWorkflow(path("cartId")).call(CheckoutWorkflow::start).invokeSuspending(path("cartId"))
}

@post("/{cartId}/checkouts")
async def start_checkout(self, cartId: str, mode: str) -> Done:
    """``mode`` is the body: ``ok``, ``fail`` or ``pause``."""
    return await self.client.with_metadata(self.request.metadata).for_workflow("checkout", cartId).call("start").invoke(mode or "ok", reply=Done)

/** The body is the mode: `ok`, `fail` or `pause`. */
startCheckout: post("/{cartId}/checkouts", s.string, Done, (ep: ShoppingCartEndpoint, req, mode) =>
  ep.client.of(CheckoutWorkflow, req.params.cartId).call(CheckoutWorkflow.handlers.start).invoke(mode || "ok"),
),

/// `mode` is the body: `ok`, `fail` or `pause`.
fn start_checkout(request: &Request, mode: String) -> Result<Done, HttpProblem> {
    let cart_id = request.path("cartId");
    Ok(request
        .client()
        .invoke(CheckoutWorkflow, cart_id, "start", mode)?)
}

// The handler's ctx carries the request's trace, so the call is a child of the request in the
// console rather than a trace of its own; mode is the body: ok, fail or pause.
ankka.Post("/{cartId}/checkouts", func(ctx context.Context, req *ankka.Request, mode string) (string, error) {
    if mode == "" {
        mode = "ok"
    }
    return ankka.Invoke[string](ctx, "checkout", req.Param("cartId"), "start", mode)
}),

with_metadata and withMetadata return a client whose calls carry that metadata; a call made without it starts a trace of its own. The Scala form has the caveat in the request context does not follow work to another thread: work handed to another thread cannot see the context, and there the metadata must be passed explicitly.