Use an agent - TrueFoundry Docs

Agent SDK is experimental and will have breaking changes!

This guide covers everything you need to run a saved agent: open a session, send turns, consume the event stream, and handle pauses, threads, and disconnects.

Prerequisite: an agent saved in the Registry. See Create an agent or the Agent Playground. For the conceptual overview and a full walkthrough, see SDK overview.

Install and connect

Running agents uses the gateway SDK, not the truefoundry client.

pip install -U truefoundry-gateway-sdk

For TypeScript, install truefoundry-gateway-sdk from npm and set the same environment variables. Every example below comes in both languages.

Create a session

A session is the conversation context for a saved agent — one customer working through one issue. It persists across many turns; persist session.id to resume later.

Open a session

A Session is the conversation context for a saved agent. Create one by name with create_session().

import os
from truefoundry_gateway_sdk import TrueFoundryGateway

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)

# A Session is the conversation context for a saved agent.
session = client.agents.create_session(agent_name="support-bot")

List sessions for an agent

List the existing Sessions for a saved agent with list_sessions(). It iterates newest-first, transparently paginating. Use it to resume an earlier conversation - each Session exposes its id (persist or reuse it to continue).

import os
from truefoundry_gateway_sdk import TrueFoundryGateway

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)

for session in client.agents.list_sessions(agent_name="support-bot"):
    print("session id:", session.id)
    print("created at:", session.created_at)

Create a turn

A turn is one request/response cycle. Send input, the agent runs until it finishes or pauses, then the stream closes. Create another turn on the same session to continue — turns chain automatically.

Create and stream a turn

Create a Turn in the session with create_turn(), then call stream() to start the Turn and stream its events as they arrive. The first event is always turn.created (carrying the turn_id); the stream closes with turn.done. Once the stream closes, the Turn is terminal. Call state() to read its final state. To continue the conversation, create another Turn on the same session. Turns within a session are chained automatically, so each Turn sees the history of the ones before it.

Creating a new Turn in a session automatically cancels any other Turn that is still running in that session.

import os
from truefoundry_gateway_sdk import TrueFoundryGateway, UserMessage

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

# Each Turn is one request/response cycle.
turn = session.create_turn(
    input=[UserMessage(content="I would like to file a support ticket.")]
)
for message in turn.stream():
    print(message)

# Once the stream closes the Turn is terminal. state() returns its final state.
turn_state = turn.state()
print("final status:", turn_state.status)

# Create another Turn on the same session. Turns are chained automatically,
# so this turn sees the history of the first one.
turn = session.create_turn(
    input=[UserMessage(content="Use my last order, ORD-2031.")]
)
for message in turn.stream():
    print(message)

List turns

List the Turns in a session with list_turns(). Each Turn exposes its input, and state() returns the current state of the run.

list_turns() always returns Turns in descending order (newest-first).

import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    TurnRunningState,
    TurnDoneState,
    TurnCancelledState,
    TurnErrorState,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)

session = client.agents.get_session(session_id="sess-abc123")

# list_turns() always returns Turns in descending order (newest-first).
for turn in session.list_turns():
    turn_state = turn.state()
    print("turn id:", turn.id)
    # input is the list that started the turn: UserMessage(s), or
    # UserToolApproval / UserToolResponse items.
    print("input:", turn.input)

    if isinstance(turn_state, TurnRunningState):
        print("status: running (no output yet)")
    elif isinstance(turn_state, TurnDoneState):
        print("status: done")
        # output is the final assistant message (model.message), or None.
        if turn_state.output is not None:
            print("output:", turn_state.output.content)
        # required_actions lists any pause events the Turn surfaced
        # (tool.approval_required, tool.response_required, mcp.auth_required).
        for item in turn_state.required_actions:
            print("required action:", item)
    elif isinstance(turn_state, TurnCancelledState):
        print("status: cancelled", turn_state.reason)
    elif isinstance(turn_state, TurnErrorState):
        print("status: error", turn_state.message)

Non-streaming turn

If you don’t need live events, skip stream() and call wait_for_completion() instead. It blocks until the Turn finishes.

import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    TurnDoneState,
    TurnCancelledState,
    TurnErrorState,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

turn = session.create_turn(
    input=[UserMessage(content="Summarize my last order.")],
)

