Talk to a model with interact_with_llm

Most agent work is a conversation with a model. A11 ships that conversation as an ordinary action, interact_with_llm, whose ports stream the model's output as it arrives. You drive it exactly like any other action: feed its input ports, read its output ports.

Alongside import a11 this page needs a few names from the SDK, which is where the model helpers live:

import a11
from a11.sdk.interact_with_llm import (
    INTERACT_WITH_LLM_SCHEMA,
    interact_with_llm,
)
from a11.sdk.llm import Interaction, LlmHeaders, Role

Build and start the action

Construct the action from its schema, bind the handler, and set the provider / model / key as headers. .run() starts it in the background and hands back the running action, whose ports you now read and write:

import os

interact = (
    a11.Action(INTERACT_WITH_LLM_SCHEMA)
    .bind_handler(interact_with_llm)
    .set_header(LlmHeaders.PROVIDER.value, "gemini")
    .set_header(LlmHeaders.MODEL.value, "gemini-3.5-flash")
    .set_header(LlmHeaders.API_KEY.value, os.environ["GEMINI_API_KEY"])
    .run()
)

Ports are async nodes, reached with interact["<port>"] — the same AsyncNode you met in streaming.

Stream the reply as it arrives

The assistant's visible text lands, already extracted from the raw provider events, on the text_output port. Draining it in a task lets tokens print while the rest of the interaction is still in flight:

async def stream_text():
    async for chunk in interact["text_output"]:
        print(chunk, end="", flush=True)


stream_task = asyncio.create_task(stream_text())

Feed the conversation in

The input side takes three ports:

• interactions — the conversation so far, ending with the new user turn;
• config — model settings (defaults are fine, so we just open and close it);
• tools — tool definitions; here there are none, so we close it empty.

An Interaction is a role plus content chunks:

user_turn = Interaction(
    role=Role.USER,
    content=[a11.to_chunk({"role": "user",
                           "content": [{"type": "text", "text": "Hi!"}]})],
)

async with (
    interact["interactions"] as interactions,
    interact["config"],
    interact["tools"] as tools,
):
    await interactions.put_final(user_turn)   # marks the input turn final
    await tools.put_null_final()              # "no tools this time"

Closing each port (the async with exit) tells the handler that side is complete.

Collect the result

The turns the model produced — its text, and any tool calls — arrive on new_interactions. Read it to completion, then await the streaming task so the last tokens have printed:

new_interactions = []
async for interaction in interact["new_interactions"]:
    new_interactions.append(interaction)
await stream_task

Keep new_interactions around and prepend them (plus the user turn) to the next call's interactions to carry the conversation forward.

Putting it together

import asyncio
import os

import a11
from a11.sdk.interact_with_llm import (
    INTERACT_WITH_LLM_SCHEMA,
    interact_with_llm,
)
from a11.sdk.llm import Interaction, LlmHeaders, Role


async def ask(text: str) -> None:
    interact = (
        a11.Action(INTERACT_WITH_LLM_SCHEMA)
        .bind_handler(interact_with_llm)
        .set_header(LlmHeaders.PROVIDER.value, "gemini")
        .set_header(LlmHeaders.MODEL.value, "gemini-3.5-flash")
        .set_header(LlmHeaders.API_KEY.value, os.environ["GEMINI_API_KEY"])
        .run()
    )

    async def stream_text():
        async for chunk in interact["text_output"]:
            print(chunk, end="", flush=True)

    stream_task = asyncio.create_task(stream_text())

    user_turn = Interaction(
        role=Role.USER,
        content=[a11.to_chunk({"role": "user",
                               "content": [{"type": "text", "text": text}]})],
    )
    async with (
        interact["interactions"] as interactions,
        interact["config"],
        interact["tools"] as tools,
    ):
        await interactions.put_final(user_turn)
        await tools.put_null_final()

    async for _ in interact["new_interactions"]:
        pass
    await stream_task


asyncio.run(ask("Say hello in three languages."))

The full multi-turn, multi-provider version is examples/002-llm-interactions.

Next: the model call above ran in your process. See how to move it behind a server you call over the network, then give the model a tool it can call back into.