Add streaming and remote calls to an existing AI application

Run a streaming action, then add A11 to an operation that needs streams, lifecycle status or remote execution. Application policy stays in the code that already owns it.

Keep application policy in application code#

Consider a research feature where the model proposes topics and the application decides how many searches to run, which sources are allowed, how long to wait and whether the user can cancel. These product decisions deserve the same explicit code and tests as authorization or billing.

A graph can be a useful representation when a workflow itself is data: a user edits steps, an operator inspects a durable workflow, or a scheduler resumes work after a restart. It adds less when all you need is a loop and two concurrent requests. Choose it because the application needs those semantics.

An ordinary function provides a useful starting interface. Its input says what the caller must provide, its output says what the caller may rely on, and its implementation can change without requiring a new architecture for the rest of the application.

Add a streaming output#

A model response, search, transcription or document parser often has useful output before it has a final answer. Waiting for a single return value hides that useful work.

This local text action shows the complete lifecycle with no provider credentials or nondeterministic output.

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())

Download the file. Install with pip install a11-kit, then run python first_stream.py.

The async generator becomes an action with a typed streaming output. The caller reads values while the handler is still working and checks completion with wait().

That distinction matters in production. A response can emit ten useful fragments and then fail. A UI can show the partial result, identify it as incomplete and let the user retry. Treating the arrival of any text as success loses that information.

Run independent work with Python concurrency#

If two operations are independent, start two tasks. If only three may run at a time, use a semaphore. If one result is required before another operation can start, await it. Keeping those choices in ordinary code makes it easier to see the actual policy.

A11 contributes the action interface: named inputs and outputs, streams, lifecycle status, and the connection to remote callers. Python remains responsible for your choice of tasks. Wrapping arbitrary Python tasks does not automatically make them nested A11 actions; use the runtime's nested-action interface when their parent/child relationship matters for debugging and cancellation.

Move execution when the operation needs it#

Eventually, a model needs a GPU, a tool needs a credential that cannot live in the browser, or a long-running job needs a service. Those are reasons to move execution.

With A11, the handler and its port contract can stay the same. The caller configures a connection and dispatches a call instead of starting a local handler. There is still deployment work: authentication, timeouts and resource ownership do not disappear. What stays stable is the application interface.

Add A11 to one operation in an existing stack#

Choose a result you can verify before changing any architecture:

Start with the row that describes a current problem. The local stream guide adds an action interface; the remote guide changes only its dispatch path.

Choose the next capability#

Function → streaming → concurrency → tools → remote execution → connectivity → debugging.

These capabilities are independent. Expose an existing operation as a tool, use LangGraph for workflow routing, or keep execution local until a separate service becomes useful.

A11 supplies the runtime interface for streams, tools and remote calls. The application retains its workflow, policy and provider choices.