turn_state = turn.wait_for_completion()
if isinstance(turn_state, TurnDoneState):
    # output and required_actions are mutually exclusive: if output is present,
    # required_actions is empty, and vice versa.
    # output is the final assistant message, or None if the Turn paused.
    if turn_state.output is not None:
        print(turn_state.output.content)
    # required_actions holds any pause events (tool.approval_required,
    # tool.response_required, mcp.auth_required).
    else:
        print("required actions:", turn_state.required_actions)
elif isinstance(turn_state, TurnCancelledState):
    print("cancelled:", turn_state.reason)
elif isinstance(turn_state, TurnErrorState):
    print("error:", turn_state.message)

print("completed at:", turn_state.completed_at)

Attach images or files to a turn

A UserMessage’s content can be a plain string or a list of content parts. Use content parts to send text alongside one or more file uploads (images, PDFs, and other documents). Each file is passed as a data URI of the form data:<mime>;base64,<payload>.

import base64
import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    TextContent,
    FileUploadContent,
    FileUpload,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

def to_data_uri(path: str, mime: str) -> str:
    with open(path, "rb") as f:
        return f"data:{mime};base64,{base64.b64encode(f.read()).decode()}"

# Attach both an image and a document in the same message.
image_uri = to_data_uri("invoice.png", "image/png")
doc_uri = to_data_uri("notes.txt", "text/plain")

turn = session.create_turn(
    input=[
        UserMessage(
            content=[
                TextContent(text="Does the invoice total match the notes?"),
                FileUploadContent(
                    file=FileUpload(filename="invoice.png", file_data=image_uri)
                ),
                FileUploadContent(
                    file=FileUpload(filename="notes.txt", file_data=doc_uri)
                ),
            ]
        )
    ]
)
for message in turn.stream():
    print(message)

Whether an attached file is understood depends on the agent’s model. Send images only to a vision-capable model; document handling (for example, PDFs) likewise depends on model and agent configuration.

Non-image files such as PDFs require the sandbox to be enabled on the agent - the harness uses it to process the document.

Cancel a turn

To stop the currently running Turn, call cancel() on its session. Cancelling aborts any in-flight model request, waits for running MCP tool calls to finish, and force-stops any sandbox the Turn provisioned. Cancellation is idempotent. You can continue by creating a new Turn, which chains on the cancelled Turn’s history.

import os
from truefoundry_gateway_sdk import TrueFoundryGateway, UserMessage

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

# Start a turn, then cancel the session partway through the stream.
turn = session.create_turn(
    input=[UserMessage(content="Run a long research task.")]
)
for i, message in enumerate(turn.stream()):
    print(message)
    if i == 5:
        # Don't break here. After cancel(), the backend closes the SSE stream
        # gracefully: it emits a terminal turn.done event and then ends the
        # stream, which closes this iterator on its own. Keep consuming until
        # the loop finishes so cancellation completes cleanly.
        session.cancel()

# A cancelled turn is terminal, so its event log is stored and can be replayed.
for event in turn.list_events(order="asc"):
    print(event)

# Continue the conversation: a new turn chains on the cancelled turn's history.
turn = session.create_turn(
    input=[UserMessage(content="Actually, just give me a quick summary instead.")]
)
for message in turn.stream():
    print(message)

Handle MCP outbound auth

When the agent needs to call a tool on an MCP server that requires a separate outbound authentication, it emits an mcp.auth_required event and the Turn ends. The event lists each server that needs authentication along with an auth_url. Depending on how the server is configured, this may be an OAuth flow or an API-key entry - see MCP authentication scenarios for details. Send the user to that URL to complete authentication, then resume by creating a new Turn.

When the previous Turn ended with mcp.auth_required, passing a UserMessage in the resuming Turn is not allowed.

import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    McpAuthRequiredEvent,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

# First turn: the agent needs an MCP server the user hasn't authorized yet.
# A Turn emits at most one mcp.auth_required event.
pending_auth = None
turn = session.create_turn(
    input=[UserMessage(content="Summarize my latest GitHub issues.")]
)
for message in turn.stream():
    print(message)
    if isinstance(message, McpAuthRequiredEvent):
        pending_auth = message

# Direct the user to each server's auth URL and wait for them to finish.
for server in pending_auth.servers:
    print(f"Authorize {server.mcp_server_name}: {server.auth_url}")
input("Press Enter once you have authorized the server(s)...")

# Second turn: resume - the agent continues the interrupted work.
turn = session.create_turn()
for message in turn.stream():
    print(message)

Handle tool approvals

