Middleware and Continuations

Python and Rust workers expose the same registration model even though their closure syntax differs. The checked examples share configuration and observable behavior so a reader can compare runtime mechanics instead of reverse-engineering two unrelated demonstrations.

Callback Families

The Event metadata injector, subscriber, three event sanitizers, five tool registrations, and six LLM registrations make 16 surfaces. Registration names are component-local; priority and break_chain are sent in registration metadata and enforced by the host after it merges visible middleware.

Inject Event Metadata

Choose a worker language to register a callback that proposes metadata additions:

1
async def add_plugin_context(event):
2
    del event
3
    return {"app.source": "plugin"}
4
5
ctx.register_event_metadata_injector(
6
    "plugin-context",
7
    add_plugin_context,
8
    priority=0,
9
)

Relay sends the immutable Event snapshot through the existing unary Invoke RPC. It validates and merges accepted additions before Event sanitizers run. A callback error omits that callback’s additions without dropping the Event. Stopping the worker removes the registration.

Register Synchronous and Asynchronous Callbacks

Python callbacks can be ordinary functions or coroutines on surfaces that the SDK normalizes. This excerpt uses a synchronous event sanitizer and an asynchronous LLM sanitizer because codec operations call back to the host. All three event sanitizers return the same complete field object.

1
def sanitize_event(_event, fields):
2
    metadata = _redact(fields.get("metadata") or {}, redact_keys)
3
    if not isinstance(metadata, dict):
4
        metadata = {"original": metadata}
5
    return {
6
        "data": _redact(fields.get("data"), redact_keys),
7
        "category_profile": _redact(
8
            fields.get("category_profile"), redact_keys
9
        ),
10
        "metadata": {**metadata, "plugin_tag": tag},
11
    }
12
13
async def sanitize_llm_request(request, codec_context):
14
    request = deepcopy(request)
15
    codec = codec_context.resolve_codec()
16
    if codec is not None:
17
        annotated = await codec.decode(request)
18
        annotated = _redact(annotated, redact_keys)
19
        request = await codec.encode(annotated, request)
20
    request["content"] = _redact(request.get("content"), redact_keys)
21
    return request
22
23
ctx.register_mark_sanitize_guardrail(
24
    "documentation_mark_sanitizer", sanitize_event, priority=10
25
)
26
ctx.register_scope_sanitize_start_guardrail(
27
    "documentation_scope_start_sanitizer", sanitize_event, priority=10
28
)
29
ctx.register_scope_sanitize_end_guardrail(
30
    "documentation_scope_end_sanitizer", sanitize_event, priority=10
31
)
32
ctx.register_llm_sanitize_request_guardrail(
33
    "documentation_llm_request_sanitizer",
34
    sanitize_llm_request,
35
    priority=10,
36
)

The equivalent Rust closure always returns a future. The codec proxy is asynchronous in a worker because decode and encode are authenticated host-runtime RPCs.

1
context.register_llm_sanitize_request_guardrail(
2
    "documentation_llm_request_sanitizer",
3
    10,
4
    {
5
        let keys = config.observe.redact_keys.clone();
6
        move |mut request, codec_context| {
7
            let keys = keys.clone();
8
            async move {
9
                if let Some(codec) = codec_context.resolve_codec() {
10
                    let annotated = codec.decode(&request).await?;
11
                    let annotated = serde_json::to_value(annotated)
12
                        .map(|value| redact(value, &keys))
13
                        .and_then(serde_json::from_value)
14
                        .map_err(|error| {
15
                            WorkerSdkError::InvalidInput(error.to_string())
16
                        })?;
17
                    request = codec.encode(&annotated, &request).await?;
18
                }
19
                request.content = redact(request.content, &keys);
20
                Ok(Some(request))
21
            }
22
        }
23
    },
24
);

The codec object is scoped to this invocation. Its opaque capability expires when the callback finishes, even if the worker retained a language-level object that previously referenced it.

Return Complete Request Outcomes

Request intercepts affect real execution. The LLM form returns more than a provider request because Relay must carry its normalized annotation and accounting separately.

1
priority = requests["priority"]
2
break_chain = requests["break_chain"]
3
header_name = requests["header_name"]
4
header_value = requests["header_value"]
5
6
def llm_request(_name, request, annotated):
7
    rewritten = deepcopy(request)
8
    rewritten["headers"] = {
9
        **(rewritten.get("headers") or {}),
10
        header_name: header_value,
11
    }
12
    marks = (
13
        [PendingMarkSpec(
14
            name="example.python_worker.llm_request",
15
            data={"tag": tag},
16
        )]
17
        if execution["emit_pending_marks"]
18
        else []
19
    )
20
    return LlmRequestInterceptOutcome(
21
        request=rewritten,
22
        annotated_request=annotated,
23
        pending_marks=marks,
24
        optimization_contributions=[
25
            LlmOptimizationContribution(
26
                producer="examples.python_grpc_worker",
27
                kind="request_rewrite",
28
                applied=True,
29
            )
30
        ],
31
    )
