Sessions, Streaming & ModelConfig

#Sessions, Invocation, and Streaming

A session is one run of your agent: you give it a message, it works through the request, and it hands back a result. You can run it and wait for the answer, or watch it work live as it goes. This guide covers how to start a session, read what comes back, tune a single run, and run many at once.

#What a session is

Every call to invoke(), stream(), or one of their async variants starts a session — one turn of work for an Agent or Swarm. Within that turn the runtime evaluates your request, runs any tools it needs, and produces a response. When the turn ends, the session is complete.

Most useful conversations span more than one turn. To keep turns connected, pass a thread_id:

from maivn import Agent
from maivn.messages import HumanMessage

agent = Agent(name='assistant', api_key='...')

# First turn
agent.invoke(
    [HumanMessage(content='My name is Dana.')],
    thread_id='conversation-1',
)

# Later turn — same thread, so the agent has the earlier context
agent.invoke(
    [HumanMessage(content='What is my name?')],
    thread_id='conversation-1',
)

A thread_id is just a string you choose. Reuse it across turns that belong to the same conversation, and use a fresh one to start a clean session. Threads govern the conversational recall an agent carries from turn to turn. See Memory and Recall for the cross-thread reuse layer (skills, bound resources, promoted insights).

Note Independent runs should use different threads (or none). A shared thread_id makes concurrent calls write into the same conversation, which is rarely what you want for unrelated work. See Running many at once.

#Explicit thread operations

For a conversation in application code, bind a thread once and keep using the same handle. Client, Agent, and Swarm all expose thread():

conversation = agent.thread()  # Selects a new local ID; sends no request.
first = conversation.send([HumanMessage(content='My name is Dana.')])
second = conversation.send('What is my name?')
print(second.response)

# Reattach using an ID saved by your application.
conversation = agent.thread(conversation.thread_id)

send / asend return the usual InvokeResponse, and stream / astream yield the usual events. They use the owner's normal execution path, including full message lists, tools, hooks, attachments, and structured output. Scope defaults still apply. Per-call invocation arguments can be passed as usual. The handle fixes thread_id and rejects attempts to change it.

Use state / astate to inspect the thread, or events / aevents to replay the current turn's events. The handle retains the observed session ID and last event position, so it can reconnect after a partially consumed stream. Starting a new turn resets them. Positions belong to one session, not the entire conversation. A handle with an observed session keeps replaying that run; attach a fresh handle to follow a turn started by another client. Event replay observes execution; it does not start a local Python tool broker. Consume the original invocation stream when local tools need to run.

A handle permits one active send or stream at a time. Close an unfinished stream before starting the next operation. Explicit invoke(..., thread_id=...) and RunOptions(thread_id=...) remain supported.

#Dispatch and receive an acceptance receipt

The explicit thread APIs also have async counterparts on Agent, Swarm, and Client: astart_thread, apost_thread_message, atime_travel_thread, asubmit_approval, aget_thread, and athread_events. Scope methods retain the same resource preparation, system prompt, and options as their synchronous counterparts.

start_thread, post_thread_message, and time_travel_thread return an acceptance receipt without waiting for completion. They forward supported run options, skills, tools, private data, and swarm members to the runtime. Merely following a receipt with thread_events does not execute SDK-local Python tools. Prefer the bound handle's send or stream when using those tools.

from maivn import RunOptions


async def continue_conversation(agent):
    started = await agent.astart_thread("Prepare a short summary.")
    async for event in agent.athread_events(
        started.thread_id,
        options=RunOptions(resume_from_position=started.stream_position),
    ):
        print(event.event_type)
    return await agent.apost_thread_message(started.thread_id, "Make it shorter.")

post_thread_message and time_travel_thread, including their async and streaming variants, require exactly one message: a string, a message object, or a one-element sequence. Empty or multiple-message sequences raise ValueError before a thread request is sent. Use start_thread when supplying an initial conversation. Earlier SDK versions silently kept only the last message in continuation calls. This restriction does not apply to invoke, stream, or bound-handle send and stream: those accept full message lists on every turn.

#invoke vs stream vs ainvoke vs astream

The four entry points differ on two axes: sync or async, and wait for the result or watch it live.

All four exist on both Agent and Swarm, and they accept the same tuning parameters.

#invoke — run and wait

invoke() runs the agent and returns a single InvokeResponse once the turn is finished.

response = agent.invoke([HumanMessage(content='What is the weather in Austin?')])
print(response.response)

#stream — watch it work live

stream() yields events as the turn unfolds: tool execution, phase changes, assistant text deltas, and a terminal final event carrying the completed payload. Each item is a StreamEvent with canonical event_type/data and compatibility name/payload properties.