When a tool call is configured to require human approval, the agent emits a tool.approval_required event and the Turn ends. Each event carries a thread_id and the tool_calls awaiting a decision - a single Turn can emit more than one (for example, when parallel threads each call a gated tool), so collect all of them. Resume by creating a new Turn with one UserToolApproval per pending tool call - allow it, or deny it with an optional reason. Each pending ToolCallRef carries the event_id of the model.message that emitted the tool call. Keep the same id-keyed event index you build while streaming, then look up events[event_id] to read the tool’s name and arguments - no separate bookkeeping required.

import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    TurnEvent,
    UserMessage,
    UserToolApproval,
    ApprovalAllow,
    ApprovalDeny,
    ToolApprovalRequiredEvent,
    is_event_delta,
    merge_event_delta,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

# First turn: the agent calls one or more tools that need approval.
events: dict[str, TurnEvent] = {}  # event id -> assembled event
pending_approvals = []             # a Turn can emit multiple tool.approval_required events

turn = session.create_turn(
    input=[UserMessage(content="Restart the billing service.")]
)
for message in turn.stream():
    if is_event_delta(message):
        merge_event_delta(events[message.id], message)
    else:
        events[message.id] = message
    if isinstance(message, ToolApprovalRequiredEvent):
        pending_approvals.append(message)

# Decide on each pending tool call, showing its name and arguments.
approvals = []
for pending in pending_approvals:
    for ref in pending.tool_calls:
        # Look up the model.message that emitted this tool call, then find the call.
        model_message = events[ref.event_id]
        tool_call = next(tc for tc in model_message.tool_calls if tc.id == ref.id)
        print(f"\napproval needed: {tool_call.tool_info.name}")
        print(f"  arguments: {tool_call.function.arguments}")
        answer = input("allow? [y/n] ").strip().lower()
        approvals.append(
            UserToolApproval(
                thread_id=pending.thread_id,
                tool_call_id=ref.id,
                approval=ApprovalAllow() if answer == "y" else ApprovalDeny(reason="denied by user"),
            )
        )

# Second turn: resume with the approval decisions.
turn = session.create_turn(input=approvals)
for message in turn.stream():
    print(message)

Answer agent questions

When the agent needs input it cannot safely assume, it can ask the user a structured question via the built-in client-side ask_user_question tool. Since the tool runs on your side, the agent emits a tool.response_required event and the Turn ends. The pending tool call’s arguments carry the question and its options. Collect the user’s answer and resume by creating a new Turn with one UserToolResponse per pending tool call. Each pending ToolCallRef carries the event_id of the model.message that emitted the tool call. Keep the same id-keyed event index you build while streaming, then look up events[event_id] to read the tool’s name and arguments - no separate bookkeeping required.

import json
import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    TurnEvent,
    UserMessage,
    UserToolResponse,
    ToolResponseRequiredEvent,
    TrueFoundrySystemToolInfo,
    is_event_delta,
    merge_event_delta,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

# First turn: the agent asks the user a question.
events: dict[str, TurnEvent] = {}  # event id -> assembled event
pending_questions = []             # a Turn can emit multiple tool.response_required events

turn = session.create_turn(
    input=[UserMessage(content="Create a new environment called testing-1.")]
)
for message in turn.stream():
    if is_event_delta(message):
        merge_event_delta(events[message.id], message)
    else:
        events[message.id] = message
    if isinstance(message, ToolResponseRequiredEvent):
        pending_questions.append(message)

# Answer each pending question, reading the prompt and options from arguments.
responses = []
for pending in pending_questions:
    for ref in pending.tool_calls:
        # Look up the model.message that emitted this tool call, then find the call.
        model_message = events[ref.event_id]
        tool_call = next(tc for tc in model_message.tool_calls if tc.id == ref.id)
        # tool.response_required covers any client-side tool; handle ask_user_question here.
        if not (
            isinstance(tool_call.tool_info, TrueFoundrySystemToolInfo)
            and tool_call.tool_info.name == "ask_user_question"
        ):
            continue
        args = json.loads(tool_call.function.arguments or "{}")
        print(f"\n{args.get('question', '')}")
        for idx, option in enumerate(args.get("options") or [], 1):
            print(f"  {idx}. {option}")
        answer = input("your answer: ").strip()
        responses.append(
            UserToolResponse(
                thread_id=pending.thread_id,
                tool_call_id=ref.id,
                # content is a free-form string - the chosen option, or any text.
                content=answer,
            )
        )

# Second turn: resume with the user's answers.
turn = session.create_turn(input=responses)
for message in turn.stream():
    print(message)