32
33
ctx.register_llm_request_intercept(
34
    "documentation_llm_request",
35
    llm_request,
36
    priority=priority,
37
    break_chain=break_chain,
38
)

Returning only rewritten would be the wrong callback result for this surface. The tool request intercept does return JSON directly because it has no annotated request or LLM optimization accounting.

Continuation Behavior

When the example enables repeat_downstream, it starts two concurrent downstream calls and returns the first response after both calls settle. The first call alone determines whether the wrapper succeeds. The second call is intentional demonstration code: it can still consume provider capacity, incur cost, and cause provider-side effects even though Relay does not expose its result to the application. Its failure is deliberately ignored.

ToolNext, LlmNext, and LlmStreamNext are host proxies identified by an opaque continuation ID. Calling one issues a host-runtime RPC under the scope snapshot captured for that worker invocation. A callback can call a unary proxy zero, one, or multiple times, including concurrently when the SDK type permits cloning. Each call can repeat side effects, provider charges, events, and downstream middleware.

ToolNext returns ToolExecutionResult, not a raw JSON value. Forwarding middleware must preserve both downstream.result and downstream.annotation in its outcome. The Python example keeps the downstream annotation under upstream while adding its own worker metadata; the Rust example forwards it unchanged. A repeated tool continuation returns another independent structured result; it still does not expose downstream pending marks.

The example uses one ordinary wrapper and one explicitly requested concurrent path. It does not retry implicitly. Tool pending marks remain in the tool execution outcome; LLM annotations, pending marks, and optimization contributions remain in the request intercept outcome. The unary execution callback returns only provider-response JSON.

LlmStreamNext returns a remote stream. The worker transforms each chunk as it arrives and yields immediately. If the host cancels the invocation or the consumer abandons the stream, the Python task receives asyncio.CancelledError and the Rust callback future is aborted. Cleanup belongs in finally or a drop-safe guard. Acknowledged cancellation does not prove that external blocking work has stopped.

1
priority = execution["priority"]
2
emit_pending_marks = execution["emit_pending_marks"]
3
tag = settings["tag"]
4
5
async def tool_execution(name, args, next_call):
6
    downstream = await next_call.call(args)
7
    marks = (
8
        [PendingMarkSpec(
9
            name="example.python_worker.tool_execution",
10
            data={"tool_name": name, "tag": tag},
11
        )]
12
        if emit_pending_marks
13
        else []
14
    )
15
    return ToolExecutionInterceptOutcome(
16
        result=downstream.result,
17
        annotation={
18
            "upstream": downstream.annotation,
19
            "worker": {"tool_name": name, "tag": tag},
20
        },
21
        pending_marks=marks,
22
    )
23
24
async def llm_execution(_name, request, next_call):
25
    content = request.get("content")
26
    repeat = isinstance(content, dict) and content.get("repeat_downstream") is True
27
    if repeat:
28
        first, _second = await asyncio.gather(
29
            next_call.call(request),
30
            next_call.call(request),
31
            return_exceptions=True,
32
        )
33
        if isinstance(first, BaseException):
34
            raise first
35
        return first
36
    return await next_call.call(request)
37
38
async def llm_stream_execution(_name, request, next_call):
39
    async for chunk in next_call.call(request):
40
        if isinstance(chunk, dict):
41
            yield {**chunk, "plugin_stream": True}
42
        else:
43
            yield chunk
44
45
ctx.register_tool_execution_intercept(
46
    "documentation_tool_execution", tool_execution, priority=priority
47
)
48
ctx.register_llm_execution_intercept(
49
    "documentation_llm_execution", llm_execution, priority=priority
50
)
51
ctx.register_llm_stream_execution_intercept(
52
    "documentation_llm_stream_execution",
53
    llm_stream_execution,
54
    priority=priority,
55
)

The second unary result is awaited even though the first response is selected. That prevents an unobserved continuation from outliving the worker callback. In the stream case, each error remains an error item and no chunks are requested before the consumer polls the mapped stream.

Verify the Shared Contract

Use the following procedure to verify equivalent behavior across the two worker SDKs:

1. Assert that registration returns exactly 15 unique surface and local-name pairs.
2. Exercise all three event sanitizers and both tool and LLM sanitizer directions; prove that only observability values change.
3. Block configured tool and model names, rewrite allowed requests, and preserve an annotated LLM request through later request intercepts and the managed start event.
4. Call each unary continuation once, then use the explicit repeated path to call it twice concurrently and account for both downstream invocations.
5. Consume a transformed stream incrementally, then cancel a second stream and confirm worker cleanup.
6. Inspect emitted outcomes to ensure pending marks and optimization contributions are not present in application results.

Success means Python and Rust exhibit the same Relay semantics despite their different callback syntax. The code examples show each callback contract, while the atomic Python example tests and worker SDK integration suites verify callback behavior, authenticated transport, and host invocation.