Sessions

A session runs many turns and remembers them. agent.run(...) is a single question with no memory; a session is a conversation.

It is also a lower-level API. A turn is not a call that returns an answer - it is a stream of events you consume, so you can react while the agent is still working rather than only when it finishes.

A session can run:

• in this process. Call create_session() on an Agent instance you built here. No worker, no setup.
• against a running agent, from agents serve or agents deploy. Call create_session() on Agent.load_from(agent_name=...), and the session is durable.

Note that in-process sessions have some limitations regarding the Session Protocol :

import asyncio

from mistralai.agents import agents
from mistralai.vibe.harness.app_server.session import (
    HistoryEntryCompletedEvent,
    MessageHistoryEntry,
    SessionClient,
    TextContentBlock,
)


async def ask(session: SessionClient, text: str) -> str:
    reply = ""
    async for event in session.run_turn([TextContentBlock(text=text)]):
        if isinstance(event.payload, HistoryEntryCompletedEvent):
            entry = event.payload.entry
            if isinstance(entry, MessageHistoryEntry) and entry.role == "assistant":
                reply = "".join(b.text for b in entry.content)
    return reply

async def main():
    agent = agents.Agent.load_from(agent_name="hello-agent")
    async with await agent.create_session() as session:
        await ask(session, "Remember the number 7.")
        print(await ask(session, "What number did I say? Digits only."))   # -> 7

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

import asyncio

from mistralai.agents import agents
from mistralai.vibe.harness.app_server.session import (
    HistoryEntryCompletedEvent,
    MessageHistoryEntry,
    SessionClient,
    TextContentBlock,
)


async def ask(session: SessionClient, text: str) -> str:
    reply = ""
    async for event in session.run_turn([TextContentBlock(text=text)]):
        if isinstance(event.payload, HistoryEntryCompletedEvent):
            entry = event.payload.entry
            if isinstance(entry, MessageHistoryEntry) and entry.role == "assistant":
                reply = "".join(b.text for b in entry.content)
    return reply

async def main():
    agent = agents.Agent.load_from(agent_name="hello-agent")
    async with await agent.create_session() as session:
        await ask(session, "Remember the number 7.")
        print(await ask(session, "What number did I say? Digits only."))   # -> 7

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

agent_name is the name you gave the agent, the first one agents serve prints, not the task queue it says the agent is serving on. The control plane routes the agent name to its worker. You can also pass agent_id= with the agent's ID.

To continue a session later, from any process, pass its ID. You do not need the agent's code or name:

session = await agents.Agent.load_session(session_id)

session = await agents.Agent.load_session(session_id)

Events are typed: event.payload is a union of HistoryEntryAddedEvent, HistoryEntryPatchedEvent, HistoryEntryCompletedEvent, SessionUpdatedEvent and friends, so you match on the class rather than digging through a dict. The entry and its content blocks are objects too, which is why the reply is b.text and not b["text"].

A session opened against a served or deployed agent speaks the canonical Session v1 stream by default - session_protocol=SessionProtocol.SESSION_V1 is the default, so you do not pass it. Pass session_protocol=SessionProtocol.LEGACY_TASK_PROTOCOL only if you need the legacy task protocol.

It is fixed for the lifetime of the session, and it applies only to a session opened against a served or deployed agent. Passing it to an in-process agent raises:

SdkError: session_protocol applies only to a deployment-backed Session

SdkError: session_protocol applies only to a deployment-backed Session

A v1 stream serves history entries, turns, session status and callbacks. The event union also declares notice and error payloads, but v1 never emits them, so do not build on them: how a turn ended is on TurnTerminatedEvent, as turn.termination.

agent.create_session(
    *,
    session_protocol=SessionProtocol.SESSION_V1,  # the default; pass LEGACY_TASK_PROTOCOL to opt out
    history=None,              # seed the conversation with existing entries
    agent=None,                 # optional Agent customization for this Session
    session_inputs=None,       # caller-supplied values for tools and hooks, keyed by input ID
)

agent.create_session(
    *,
    session_protocol=SessionProtocol.SESSION_V1,  # the default; pass LEGACY_TASK_PROTOCOL to opt out
    history=None,              # seed the conversation with existing entries
    agent=None,                 # optional Agent customization for this Session
    session_inputs=None,       # caller-supplied values for tools and hooks, keyed by input ID
)

Use agent to append instructions or override the model and reasoning effort for one Session. The saved Agent and its deployment are unchanged, and the override stays scoped to this Session: the next Session you create on the same Agent starts from its saved config. The same customization is used after a remote Session resumes:

session = await agent.create_session(
    agent={
        "additional_instructions": "Keep changes small.",
        "model": "mistral-large-latest",
        "reasoning_effort": "high",
    },
)

session = await agent.create_session(
    agent={
        "additional_instructions": "Keep changes small.",
        "model": "mistral-large-latest",
        "reasoning_effort": "high",
    },
)

The addition follows the Agent's base instructions, separated by one blank line. An empty string adds nothing. The addition can be at most 8,000 characters. A deployment-backed Session continues to select the saved Agent by name.

model and reasoning_effort are optional. Reasoning accepts none, minimal, low, medium, high, or xhigh. A custom completion provider cannot use these generic overrides; configure that provider directly instead.

permission_mode and response_format exist on create_session and apply to an in-process session only. A deployment-backed session refuses both and raises UnsupportedSessionProcedureError: it uses the permission mode and completion config pinned in its deployed agent.

The session is an async context manager, and closing it is not optional: an open session keeps an event-reading task and its HTTP connections alive in your process. async with closes it when the block ends, including when a turn raises:

