Skip to content
Peppy

Services

Services implement a request-response communication pattern between nodes. A node exposes a service to handle incoming requests, and other nodes consume that service to send requests and receive responses.

Use services for operations that need a result, such as querying a node’s state, toggling a feature, or triggering a one-time computation. For long-running work that needs progress feedback or cancellation, use an action instead; for the full map of mechanisms, see Choosing a communication pattern.

Exposing a service

Section titled “Exposing a service”

A node that handles service requests declares its services under interfaces.services.exposes in its peppy.json5. Each service defines a name, an optional request_message_format, and an optional response_message_format:

{
  peppy_schema: "node/v1",
  manifest: {
    name: "uvc_camera",
    tag: "v1",
  },
  interfaces: {
    services: {
      exposes: [
        {
          name: "enable_camera",
          request_message_format: {
            enable: "bool",
          },
          response_message_format: {
            enabled: "bool",
            error_msg: {
              $type: "string",
              $optional: true
            },
          },
        },
        {
          // A service without a request body; the caller just needs the response.
          name: "get_camera_info",
          response_message_format: {
            card_type: "string",
            size: "string",
            interval: "string"
          },
        },
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "uvc_camera"]
  },
}

Both request_message_format and response_message_format are optional. A service with no request body acts like a simple getter, and a service with no response body acts like a fire-and-forget trigger.

Handling requests

Section titled “Handling requests”

After running peppy node sync, the code generator creates a module for each exposed service under peppygen.exposed_services (Python) / peppygen::exposed_services (Rust). Use handle_next_request to process incoming requests. Each request is handled in a separate task so that the main async block is not blocked:

import asyncio


from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.exposed_services import enable_camera




async def handle_requests(node_runner: NodeRunner):
    def handler(request):
        print(
            f"enable_camera request from {request.instance_id}: "
            f"enable = {request.data.enable}"
        )
        return enable_camera.Response(
            enabled=request.data.enable,
            error_msg="ok",
        )


    await enable_camera.handle_next_request(node_runner, handler)




async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(handle_requests(node_runner))]




def main():
    NodeBuilder().run(setup)




if __name__ == "__main__":
    main()

The handler may be a plain function or an async def; an async handler is awaited before the response is sent.

The request argument contains:

  • instance_id: the consumer instance that sent the request, read straight from the request context. The producer is binding-agnostic; it doesn’t know which slot on the consumer this call is heading to.
  • core_node: the core node hosting that consumer instance, read from the request context.
  • data: the deserialized request payload (only present when a request_message_format is defined).

handle_next_request processes a single request and returns. To serve requests continuously, call it in a loop inside a spawned task:

async def serve_requests(node_runner: NodeRunner):
    while True:
        await enable_camera.handle_next_request(node_runner, handler)




# in setup:
return [asyncio.create_task(serve_requests(node_runner))]

For a service without a request body, the handler receives a Request with only the request context (instance_id and core_node):

from peppygen.exposed_services import get_camera_info




async def handle_info_requests(node_runner: NodeRunner):
    def handler(request):
        print(f"get_camera_info request from {request.instance_id}")
        return get_camera_info.Response(
            card_type="UVC Webcam",
            size="1920x1080",
            interval="30fps",
        )


    await get_camera_info.handle_next_request(node_runner, handler)

A producer exposes its service exactly once and serves any consumer that calls it. Starting the producer before any consumer (or after them) is equally valid.

Consuming a service

Section titled “Consuming a service”

A node that calls a service declares what it consumes under interfaces.services.consumes. Dependencies are declared in manifest.depends_on and referenced by link_id in the interface:

{
  peppy_schema: "node/v1",
  manifest: {
    name: "robot_brain",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "uvc_camera", tag: "v1", link_id: "uvc_camera" },
      ]
    },
  },
  interfaces: {
    services: {
      consumes: [
        {
          link_id: "uvc_camera",  // References depends_on.nodes[].link_id
          name: "enable_camera",        // Service name on that node
        },
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "robot_brain"]
  },
}

Calling a service

Section titled “Calling a service”