for event in agent.stream([HumanMessage(content='Think through this problem')]):
    if event.name == 'system_tool_chunk':
        print(event.payload.get('text', ''), end='')
    elif event.name == 'final':
        print('\nDone')

Status-message and enrichment events may appear on the same stream. Consumers should branch on event names rather than enabling a separate invocation flag.

For product integrations, treat stream() as the raw transport layer. If you are forwarding execution state to a browser or app, normalize the raw events first:

from maivn.events import normalize_stream

raw_events = agent.stream(
    [HumanMessage(content='Think through this problem')],
)

for event in normalize_stream(raw_events, default_agent_name='assistant'):
    send_to_frontend(event.model_dump())

See Frontend Events for the wire format, ready-made client examples, and the enrichment phase labels surfaced during a turn.

#ainvoke / astream — the async variants

ainvoke() and astream() use asynchronous execution and streaming. They take the same parameters and can be awaited or iterated inside an event loop.

# Awaitable single result
response = await agent.ainvoke([HumanMessage(content='Summarize Q1')])

# Async event stream
async for event in agent.astream([HumanMessage(content='Walk me through it')]):
    handle(event)

#Stopping a stream early

When leaving before exhaustion, close the iterator explicitly so the SDK can release its HTTP reader and stream worker. Use closing for synchronous streams and aclosing for asynchronous streams:

from contextlib import closing, aclosing

with closing(agent.stream('Explain the plan')) as events:
    for event in events:
        if event.event_type == 'final':
            break

async with aclosing(agent.astream('Explain the plan')) as events:
    async for event in events:
        if event.event_type == 'final':
            break

The synchronous bridge keeps a bounded event buffer and propagates closure to its asynchronous source. A bare break does not close an iterator you retain. Closing consumption is a local cleanup operation. To cancel remote execution, use client.cancel(session_id) explicitly.

Swarm note Swarm.invoke() / Swarm.stream() and their async variants work the same way, but accept either a single message or a sequence, and all tuning arguments are keyword-only. See Multi-Agent.

#Reading results

A finished turn returns a InvokeResponse. The two fields you reach for most often are response and result.

Use .response when you want the natural-language answer, and .result when the turn produced a typed object you intend to consume programmatically.

# Conversational answer
response = agent.invoke([HumanMessage(content='Summarize this thread.')])
print(response.response)

# Structured answer (see Structured Output guide for the full pattern)
response = agent.invoke(
    [HumanMessage(content='Build the weather report.')],
    force_final_tool=True,
)
print(response.result)  # the validated final-tool object

When a turn produces no structured output, .result is empty and .response carries the answer. When you force a final tool, .result holds the validated schema. The two are complementary, not exclusive: a turn can populate both.

For per-call token accounting, read response.token_usage. See Billing and Usage.

#Tuning a run without internals

Two knobs let you shape the trade-off between speed and depth on a per-call basis. Most calls should use broad intent hints and let the runtime handle the details. When you need stricter control, ModelConfig lets you scope tier or exact-model choices to the public configurable framework parts.

#model - tier or scoped model config

response = agent.invoke(
    [HumanMessage(content='Draft a launch plan.')],
    model='balanced',
)

Leave model unset (or 'auto') and the runtime chooses for you. The selection logic is intentionally internal for the broad tier path.

For scoped control, pass a ModelConfig:

from maivn import ModelChoice, ModelConfig

response = agent.invoke(
    [HumanMessage(content='Review this incident report and draft the response.')],
    model=ModelConfig(
        response=ModelChoice(model_id='provider/model-id'),
        thinking=ModelChoice(tier='max'),
        repl=ModelChoice(tier='fast'),
    ),
)

ModelConfig supports these parts:

The response and system-tool parts accept ModelChoice with one of the tiers auto, fast, balanced, max, or ultra, or an exact model_id. Planning uses PlanningChoice with auto, standard, advanced, or deep, or an exact model_id. Set exactly one of tier and model ID in each choice. Unspecified parts keep the runtime default.

To apply one exact model ID to all parts, including planning, use:

response = agent.invoke(
    [HumanMessage(content='Summarize the account.')],
    model=ModelConfig.for_all(model_id='provider/model-id', include_planning=True),
)

for_all leaves planning unset by default because planning has a separate tier vocabulary. include_planning=True requires an exact model ID.

#reasoning — how much deliberation to allow

response = agent.invoke(
    [HumanMessage(content='Reconcile these conflicting reports.')],
    reasoning='high',
)