async with await agent.create_session() as session:
    ...  # your turns here

async with await agent.create_session() as session:
    ...  # your turns here

create_session() is awaited because opening the session is a network round-trip.

Close a session that outlives the block, one per chat in a bot for instance, where you replace or drop it:

previous = sessions.pop(chat_id, None)
if previous is not None:
    await previous.close()
sessions[chat_id] = await agent.create_session()

previous = sessions.pop(chat_id, None)
if previous is not None:
    await previous.close()
sessions[chat_id] = await agent.create_session()

The Telegram weather bot does this in its /new handler, and closes the sessions still open in a post_shutdown hook.

A Session input is a value you give one Session, such as a repository or a revision, for your tools and hooks to use. It is fixed for the Session's lifetime and delivered only to the tools and hooks that declare it. It is not added to the model's context or to a tool's schema.

Subclass SessionInput with a stable input_id, and declare it on a tool with Depends:

from typing import Annotated

from pydantic import BaseModel

from mistralai.agents import agents


class RepositoryContext(agents.SessionInput):
    input_id = "acme.repository"

    provider: str
    repository: str
    revision: str


class CreatePrArgs(BaseModel):
    title: str


@agents.tool(
    name="create_pr",
    description="Open a pull request.",
    input_schema=CreatePrArgs,
)
async def create_pr(
    args: CreatePrArgs,
    repo: Annotated[RepositoryContext, agents.Depends(RepositoryContext)],
) -> str:
    return f"Opened {args.title!r} on {repo.provider}"

from typing import Annotated

from pydantic import BaseModel

from mistralai.agents import agents


class RepositoryContext(agents.SessionInput):
    input_id = "acme.repository"

    provider: str
    repository: str
    revision: str


class CreatePrArgs(BaseModel):
    title: str


@agents.tool(
    name="create_pr",
    description="Open a pull request.",
    input_schema=CreatePrArgs,
)
async def create_pr(
    args: CreatePrArgs,
    repo: Annotated[RepositoryContext, agents.Depends(RepositoryContext)],
) -> str:
    return f"Opened {args.title!r} on {repo.provider}"

The model never sees the repo parameter, and the function receives a RepositoryContext. A hook declares an input the same way.

Pass the value once, keyed by its input_id, on an in-process, served or deployed agent:

async with await agent.create_session(
    session_inputs={
        RepositoryContext.input_id: RepositoryContext(
            provider="github", repository="acme/api", revision="main"
        ),
    },
) as session:
    ...

async with await agent.create_session(
    session_inputs={
        RepositoryContext.input_id: RepositoryContext(
            provider="github", repository="acme/api", revision="main"
        ),
    },
) as session:
    ...

run_turn(...) yields a ClientEvent per step. Every event carries the same envelope - event_id, emitted_at, session_id, root_session_id - and the part you match on is payload. Those names are camelCase on the wire and snake_case on the Python object, so it is event.event_id, not event.eventId.

Against a served or deployed agent on Session v1, one turn of a plain question walks the entry and turn lifecycle in this order:

Wait for a HistoryEntryCompletedEvent whose entry is a MessageHistoryEntry with role == "assistant" to get a finished answer. Follow HistoryEntryPatchedEvent instead when you want to render the reply as it is written.

A turn emits more entries than that as soon as the agent uses a tool: the call and the result each get their own entry, carrying a ToolCallHistoryEntry and a ToolResultHistoryEntry. Neither has a role, which is why the example narrows on MessageHistoryEntry before reading one. Match on the events you handle and ignore the rest; new ones are added over time.

steer_turn and interrupt_turn take the id of the turn you expect to be running. start_turn returns a TurnStartResponse, so pass its response.turn.id, not the response itself. A steered turn finishes its original work and the steer, then terminates normally; an interrupted one terminates with a CancelledTermination and the next start_turn runs as usual. Naming a turn that has already ended raises RemoteSessionConflictError.

History records what the runtime executed, not what your middleware changed on the way: see Custom hooks for what a hook's rewrites and refusals leave behind.

The terminal running agents serve may print:

RuntimeWarning: coroutine 'AsyncClient.aclose' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

RuntimeWarning: coroutine 'AsyncClient.aclose' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

This is the mistralai client, not your session. Mistral.__init__ registers a weakref.finalize that closes the client's HTTP connection when the object is garbage-collected, and that close needs a running event loop; when none is left to run it, the coroutine is never awaited and Python warns. The second line is Python's standard hint: re-run with python -X tracemalloc=5 for the allocation traceback.

SessionClient registers no finalizer, so a session you drop without closing leaks without printing anything.

async with Mistral(...) does not silence the warning: __aexit__ does not detach the finalizer that __init__ registered. Passing your own client does, because the finalizer skips a client it did not create. You then own its settings too, so set follow_redirects=True, which is what the client Mistral builds for itself uses:

import os

import httpx
from mistralai.client import Mistral


async def call_mistral() -> None:
    http = httpx.AsyncClient(follow_redirects=True)
    async with http, Mistral(api_key=os.environ["MISTRAL_API_KEY"], async_client=http) as client:
        ...

import os

import httpx
from mistralai.client import Mistral


async def call_mistral() -> None:
    http = httpx.AsyncClient(follow_redirects=True)
    async with http, Mistral(api_key=os.environ["MISTRAL_API_KEY"], async_client=http) as client:
        ...

• The Agent - agent.run(...), the one-shot alternative.
• Run modes - agents serve and agents deploy, which give you an agent to attach to.
• Tools - what adds entries to a turn.
• Custom hooks - code that runs inside a turn, around every tool call.