The code generator creates a module for each consumed service under peppygen.consumed_services (Python) / peppygen::consumed_services (Rust). Use poll to send a request and wait for a response. The caller selects which bound producer handles the call by passing one explicit target, obtained from the slot’s cardinality-typed accessor: on a one slot bound_producer() returns the sole producer directly, on a one_or_more slot bound_producers() returns a never-empty set whose first() needs no unwrap, and on a zero_or_more slot it returns a possibly empty slice whose empty case the caller must handle. The explicit target parameter itself has the same shape for every cardinality. The candidates are fixed by the application bindings at launch, so by the time poll runs the route set is already validated.

import asyncio


from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_services.uvc_camera import enable_camera




async def call_service(node_runner: NodeRunner):
    # `one`: launch resolved the slot to exactly one producer, so the
    # accessor is singular and infallible; no emptiness handling exists.
    # The target is still passed explicitly, same discipline as the
    # multi cardinalities.
    camera = enable_camera.bound_producer(node_runner)
    request = enable_camera.Request(enable=True)
    response = await enable_camera.poll(
        node_runner,
        camera,  # the slot's sole producer
        request,
        5.0,  # timeout (seconds)
    )


    error_msg = response.data.error_msg if response.data.error_msg is not None else "<none>"
    print(
        f"enable_camera result: instance={response.instance_id} "
        f"enabled={response.data.enabled} error={error_msg}"
    )




async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(call_service(node_runner))]




def main():
    NodeBuilder().run(setup)




if __name__ == "__main__":
    main()

The target must be a member of the slot’s own bound set, the one its bound-producer accessor returns; a target outside the set (including a producer bound to a different slot of the same consumer) fails with a runtime error before anything reaches the wire. A ProducerRef yielded by the slot’s own topic subscription is a member by construction, so the natural receive-a-frame-then-call-that-camera flow needs no defensive code.

The response contains:

  • instance_id: the producer instance that handled the request, read from the response context.
  • core_node: the core node hosting that producer instance, read from the response context.
  • data: the deserialized response payload.

For a service without a request body, poll simply takes no request:

from peppygen.consumed_services.uvc_camera import get_camera_info


camera = get_camera_info.bound_producer(node_runner)
response = await get_camera_info.poll(node_runner, camera, 5.0)


print(f"Camera: {response.data.card_type} {response.data.size}")

Calling every bound producer

Section titled “Calling every bound producer”

Codegen provides only single-target operations; calling every member of a multi-cardinality slot is a plain loop at the call site (a one slot has no set to loop over: its singular bound_producer() is the sole target). Sequential versus concurrent execution, partial-result collection, and whether one failure cancels the other calls are application decisions; there is no atomic broadcast. Here the camera slot declares cardinality: "one_or_more":

# Runs once per bound producer, in binding declaration order.
# `one_or_more`: the list is never empty, so the body runs at least once.
# `zero_or_more`: an empty list makes the loop a no-op; no request is sent.
for camera in enable_camera.bound_producers(node_runner):
    request = enable_camera.Request(enable=True)
    response = await enable_camera.poll(node_runner, camera, request, 5.0)
    print(f"{response.instance_id}@{response.core_node}: enabled={response.data.enabled}")

To address one member of a multi slot instead of all of them, pick it from the same set; on a one_or_more slot the set is never empty, so selecting the first member needs no unwrap:

camera = enable_camera.bound_producers(node_runner)[0]

Bindings and routing

Section titled “Bindings and routing”

Routing for services is the same consumer-side model used by topics. A binding KEY: VALUE creates a private channel from producer instance VALUE to one of the consumer’s declared slots; the producer itself doesn’t know or care about bindings.

A service slot resolves through its bindings: the generated poll checks the caller-selected target against the slot’s bound set and sends the wire request directly to it, carrying the producer’s full (core_node, instance_id) wire address. No discovery is involved. How many producers may be bound to the slot is its declared cardinality (one when omitted); a multi-cardinality slot takes an array of instance ids, and a one / one_or_more slot with no binding is rejected at launch validation, before anything spawns. Because a request/response call needs exactly one responder, every call selects exactly one member of the bound set.

In a launcher / stack config:

{
  source: { local: "./consumer" },
  instances: [{
    instance_id: "my_consumer",
    links: { uvc_camera: "my-camera-instance" },
  }],
}