Every turn emits a stream of events over SSE. The stream opens with turn.created and closes with turn.done. See the Turn events reference for every event type.

Handling Event Delta while streaming

Most events in a Turn are complete on their own - a single payload you can use directly. Some updates are streamed instead: the base event arrives first, followed by a series of Event Deltas - incremental fragments that you merge into the base. All deltas for one update share the base event’s id, while sequence_number increases across the base and all its deltas.

// Base assembled event arrives first
{ "type": "model.message", "id": "0f3a9c2b-7d41-4e8a-b2c6-1a5f9e3d2b48", "sequence_number": 6, "content": "" }

// Deltas follow: same id, increasing sequence_number, each merged into the base
{ "type": "model.message.delta", "id": "0f3a9c2b-7d41-4e8a-b2c6-1a5f9e3d2b48", "sequence_number": 7, "content": "Hel" }
{ "type": "model.message.delta", "id": "0f3a9c2b-7d41-4e8a-b2c6-1a5f9e3d2b48", "sequence_number": 8, "content": "lo" }
{ "type": "model.message.delta", "id": "0f3a9c2b-7d41-4e8a-b2c6-1a5f9e3d2b48", "sequence_number": 9, "content": "!", "finish_reason": "stop" }

// After merging every delta, the base holds the full message
{ "type": "model.message", "id": "0f3a9c2b-7d41-4e8a-b2c6-1a5f9e3d2b48", "content": "Hello!", "finish_reason": "stop" }

Because every delta carries the base event’s id, keep an id-keyed index of assembled events: store each non-delta event under its id, and merge each delta into the base with the same id. The id is unique per message, so deltas from concurrently streaming threads (the main agent and any sub-agents) always merge into the right base.

Event Deltas appear only while streaming. When you list events via the events API, the deltas are already merged into a single assembled event.

The most common example is the assistant’s model output: a base ModelMessageEvent followed by ModelMessageEventDelta deltas that carry incremental text and tool-call chunks. Merge the deltas into the base as they arrive - read the base’s growing content for a live typing effect.

import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    TurnEvent,
    UserMessage,
    is_event_delta,
    merge_event_delta,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

events: dict[str, TurnEvent] = {}  # event id -> assembled event

turn = session.create_turn(
    input=[UserMessage(content="Summarize my last order.")]
)
for message in turn.stream():
    if is_event_delta(message):
        base = events[message.id]
        merge_event_delta(base, message)
    else:
        events[message.id] = message

Resume streaming

When you lose the original create_turn() stream (for example, after a page reload), fetch the Turn again with get_turn() using its turn_id and check its state. If it is still running, reconnect to its live event stream with stream() and keep merging; if it has already finished, rebuild the index from its stored event log with list_events() instead. When resuming a running Turn, pass after_sequence_number to continue after a known point (defaults to 0, replaying from the first event); the SDK also resumes transparently from the last delivered event on any mid-stream connection drop, and closes when the Turn reaches a terminal state. Either way, merge into the same id-keyed index of assembled events, so base events and their deltas keep merging seamlessly across the reconnect.

from truefoundry_gateway_sdk import TurnRunningState, is_event_delta, merge_event_delta

# `session_id`, `turn_id`, `events`, and `last_sequence_number` already exist from the original stream:
#   session_id: str               - the id of the Session
#   turn_id: str                  - the id of the Turn you were streaming
#   events: dict[str, TurnEvent]  - the id-keyed index of assembled events
#   last_sequence_number: int     - the sequence_number of the last event you processed
# Persist all four (for example, in your session store) so you can resume after a restart.

session = client.agents.get_session(session_id=session_id)
turn = session.get_turn(turn_id=turn_id)

if isinstance(turn.state(), TurnRunningState):
    # Still running: reconnect to the live stream. after_sequence_number skips events
    # you've already processed; the SDK also auto-resumes from the last delivered
    # event on any mid-stream connection drop.
    for message in turn.stream(after_sequence_number=last_sequence_number):
        last_sequence_number = message.sequence_number
        if is_event_delta(message):
            merge_event_delta(events[message.id], message)
        else:
            events[message.id] = message
else:
    # Already finished: rebuild the index from the stored event log, which is
    # authoritative and overwrites any partial state from the lost stream.
    # list_events() returns already-merged events, so there are no deltas to merge here.
    for event in turn.list_events():
        events[event.id] = event

Handle threads

A single Turn stream interleaves events from the root agent and any sub-agents that run in parallel. Every event carries a thread_id:

"main" — the root agent.
A unique ID — a sub-agent thread. thread.created and thread.done mark sub-agent thread lifecycle; they are not emitted for the main thread.
null — a turn-level event, not tied to any thread: turn.created and turn.done (first and last on the stream), sandbox.created, and mcp.auth_required.

See Turn Events for the full reference.

import os
from collections import defaultdict
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    TurnEvent,
    UserMessage,
    ModelMessageEvent,
    ModelMessageEventDelta,
    ThreadCreatedEvent,
    ThreadDoneEvent,
    TrueFoundrySystemToolInfo,
    is_event_delta,
    merge_event_delta,
)

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)
session = client.agents.create_session(agent_name="support-bot")

threads: dict[str, dict[str, TurnEvent]] = defaultdict(dict)

turn = session.create_turn(
    input=[UserMessage(content="Summarize my last order.")]
)
for message in turn.stream():
    if message.thread_id is None:
        print("turn level event:", message)
        continue

    if isinstance(message, (ModelMessageEvent, ModelMessageEventDelta)):
        for tool_call in message.tool_calls or []:
            if (
                isinstance(tool_call.tool_info, TrueFoundrySystemToolInfo)
                and tool_call.tool_info.name == "create_sub_agent"
            ):
                print("sub agent is getting initialized")
    if isinstance(message, ThreadCreatedEvent):
        print(f"{message.thread_id} created: {message.title}")

    if is_event_delta(message):
        merge_event_delta(threads[message.thread_id][message.id], message)
    else:
        threads[message.thread_id][message.id] = message

    if isinstance(message, ThreadDoneEvent):
        events_for_thread = threads.pop(message.thread_id)
        print(f"{message.thread_id} done: {message.title} ({len(events_for_thread)} events)")

Listing Events

Fetch the full event log of a finished Turn with list_events(). It yields the Turn’s events in order, auto-paginating. Pass order="asc" (default, oldest-first) or order="desc" to control the direction. Unlike stream(), list_events() returns already-merged events: each model message arrives as a single assembled ModelMessageEvent, never as a base plus deltas. There is nothing to merge - you can use each event directly. For example, a model message that stream() delivers as a base event followed by deltas:

// On stream(): a base event plus its deltas, all sharing one id
{ "type": "model.message",       "id": "0f3a...", "content": "" }
{ "type": "model.message.delta", "id": "0f3a...", "content": "Hel" }
{ "type": "model.message.delta", "id": "0f3a...", "content": "lo!", "finish_reason": "stop" }

is returned by list_events() as one fully-merged event:

// On list_events(): the same message, already assembled
{ "type": "model.message", "id": "0f3a...", "content": "Hello!", "finish_reason": "stop" }

list_events() is only available for Turns that have completed. A running Turn has no stored event log yet - use stream() for live delivery instead.

import os
from truefoundry_gateway_sdk import TrueFoundryGateway, TurnRunningState

client = TrueFoundryGateway(
    base_url=os.environ["GATEWAY_BASE_URL"],
    api_key=os.environ["TFY_API_KEY"],
)

session = client.agents.get_session(session_id="sess-abc123")

# Find the latest completed turn (newest-first).
for turn in session.list_turns():
    # list_events() is only available once the turn has completed.
    if isinstance(turn.state(), TurnRunningState):
        continue
    # order="asc" (default) yields oldest-first; use "desc" for newest-first.
    for event in turn.list_events(order="asc"):
        print(event)
    break

Complete example

The Complete example is a runnable terminal chat client that implements every pattern in this guide — streaming, delta merging, approvals, questions, MCP auth, sub-agent threads, and multi-turn chaining.

​Install and connect

​Create a session

​Open a session

​List sessions for an agent

​Create a turn

​Create and stream a turn

​List turns

​Non-streaming turn

​Attach images or files to a turn

​Cancel a turn

​Handle MCP outbound auth

​Handle tool approvals

​Answer agent questions

​Subscribe to events

​Handling Event Delta while streaming

​Resume streaming

​Handle threads

​Listing Events

​Complete example

Install and connect

Create a session

Open a session

List sessions for an agent

Create a turn

Create and stream a turn

List turns

Non-streaming turn

Attach images or files to a turn

Cancel a turn

Handle MCP outbound auth

Handle tool approvals

Answer agent questions

Subscribe to events

Handling Event Delta while streaming

Resume streaming

Handle threads

Listing Events

Complete example