Run an A11 action in a second process over WebSocket

This guide produces a working two-process application on your machine. One terminal serves a registered text action; a second terminal discovers it, sends input and prints the streamed result. The action implementation remains independent of WebSocket setup, so the same interface can connect an existing model worker or internal service.

Start the worker#

Open a fresh directory and activate a Python 3.11+ virtual environment. The commands in Run this guide above download both files, start the worker in terminal one, and call it from terminal two.

Leave this terminal running. The server imports first_stream.py, finds its REGISTRY and listens at ws://127.0.0.1:18011/ws.

The served file also works as a local program: python first_stream.py prints a sentence a word at a time. Its if __name__ == "__main__" guard keeps that local caller from running when the server imports it.

Call the worker from a second terminal#

Open terminal two in the same directory, activate the same virtual environment and run python remote_client.py if you have not already done so.

Expected output, arriving incrementally:

The function stays the function.

If the connection is refused, check that terminal one is still running and that its port is 18011. If that port is already occupied, choose another port in both the server command and the URL in remote_client.py. A path mismatch matters too: this example uses /ws.

Keep the handler independent of the transport#

Test the word-splitting behavior locally with the same handler. It accepts text and writes to a declared output:

from typing import Annotated

import a11


@REGISTRY.action
async def read(
    text: str,
    words: Annotated[a11.AsyncNode, a11.OutputPort(mimetype="text/plain")],
):
    for word in text.split():
        await words.put(f"{word} ")

It accepts a string and writes strings. Annotated gives the words node its text media type, so the registered interface has a typed input named text and a streaming output named words. The handler contains no socket, hostname or client configuration.

The landing example adds a second output for structured progress. This remote exercise keeps just the word stream so you can focus on connecting the caller. A stable action interface keeps changes in execution location inside the calling and serving code.

Connect from the existing application#

Your worker is running in terminal one, but the caller in terminal two has no access to its Python objects. The caller needs the worker’s address and an action interface it can send input through. This is the same problem a desktop app faces when its model runs in a separate service.

GatewayConnection.connect opens that connection. The ten-second limit lets the caller report an unreachable worker instead of waiting indefinitely:

caller = await GatewayConnection.connect(
    "ws://127.0.0.1:18011/ws",
    timeout=timing.Duration.seconds(10),
)

Once connected, the caller asks which actions the worker serves. Discovery provides the read schema, including its text input and words output, so the client can construct a call:

registry = a11.ActionRegistry()
await install_peer_actions(caller, registry)
action = caller.action("read", registry.get_schema("read"))
await action.call()

Compare the two ways to start the same registered handler. A local caller uses:

action = REGISTRY.make_action("read").run()
await action["text"].finalize("The function stays the function.")

await action.call() in the previous snippet dispatches the remote call. In the local example, action.run() starts a local handler. This is the part that changes. Neither the handler nor the names and representations of its inputs and outputs change.

The caller sends its input, reads values and checks completion:

await action["text"].finalize("The function stays the function.")
async for word in action["words"]:
    print(word, end="", flush=True)
await action.wait()

finalize closes the input after its one value. The async loop consumes incremental output. wait checks the action’s final status. If the worker sends half a document and then crashes, a UI can keep the received text visible while marking the extraction as failed. Only a successful final status should enable an action such as “Use extracted text.”

Change the sentence, then change the worker’s speed#

Change the sentence in remote_client.py, then run the client again. The server keeps serving the same handler. Change the handler's delay, stop terminal one with Ctrl+C, restart its serve command and run the client again. The pacing changes without changing the caller's output loop.

The complete remote client puts the output loop and final status check inside one deadline:

async with asyncio.timeout(10):
    async for word in action["words"]:
        print(word, end="", flush=True)
    await action.wait()

Its enclosing try always releases the connection, including when reading fails or the user presses Ctrl+C:

finally:
    await caller.aclose()

The caller owns connection cleanup; the word-producing handler contains no connection code. Stop the server with Ctrl+C when finished. Both processes run on the same machine through the loopback address.

Put image generation on a machine with enough GPU memory#

Your image model may not fit on the laptop running the application. Put the generation handler and its weights on a GPU server; keep the prompt input, progress reader and image viewer in the application. The image-generation walkthrough already makes this split: its Python client receives progress and a PNG from the hosted GPU.

Two other concrete reasons to use a separate machine are:

• Several applications use the same large model. A shared worker can keep it loaded between requests instead of making every application load its own copy. The worker controls how many requests the GPU runs at once.
• A tool needs an internal database. Run the lookup beside that database, where the required network access and credentials are available. Expose an authenticated action for the lookup; callers consume its result.

The two-terminal exercise establishes the action contract. Moving its server off your laptop also requires a reachable address and authentication. The loopback URL 127.0.0.1 always refers to the caller’s own machine, so it cannot address that remote worker. The hosting guide explains how to give an A11 service a named reachable endpoint and configure access. The handler’s input and output contract can remain the same through that deployment change.

Complete source#

Download the server's first_stream.py and remote_client.py. Both files are self-contained and use the public SDK directly.

import asyncio
from typing import Annotated

import a11

REGISTRY = a11.ActionRegistry()


@REGISTRY.action
async def read(
    text: str,
    words: Annotated[a11.AsyncNode, a11.OutputPort(mimetype="text/plain")],
):
    for word in text.split():
        await words.put(f"{word} ")


async def main():
    action = REGISTRY.make_action("read").run()
    await action["text"].finalize(
        "Your functions. Your architecture. AI-native."
    )
    async for word in action["words"]:
        print(word, flush=True)
    await action.wait()


if __name__ == "__main__":
    asyncio.run(main())

import asyncio

import a11
from a11 import timing
from a11.client.connection import GatewayConnection
from a11.client.discovery import install_peer_actions


async def main():
    caller = await GatewayConnection.connect(
        "ws://127.0.0.1:18011/ws",
        timeout=timing.Duration.seconds(10),
    )
    try:
        registry = a11.ActionRegistry()
        await install_peer_actions(caller, registry)
        action = caller.action("read", registry.get_schema("read"))
        await action.call()
        await action["text"].finalize("The function stays the function.")
        async with asyncio.timeout(10):
            async for word in action["words"]:
                print(word, end="", flush=True)
            await action.wait()
        print()
    finally:
        await caller.aclose()


if __name__ == "__main__":
    asyncio.run(main())

To find out why a remote stock lookup returned “Choose an alternative,” continue with the debugging walkthrough.