or, when launching a single node during development:

Terminal window
peppy node run --link uvc_camera@my-camera-instance .

Worked example: openarm01_backbone

Section titled “Worked example: openarm01_backbone”

A consumer that wires two depth cameras to two dedicated slots:

openarm01_backbone/peppy.json5
{
  manifest: {
    name: "openarm01_backbone",
    tag: "v1",
    depends_on: {
      contracts: [
        { name: "depth_camera", tag: "v1", link_id: "wrist_left_camera" },
        { name: "depth_camera", tag: "v1", link_id: "wrist_right_camera" },
      ],
    },
  },
  // ...
}
peppy_launcher.json5
{
  deployments: [
    { source: { name: "depth_camera:v1" }, instances: [
      { instance_id: "left_cam" },
      { instance_id: "right_cam" },
    ]},
    { source: { name: "openarm01_backbone:v1" }, instances: [
      { instance_id: "backbone_inst_1", links: {
        wrist_left_camera: "left_cam",
        wrist_right_camera: "right_cam",
      }},
    ]},
  ],
}

Three contract statements follow from this manifest:

  1. poll on the nested wrist_left_camera.<service> (Python) / wrist_left_camera::<service> (Rust) module reaches left_cam.
  2. poll on the nested wrist_right_camera.<service> (Python) / wrist_right_camera::<service> (Rust) module reaches right_cam.
  3. If the wrist_right_camera binding line were removed, validation would reject the launch (every declared one / one_or_more slot must be bound): a service call has no wildcard fallback.

Why an explicit single target?

Section titled “Why an explicit single target?”

The underlying Zenoh transport would broadcast an untargeted service query to every matching producer (QueryTarget::All), and every producer’s user handler would run even though the consumer only ever consumes the first reply. For idempotent reads that wastes work; for state-changing services it can cause real-world side effects on producers the consumer never intended to reach. Requiring every call to name one member of the slot’s validated bound set removes that hazard by construction: the selected target carries the producer’s full (core_node, instance_id) wire address, so the call addresses exactly one producer with no probe round-trip and no discovery race, both the request and the response stay pinned to it, and an out-of-set target (which plan-time binding validation cannot check) is rejected before it reaches the wire.

If the selected producer dies, the call surfaces ServiceUnreachable (a ConnectionError in Python) and the caller can retry once the producer is back; the bound set itself never shrinks or rebinds at runtime.

Validator rules

Section titled “Validator rules”

The launcher validator runs these checks before the stack starts:

  1. Every KEY must name a declared slot, and every declared slot must resolve. A binding whose KEY matches no depends_on link_id is rejected; there are no free-form keys. A declared one / one_or_more slot with no binding entry fails the launch before anything is spawned; a zero_or_more slot with no entry resolves to the empty set.
  2. The value’s shape must match the slot’s cardinality. A one slot takes a scalar, a multi slot takes an array, an empty array meets only zero_or_more, and duplicate targets within one slot are rejected. Repeated --link KEY@… flags accumulate on a multi slot and are a hard error on a one slot.
  3. Every target must satisfy the slot, checked per bound instance. A target instance_id that deploys a different node than the slot expects (or one that does not implement the requested contract) is rejected.
  4. Stack-wide instance_id uniqueness. Every instance_id must be unique across the entire stack, not just within a (node_name, node_tag) group. The --link syntax names producers by instance_id, so a duplicate would make the binding ambiguous.
  5. Bindings are stamped with the daemon’s core_node. The wire addresses producers by the full (core_node, instance_id) pair; the validator stamps the launching daemon’s core_node into every resolved binding, preserving application declaration order, so generated calls address exactly the selected producer and never match on instance_id alone.

Error handling

Section titled “Error handling”

Service calls can fail with three error types:

  • ServiceUnreachable (ConnectionError in Python): no instance is listening for that service.
  • ServiceTimeout (TimeoutError in Python): no response was received within the timeout.
  • ServiceError (RuntimeError in Python): the handler returned an error, which is propagated back to the caller.

If the service handler returns an Err in Rust or raises an exception in Python, the error is forwarded to the caller rather than silently timing out. This means a failing handler does not block the service from continuing to accept new requests.