reasoning ranges from 'minimal' through 'low', 'medium', to 'high'. Lower settings return faster; higher settings allow more deliberation for complex, multi-step reasoning. Like model, it is a hint. Leave it unset to use the runtime default.

#metadata — compatibility argument

Agent and Swarm accept metadata for source compatibility but do not forward or return it. Keep application correlation labels in your own tracing context. Do not place runtime controls in metadata; use supported typed arguments such as memory_config and system_tools_config.

#verbose — local tracing

verbose=True enables legacy terminal tracing for a call. For structured, filterable event reporting prefer the event builder:

response = agent.events().invoke([HumanMessage(content='Debug this')])

See events() for the include/exclude categories and the on_event callback.

#targeted_tools and structured_output

These parameters can be used together. Tool selection runs before the SDK adds or promotes a requested structured-output model as a final tool.

#targeted_tools — run only these tools

Pass a list of tool names to restrict the turn to those tools (plus their dependencies):

response = agent.invoke(
    [HumanMessage(content='Run diagnostics.')],
    targeted_tools=['fetch_data', 'analyze_data'],
)

#structured_output — validated typed output

Pass a Pydantic type to request that schema and validate the returned result. The builder and inline keyword use the same final-tool execution path, including any required tool work. Invalid output and failed runs still need error handling:

from pydantic import BaseModel, Field

class SentimentAnalysis(BaseModel):
    sentiment: str = Field(..., description='positive, negative, or neutral')
    confidence: float = Field(..., ge=0, le=1)

response = agent.structured_output(SentimentAnalysis).invoke(
    [HumanMessage(content='Analyze: "I love this product!"')]
)
print(response.result)  # a validated SentimentAnalysis instance

The structured_output= parameter on invoke() is the advanced, inline form of the same idea. For most code, prefer agent.structured_output(Model).invoke(...). See Structured Output for the full comparison with the final_tool pattern.

#Selection and final-tool rules

• targeted_tools selects registered tools and their declared tool dependencies. Empty or unknown selections raise ValueError.
• structured_output=Model then adds or promotes that model as a forced final constructor. It can be combined with targeted tool selection.
• Without a structured-output model, force_final_tool=True requires a registered final tool. When targeting tools, include the intended final tool in the selection.

#Running many at once

When you have a list of independent requests, batch() runs them concurrently and returns the responses in input order:

batch_inputs = [
    [HumanMessage(content='Summarize ticket A')],
    [HumanMessage(content='Summarize ticket B')],
    [HumanMessage(content='Summarize ticket C')],
]

responses = agent.batch(batch_inputs, max_concurrency=3)
for response in responses:
    print(response.response)

• Each input item is passed as the first argument to invoke(), so for an Agent each item is normally a Sequence[BaseMessage].
• max_concurrency caps how many run at once (must be >= 1). Omit it to use the default thread-pool worker count.
• Any keyword invoke() accepts can be passed once and is shared by every item (for example model='fast' or force_final_tool=True).
• Order is preserved even when calls finish out of order, and an exception in any one call is raised to the caller.

abatch() is the async variant with the same semantics:

responses = await agent.abatch(batch_inputs, max_concurrency=3)

Note Batch items are independent runs. Only share a thread_id across items when you intentionally want concurrent calls writing to the same conversation.

#Scheduling a run later

Use cron or interval for a durable remote trigger (the default), or pass mode='local' for the in-process compatibility scheduler. at is the local one-shot builder. A durable registration needs a key and deployed owner:

trigger = (
    agent.cron('0 9 * * MON-FRI', tz='America/New_York')
    .key('daily-ops-briefing')
    .invoke('Compose the daily ops briefing.')
)

The trigger inherits the agent's agent_id and version (by default its name and version 1). Use .target('agent-123', version=2) to select another deployed target explicitly.

See the Scheduled Invocation guide for the distinct remote trigger and local scheduler lifecycle APIs.

#Advanced: inspect compiled tools

compile_tools() resolves the current local tool metadata and model-visible schemas without executing a request:

tools = agent.compile_tools()
for tool in tools:
    print(tool.name, tool.input_schema)

This does not expose a full serialized invocation request or start a session.

#Next Steps

• Structured Output — .structured_output() and the final_tool pattern in depth
• Frontend Events — stream events to any frontend, plus the enrichment phase labels
• Scheduled Invocation — cron, interval, and one-shot runs with jitter and retry
• Memory and Recall — how threads and cross-thread assets shape what an agent remembers
• Multi-Agent — running sessions across a Swarm
• Agent API — full invoke / stream / batch reference
• Swarm API — the Swarm invocation surface