Using Agents - TrueFoundry Docs

Coming soon.

Agent Platform is currently in beta. The API may introduce breaking changes.

Overview

This page covers running an agent that is already saved in the Agent Registry. To create or configure an agent, see Agent. The Agent Harness ships with a Python and TypeScript SDK for interacting with agents and consuming the event stream. We also provide a React component for building a chat interface for an agent.

How interaction works

Interaction is organized as a hierarchy: Session → Turn → Events, with threads running underneath.

Session — a long-running agent process. The session holds the agent configuration and persists across many Turns.
Thread — an execution context inside the session. The main thread is the root agent, which interacts with the user and acts as a coordinator for other agents. It can spin up additional threads, each running a sub-agent. Threads are long-lived - they can live across Turns.
Turn — a single user interaction with the session, and the request boundary. Turns within a session are chained, so each Turn builds on the history of the ones before it. Only one Turn can be running in a session at any point in time. A Turn runs until the agent finishes, or until it hits the iteration limit or timeout configured for the agent. A Turn can be cancelled while it is running.
Events — a Turn streams events as the agent works. Each event identifies the thread it came from. A Turn begins with a creation event and ends with a terminal event.

Creating a Session

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

Python

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

Listing 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).

Python

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)

Interacting with a Session by creating a Turn

Create a Turn in the session with create_turn(). It streams the Turn’s events as they arrive. The first event is always turn.created (carrying the turn_id); the stream closes on a terminal event. 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.

Python

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.
for event in session.create_turn(
    input=[UserMessage(content="I would like to file a support ticket.")]
):
    print(event)

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

Non-streaming Turn

If you don’t need live events, pass stream=False. Instead of an event iterator, create_turn() returns a Turn directly once the Turn is created server-side. Block until it finishes with wait(), which returns the terminal state.

Python

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

# stream=False returns a Turn directly instead of an event iterator.
turn = session.create_turn(
    input=[UserMessage(content="Summarize my last order.")],
    stream=False,
)

# Block until the Turn reaches a terminal state.
state = turn.wait()
print(state)

Attaching 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>.

Python

import base64
import os
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    TextContentPart,
    FileUploadContentPart,
    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")

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

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.

Listing Turns

List the Turns in a session with list_turns(). It defaults to order="desc" (newest-first); pass order="asc" for oldest-first. Each Turn exposes its input, and state() returns the current status (and output once the Turn is terminal).

Python

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() defaults to order="desc" (newest-first); use "asc" for oldest-first.
for turn in session.list_turns():
    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(state, TurnRunningState):
        print("status: running (no output yet)")
    elif isinstance(state, TurnDoneState):
        print("status: done")
        # output is a list of TurnOutputEvent items: accumulated model.message
        # events (the assistant's final text) plus any terminal events
        # (tool.approval_required, tool.response_required, mcp.auth_required).
        for item in state.output:
            print("output:", item)
    elif isinstance(state, TurnCancelledState):
        print("status: cancelled", state.cancellation_reason)
    elif isinstance(state, TurnErrorState):
        print("status: error", state.error)

Cancelling 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. The cancelled Turn is terminal, so its event log is stored and can be replayed - and you can continue the conversation by creating a new Turn, which chains on the cancelled Turn’s history.

Python

import os
from truefoundry_gateway_sdk import TrueFoundryGateway, TurnCreated, 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_id = None
for i, event in enumerate(session.create_turn(
    input=[UserMessage(content="Run a long research task.")]
)):
    print(event)
    # The first event is always turn.created - save the id so we can fetch
    # the turn's stored events after cancellation.
    if isinstance(event, TurnCreated):
        turn_id = event.turn_id
    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.
turn = session.get_turn(turn_id=turn_id)
for event in turn.events(order="asc"):
    print(event)

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

Streaming a running Turn’s events

Reconnect to a running Turn’s live event stream with stream(). This is useful when you have lost the original create_turn() stream (for example, after a page reload). It resumes transparently from the last delivered event on any connection drop and closes when the Turn reaches a terminal state.

stream() works only while the Turn is running. For a Turn that has already finished, use events() instead (see below) - SSE events are not stored, so there is nothing to stream once a Turn is terminal.

Python

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 running turn (a session has at most one).
for turn in session.list_turns():
    # stream() is only available while the turn is running.
    if isinstance(turn.state(), TurnRunningState):
        # Every event carries a monotonically increasing sequence_id within the turn.
        # Track the last one so you can resume from it after a disconnect via
        # stream(after_sequence_id=last_sequence_id).
        last_sequence_id = 0
        for event in turn.stream(after_sequence_id=last_sequence_id):
            last_sequence_id = event.sequence_id
            print(event.sequence_id, event)
        break

Getting events for a completed Turn

Fetch the full event log of a finished Turn with events(). It yields the Turn’s events in order, auto-paginating. Pass order="asc" (default, oldest-first) or order="desc" to control the direction.

events() is only available for Turns that are not running. A running Turn has no stored event log yet - use stream() for live delivery instead. turn.created and turn.done are SSE-only and do not appear in events().

Python

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():
    # events() is only available once the turn is terminal (not running).
    if isinstance(turn.state(), TurnRunningState):
        continue
    # order="asc" (default) yields oldest-first; use "desc" for newest-first.
    for event in turn.events(order="asc"):
        print(event)
    break

Handling MCP Outbound Auth

When the agent needs to call a tool on an MCP server that requires a separate outbound authentication, it emits a terminal 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.

Python

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

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.
pending_auth = None
for event in session.create_turn(
    input=[UserMessage(content="Summarize my latest GitHub issues.")]
):
    print(event)
    if isinstance(event, McpAuthRequired):
        pending_auth = event

# 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.
for event in session.create_turn():
    print(event)

Merging the model message stream

The assistant’s response arrives as a sequence of ModelMessageDelta events - streaming deltas that carry incremental text and tool-call chunks. Render them as they arrive for a live typing effect, or, if you only need the finished message, merge the deltas into a single complete message. The SDK ships a ModelMessageBuilder utility for this. Feed it each ModelMessageDelta; when a delta arrives with a non-null finish_reason, the message is complete and build_and_reset() returns the assembled ModelMessage (concatenated content and fully reconstructed tool_calls). Group by thread_id, since the main agent and any sub-agents stream concurrently.

Python

import os
from collections import defaultdict
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    ModelMessageDelta,
    ModelMessageBuilder,
)

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

# One builder per thread - the main agent and sub-agents stream concurrently.
builders = defaultdict(ModelMessageBuilder)

for event in session.create_turn(
    input=[UserMessage(content="Summarize my last order.")]
):
    if isinstance(event, ModelMessageDelta):
        builder = builders[event.thread_id]
        builder.add(event)
        # A non-null finish_reason marks the end of one complete message.
        if event.finish_reason is not None:
            model_message = builder.build_and_reset()
            print(model_message)

Handling tool approvals

When a tool call is configured to require human approval, the agent emits a terminal 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.

Python

import os
from collections import defaultdict
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    UserToolApproval,
    ApprovalAllow,
    ApprovalDeny,
    ModelMessageDelta,
    ModelMessageBuilder,
    ToolResponse,
    ToolApprovalRequired,
)

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.
open_tool_calls = {}                         # (thread_id, tool_call_id) -> assembled ToolCall
builders = defaultdict(ModelMessageBuilder)  # thread_id -> ModelMessageBuilder
pending_approvals = []                       # a Turn can emit multiple tool.approval_required events

for event in session.create_turn(
    input=[UserMessage(content="Restart the billing service.")]
):
    print(event)
    if isinstance(event, ModelMessageDelta):
        # model.message arrives as streaming deltas; assemble them per thread.
        builder = builders[event.thread_id]
        builder.add(event)
        if event.finish_reason is not None:
            # A tool call is opened once the model finishes emitting it.
            tool_calls = builder.build_and_reset().tool_calls or []
            for tool_call in tool_calls:
                open_tool_calls[(event.thread_id, tool_call.id)] = tool_call
    elif isinstance(event, ToolResponse):
        # The tool call is closed once its result comes back.
        open_tool_calls.pop((event.thread_id, event.tool_call_id), None)
    elif isinstance(event, ToolApprovalRequired):
        pending_approvals.append(event)

# Decide on each pending tool call, showing its name and arguments.
approvals = []
for pending in pending_approvals:
    for tool_call in pending.tool_calls:
        open_call = open_tool_calls[(pending.thread_id, tool_call.id)]
        tool_name = open_call.tool_info.original_tool_name
        print(f"\napproval needed: {tool_name}")
        print(f"  arguments: {open_call.function.arguments}")
        answer = input("allow? [y/n] ").strip().lower()
        approvals.append(
            UserToolApproval(
                thread_id=pending.thread_id,
                tool_call_id=tool_call.id,
                approval=ApprovalAllow() if answer == "y" else ApprovalDeny(reason="denied by user"),
            )
        )

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

Answering agent’s 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 terminal 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.

Python

import json
import os
from collections import defaultdict
from truefoundry_gateway_sdk import (
    TrueFoundryGateway,
    UserMessage,
    UserToolResponse,
    ModelMessageDelta,
    ModelMessageBuilder,
    ToolResponse,
    ToolResponseRequired,
)

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.
open_tool_calls = {}                         # (thread_id, tool_call_id) -> assembled ToolCall
builders = defaultdict(ModelMessageBuilder)  # thread_id -> ModelMessageBuilder
pending_questions = []                       # a Turn can emit multiple tool.response_required events

for event in session.create_turn(
    input=[UserMessage(content="Create a new environment called testing-1.")]
):
    print(event)
    if isinstance(event, ModelMessageDelta):
        builder = builders[event.thread_id]
        builder.add(event)
        if event.finish_reason is not None:
            # A tool call is opened once the model finishes emitting it.
            tool_calls = builder.build_and_reset().tool_calls or []
            for tool_call in tool_calls:
                open_tool_calls[(event.thread_id, tool_call.id)] = tool_call
    elif isinstance(event, ToolResponse):
        # The tool call is closed once its result comes back.
        open_tool_calls.pop((event.thread_id, event.tool_call_id), None)
    elif isinstance(event, ToolResponseRequired):
        pending_questions.append(event)

# Answer each pending question, reading the prompt and options from arguments.
responses = []
for pending in pending_questions:
    for tool_call in pending.tool_calls:
        open_call = open_tool_calls[(pending.thread_id, tool_call.id)]
        tool_name = open_call.tool_info.original_tool_name
        # tool.response_required covers any client-side tool; handle ask_user_question here.
        if tool_name != "ask_user_question":
            continue
        args = json.loads(open_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=tool_call.id,
                # content is a free-form string - the chosen option, or any text.
                content=answer,
            )
        )

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

Turn input

Each Turn’s input is a list of one of these types. Resuming a Turn paused by mcp.auth_required needs no input - omit input or pass [].

User messages (UserMessage) cannot be mixed with tool approvals or client-side tool responses in the same input list. UserToolApproval and UserToolResponse may be mixed together.

UserMessage

Start a new conversation or send the next user message. content is either a plain string or a list of content parts, letting you attach files alongside text.

{ "type": "user.message", "content": "I would like to file a support ticket." }

{
    "type": "user.message",
    "content": [
        { "type": "text", "text": "Please review this document." },
        {
            "type": "file_upload",
            "file": {
                "filename": "report.pdf",
                "file_data": "data:application/pdf;base64,JVBERi0xLjQK..."
            }
        }
    ]
}

Field	Type	Required	Description
`type`	`"user.message"`	Yes
`content`	string \| `UserContentPart[]`	Yes	The message text, or a list of content parts (text and file uploads).

UserContentPart

A content part is one of: Text

Field	Type	Required	Description
`type`	`"text"`	Yes
`text`	string	Yes	The message text.

File upload

Field	Type	Required	Description
`type`	`"file_upload"`	Yes
`file.filename`	string	Yes	Name of the uploaded file.
`file.file_data`	string	Yes	Data URI: `data:<mime>;base64,<payload>`.

UserToolApproval

Sent to resume a Turn paused by tool.approval_required. One item per pending tool call.

{
    "type": "user.tool_approval",
    "thread_id": "main",
    "tool_call_id": "call_restart_billing",
    "approval": { "status": "allow" }
}

Field	Type	Required	Description
`type`	`"user.tool_approval"`	Yes
`thread_id`	string	Yes	`thread_id` from the `tool.approval_required` event.
`tool_call_id`	string	Yes	ID of the tool call being approved or denied.
`approval`	`ApprovalAllow` \| `ApprovalDeny`	Yes	Use `{"status": "allow"}` to permit the call, or `{"status": "deny", "reason": "..."}` to block it. `reason` is optional.

UserToolResponse

Sent to resume a Turn paused by tool.response_required. One item per pending tool call.

{
    "type": "user.tool_response",
    "thread_id": "main",
    "tool_call_id": "call_a1b2",
    "content": "tfy-prod-us (production)"
}

Field	Type	Required	Description
`type`	`"user.tool_response"`	Yes
`thread_id`	string	Yes	`thread_id` from the `tool.response_required` event.
`tool_call_id`	string	Yes	ID of the tool call whose result is being supplied.
`content`	string	Yes	The result to return to the agent.

Turn Events

create_turn() streams Server-Sent Events (SSE). Each event is a JSON object with a type field.

The stream always opens with turn.created and closes after a terminal event.
Events tied to a thread carry a thread_id. The root agent’s thread is always "main"; dynamically spawned sub-agents have unique thread IDs.
thread.created and thread.done track sub-agent threads only - they are not emitted for the main thread, which is created automatically on the first Turn and lives for the session’s lifetime. The overall Turn’s completion is signalled by turn.done.
All events carry a sequence_id, monotonically increasing within a Turn.
turn.created and turn.done appear only in the SSE stream; they are not stored in the Turn body’s events list.

turn.*

`type`	Description
`turn.created`	First event on the stream. Carries `turn_id`.
`turn.done`	Last event on every stream. Carries the final `status`.

model.* - Model output

`type`	Description
`model.message`	LLM output - assistant text and/or tool call chunks. Streamed as `ModelMessageDelta`; stored in the Turn body as the assembled `ModelMessage`. Most frequent event.

tool.* - Tool calls and results

`type`	Description
`tool.response`	Complete server-side tool result.
`tool.approval_required`	A tool call is awaiting human approval. Resume with an approval.
`tool.response_required`	A client-side tool was called and needs a result. Resume with the tool response.

thread.* - Sub-agent thread lifecycle

`type`	Description
`thread.created`	A sub-agent thread started. Not emitted for the main thread.
`thread.done`	A sub-agent thread completed (`done`) or failed (`error`). Not emitted for the main thread; does not close the stream.

mcp.* - MCP server lifecycle

`type`	Description
`mcp.initialize`	One or more MCP server sessions were initialized.
`mcp.auth_required`	MCP server(s) need OAuth before proceeding. Resume after the user authorizes.

sandbox.* - Sandbox lifecycle

`type`	Description
`sandbox.created`	A sandbox was provisioned for the Turn.

TurnCreated

First event on every Turn stream. Carries the turn_id. SSE only - not stored in the Turn body’s events.

Field	Type	Required	Description
`type`	`"turn.created"`	Yes
`turn_id`	string	Yes	Unique ID for this Turn.
`created_at`	string	Yes	ISO-8601 timestamp when the Turn was created.

ModelMessageDelta

The most frequent SSE event. Carries an incremental LLM delta for one thread - assistant text and/or tool call chunks. Group deltas by thread_id and accumulate; a delta with a non-null finish_reason signals the message is complete. The Turn body stores the assembled ModelMessage instead of these deltas. Use ModelMessageBuilder from truefoundry_gateway_sdk to handle delta accumulation.

Field	Type	Required	Description
`type`	`"model.message"`	Yes
`thread_id`	string	Yes	Thread this message belongs to. `"main"` for the root agent; a unique ID for sub-agents.
`content`	string	No	Incremental text content for this delta.
`tool_calls`	`ToolCallDelta[]`	No	Tool call chunks to accumulate by `index`; fully assembled when `finish_reason` is set.
`finish_reason`	string	No	Set on the final delta. Signals the accumulated message is complete.

Each ToolCallDelta is shaped like an OpenAI streaming tool-call chunk. id, type, and tool_info appear only on the first chunk for a given index; function.arguments is a partial JSON string concatenated across chunks.

Field	Type	Required	Description
`index`	int	Yes	Position in the `tool_calls` array; used to merge chunks.
`id`	string	No	Tool call ID. First chunk only.
`type`	`"function"`	No	First chunk only.
`function`	`ToolCallFunctionDelta`	Yes	The function name and partial arguments for this chunk.
`tool_info`	`ToolCallInfo`	No	MCP / routing metadata. First chunk only.

ToolCallFunctionDelta:

Field	Type	Required	Description
`name`	string	No	Tool name. Present only in the first chunk for this `index`.
`arguments`	string	No	Partial JSON string; concatenate across chunks.

ToolCallInfo carries metadata about the underlying MCP server. Read the tool name from original_tool_name.

Field	Type	Required	Description
`mcp_server_id`	string	Yes	ID of the MCP server backing this tool.
`mcp_server_name`	string	Yes	Name of the MCP server backing this tool.
`original_tool_name`	string	Yes	The tool’s original name on the MCP server.
`is_approval_required`	bool	No	Whether the tool call requires human approval.
`is_deferred`	bool	No	Whether the tool call is deferred.
`is_client_side`	bool	No	Whether the tool runs client-side (for example, `ask_user_question`).

ModelMessage

The assembled (non-delta) form of a model message: fully concatenated content and fully reconstructed tool_calls, with finish_reason always set. This is the form stored in the Turn body’s events (and returned by the list-events endpoint), and the value ModelMessageBuilder.build_and_reset() returns after merging ModelMessageDelta events.

Field	Type	Required	Description
`type`	`"model.message"`	Yes
`thread_id`	string	Yes	Thread this message belongs to. `"main"` for the root agent; a unique ID for sub-agents.
`content`	string	No	The complete assistant text.
`tool_calls`	`ToolCall[]`	No	Fully assembled tool calls.
`finish_reason`	string	Yes	Why the message ended (for example, `"stop"` or `"tool_calls"`).

Each ToolCall is shaped like OpenAI’s ChatCompletionMessageToolCall. Read the tool name from tool_info.original_tool_name.

Field	Type	Required	Description
`id`	string	Yes	Tool call ID.
`type`	`"function"`	Yes	Always `"function"`.
`function`	`ToolCallFunction`	Yes	The tool name and complete arguments.
`tool_info`	`ToolCallInfo`	No	MCP / routing metadata (same shape as above).

ToolCallFunction:

Field	Type	Required	Description
`name`	string	Yes	Tool name.
`arguments`	string	Yes	Complete JSON string of the tool’s input arguments.

ToolResponse

A complete server-side tool result returned to the LLM. Not a streaming delta - one event carries the full result.

Field	Type	Required	Description
`type`	`"tool.response"`	Yes
`thread_id`	string	Yes	Thread this result belongs to.
`tool_call_id`	string	Yes	Links this result back to the originating tool call.
`content`	string	Yes	The tool result.

ThreadCreated

Emitted when a sub-agent thread starts.

Not emitted for the main thread. The main thread (thread_id: "main") is the root agent; it is created automatically on the session’s first Turn and is never the subject of a thread.created or thread.done event. thread.created/thread.done track only sub-agent threads.

Field	Type	Required	Description
`type`	`"thread.created"`	Yes
`thread_id`	string	Yes	Unique ID for the sub-agent thread.
`parent`	`AgentParent`	Yes	Parent thread and tool call that spawned this sub-agent.
`agent_info`	`AgentInfo`	Yes	Name and input of the thread’s agent.
`created_at`	string	No

ThreadDone

Emitted when a sub-agent thread reaches a terminal state. Does not terminate the overall Turn stream.

Not emitted for the main thread. The main thread is never “done” - it lives across Turns for the lifetime of the session. The overall Turn’s completion is signalled by turn.done, and the session ends only when it is cancelled.

Field	Type	Required	Description
`type`	`"thread.done"`	Yes
`thread_id`	string	Yes	Unique ID for the sub-agent thread.
`status`	`"done"` \| `"error"`	Yes	`done` - completed normally, `output` is present. `error` - failed, `message` is present.
`output`	object	No	Final accumulated output. Present when `status` is `"done"`.
`parent`	`AgentParent`	Yes	Parent thread and tool call that spawned this sub-agent.
`message`	string	No	Present when `status` is `"error"`.

McpInitialized

Emitted when one or more MCP server sessions are initialized for a thread.

Field	Type	Required	Description
`type`	`"mcp.initialize"`	Yes
`thread_id`	string	Yes
`content`	`McpInitializationInfo[]`	Yes	List of initialized MCP servers, each with `mcp_server_name` and `session_id`.

SandboxCreated

Emitted once when a sandbox is provisioned for the Turn. The sandbox is reused across Turns within the session.

Field	Type	Required	Description
`type`	`"sandbox.created"`	Yes
`sandbox_id`	string	Yes	Unique identifier for the provisioned sandbox.

McpAuthRequired

Emitted when one or more MCP servers require OAuth authorization before the agent can proceed. The stream ends after this event. Resume by calling create_turn() again on the same session after the user completes the OAuth flow - no input is required (omit input or pass []).

Field	Type	Required	Description
`type`	`"mcp.auth_required"`	Yes
`servers`	`McpServerAuthInfo[]`	Yes	List of servers needing authorization, each with `mcp_server_name`, `auth_url`, and `thread_ids`.

ToolApprovalRequired

Emitted when one or more tool calls require explicit human approval before they can run. The stream ends after this event. Resume by sending UserToolApproval items - one per pending tool_call_id - on the next create_turn().

Field	Type	Required	Description
`type`	`"tool.approval_required"`	Yes
`thread_id`	string	Yes
`tool_calls`	`ToolCallRef[]`	Yes	The tool calls awaiting approval, each with an `id`.

ToolResponseRequired

Emitted when the agent has called a client-side tool and is waiting for the result. The stream ends after this event. Resume by sending UserToolResponse items - one per pending tool_call_id - on the next create_turn().

Field	Type	Required	Description
`type`	`"tool.response_required"`	Yes
`thread_id`	string	Yes
`tool_calls`	`ToolCallRef[]`	Yes	The tool calls awaiting a client-supplied result, each with an `id`.

TurnDone

Final event on every Turn stream. The status field distinguishes normal completion from cancellation or error. SSE only - not stored in the Turn body’s events. Its output mirrors the Turn body’s output field, listing any terminal events that occurred.

Field	Type	Required	Description
`type`	`"turn.done"`	Yes
`status`	`"done"` \| `"cancelled"` \| `"error"`	Yes
`output`	`TurnOutputEvent[]`	No	Terminal events that occurred this Turn (`tool.approval_required`, `tool.response_required`, `mcp.auth_required`) plus accumulated model messages.
`cancellation_reason`	string	No	Present when `status` is `"cancelled"`. Values: `"client-cancelled"`, `"server-execution-timeout"`.
`message`	string	No	Present when `status` is `"error"`. Human-readable error description.

Complete usage example

This is a complete, runnable terminal chat client built on the SDK. It handles every event the harness can emit:

streams assistant tokens in real time,
shows tool calls and tool results inline,
prompts the user to answer ask_user_question calls,
prompts the user to allow or deny tool approvals,
shows in-chat MCP OAuth prompts,
handles parallel sub-agents,
continues the conversation across Turns.

Python

"""Terminal chat client for the TrueFoundry Agent Harness SDK."""

import argparse, json, os, sys
from typing import Optional, Union

from truefoundry_gateway_sdk import (
    ApprovalAllow,
    ApprovalDeny,
    McpAuthRequired,
    McpInitialized,
    ModelMessage,
    ModelMessageDelta,
    ModelMessageBuilder,
    SandboxCreated,
    Session,
    ThreadCreated,
    ThreadDone,
    ToolApprovalRequired,
    ToolCall,
    ToolResponse,
    ToolResponseRequired,
    TrueFoundryGateway,
    TurnCreated,
    TurnDone,
    TurnDoneStatus,
    TurnInput,
    UserMessage,
    UserToolApproval,
    UserToolResponse,
)

PendingNextTurnRequest = Union[ToolApprovalRequired, ToolResponseRequired, McpAuthRequired]


class ChatSession:
    """Holds turn-to-turn state for one conversation."""

    def __init__(self, session: Session) -> None:
        self.session = session
        self.builders: dict[str, ModelMessageBuilder] = {}
        self.open_subagent_threads: set[str] = set()
        self.thread_labels: dict[str, str] = {"main": "main"}
        self.open_tool_calls: dict[str, ToolCall] = {}
        self.pending_next_turn_requests: list[PendingNextTurnRequest] = []
        self._main_streaming = False

    def label(self, thread_id: str) -> str:
        return self.thread_labels.get(thread_id, "main")

    def _handle_model_message_delta(self, event: ModelMessageDelta) -> None:
        # ModelMessageDelta events are streaming deltas. ModelMessageBuilder merges
        # content and tool_calls by index; flush once finish_reason is set.
        builder = self.builders.setdefault(event.thread_id, ModelMessageBuilder())
        builder.add(event)

        if event.thread_id == "main" and event.content:
            if not self._main_streaming:
                print("\nassistant: ", end="", flush=True)
                self._main_streaming = True
            print(event.content, end="", flush=True)

        if event.finish_reason is not None:
            self._flush_model_message(builder.build_and_reset())

    def _flush_model_message(self, msg: ModelMessage) -> None:
        thread_id = msg.thread_id
        label = self.label(thread_id)

        if thread_id == "main":
            if self._main_streaming:
                print()
                self._main_streaming = False
            elif msg.content:
                print(f"\nassistant: {msg.content}")
        elif thread_id in self.open_subagent_threads and msg.content:
            print(f"\n[{label}] {msg.content}")

        for tool_call in msg.tool_calls:
            if not tool_call.id or not tool_call.function.name:
                continue
            self.open_tool_calls[tool_call.id] = tool_call
            if tool_call.tool_info.original_tool_name == "ask_user_question":
                continue
            self._print_tool_call(thread_id, tool_call)

    def _print_tool_call(self, thread_id: str, tool_call: ToolCall) -> None:
        label = self.label(thread_id)
        try:
            args = json.loads(tool_call.function.arguments or "{}")
        except json.JSONDecodeError:
            args = tool_call.function.arguments

        args_preview = json.dumps(args)[:160] if isinstance(args, dict) else str(args)[:160]
        prefix = "main" if thread_id == "main" else label
        print(f"\n[{prefix} -> {tool_call.tool_info.original_tool_name}] {args_preview}")

    def _handle_tool_response(self, event: ToolResponse) -> None:
        self.open_tool_calls.pop(event.tool_call_id, None)

        label = self.label(event.thread_id)
        prefix = "main" if event.thread_id == "main" else label
        preview = event.content[:200].replace("\n", " ")
        print(f"\n[{prefix} tool result] {preview}")

    def _prompt_client_side_response(
        self,
        pending_event: ToolResponseRequired,
        tool_call: ToolCall,
    ) -> UserToolResponse:
        if tool_call.tool_info.original_tool_name == "ask_user_question":
            try:
                args = json.loads(tool_call.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}
            question = args.get("question", "Answer:")
            options = args.get("options") or []
            print(f"\nassistant asks: {question}")
            for idx, option in enumerate(options, 1):
                print(f"  {idx}. {option}")
            if options:
                print("  Or type a free-form answer.")
            answer = input("you: ").strip()
            if options and answer.isdigit() and 1 <= int(answer) <= len(options):
                content = options[int(answer) - 1]
            else:
                content = answer
        else:
            print(f"\nclient-side tool response required: {tool_call.tool_info.original_tool_name}")
            print(f"  arguments: {tool_call.function.arguments}")
            content = input("you: ").strip()

        return UserToolResponse(
            thread_id=pending_event.thread_id,
            tool_call_id=tool_call.id,
            content=content,
        )

    def _prompt_approval(
        self,
        pending_event: ToolApprovalRequired,
        tool_call: ToolCall,
    ) -> UserToolApproval:
        try:
            arguments = json.loads(tool_call.function.arguments or "{}")
        except json.JSONDecodeError:
            arguments = tool_call.function.arguments

        print(f"\napproval needed: {tool_call.tool_info.original_tool_name}")
        print(f"  arguments: {json.dumps(arguments, indent=2)}")
        choice = input("  allow / deny / reason for denial: ").strip().lower()
        if choice in {"a", "allow", "yes", "y"}:
            approval = ApprovalAllow()
        elif choice in {"d", "deny", "no", "n"}:
            approval = ApprovalDeny()
        else:
            approval = ApprovalDeny(reason=choice)

        return UserToolApproval(
            thread_id=pending_event.thread_id,
            tool_call_id=tool_call.id,
            approval=approval,
        )

    def _prompt_mcp_auth_required(self, event: McpAuthRequired) -> None:
        print("\n[auth] MCP authorization required. Complete OAuth for each server:")
        for server in event.servers:
            print(f"        {server.mcp_server_name}: {server.auth_url}")
        while True:
            answer = input("        Type 'yes' once you've authenticated: ").strip().lower()
            if answer in {"y", "yes"}:
                break

    def build_next_turn_input(self) -> TurnInput:
        next_turn: list[UserToolApproval | UserToolResponse] = []
        pending = self.pending_next_turn_requests[:]
        self.pending_next_turn_requests.clear()

        for pending_event in pending:
            if isinstance(pending_event, McpAuthRequired):
                self._prompt_mcp_auth_required(pending_event)
                continue
            for tool_ref in pending_event.tool_calls:
                tool_call = self.open_tool_calls[tool_ref.id]
                if isinstance(pending_event, ToolResponseRequired):
                    next_turn.append(
                        self._prompt_client_side_response(pending_event, tool_call)
                    )
                else:
                    next_turn.append(
                        self._prompt_approval(pending_event, tool_call)
                    )

        return next_turn

    def _handle_event(self, event: object) -> Optional[TurnDoneStatus]:
        if isinstance(event, TurnCreated):
            return None

        if isinstance(event, ThreadCreated):
            if event.thread_id != "main":
                self.open_subagent_threads.add(event.thread_id)
                name = event.agent_info.name
                self.thread_labels[event.thread_id] = name
                print(f"\n[subagent {name} started]")
            return None

        if isinstance(event, ThreadDone):
            if event.thread_id != "main":
                self.open_subagent_threads.discard(event.thread_id)
                label = self.label(event.thread_id)
                if event.status == "error":
                    print(f"\n[{label} error] {event.message}")
                else:
                    print(f"[subagent {label} finished]")
            return None

        if isinstance(event, McpInitialized):
            for server in event.content:
                print(f"[mcp] connected to {server.mcp_server_name}")
            return None

        if isinstance(event, McpAuthRequired):
            self.pending_next_turn_requests.append(event)
            return None

        if isinstance(event, SandboxCreated):
            print(f"[sandbox] provisioned ({event.sandbox_id})")
            return None

        if isinstance(event, ModelMessageDelta):
            self._handle_model_message_delta(event)
            return None

        if isinstance(event, ToolResponse):
            self._handle_tool_response(event)
            return None

        if isinstance(event, ToolResponseRequired):
            self.pending_next_turn_requests.append(event)
            return None

        if isinstance(event, ToolApprovalRequired):
            self.pending_next_turn_requests.append(event)
            return None

        if isinstance(event, TurnDone):
            if event.status == "error" and event.message:
                print(f"\n[error] {event.message}")
            return event.status

        print(f"\n[unknown event] {event}")
        return None

    def run_turn(self, input_items: TurnInput | None = None) -> TurnDoneStatus:
        last_status: TurnDoneStatus = "done"
        for event in self.session.create_turn(input=input_items):
            status = self._handle_event(event)
            if status is not None:
                last_status = status
        return last_status


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Terminal chat client for the TrueFoundry Agent Harness.",
    )
    parser.add_argument(
        "agent_name",
        nargs="?",
        default=os.environ.get("AGENT_NAME"),
        help="Registered agent name (default: AGENT_NAME environment variable)",
    )
    args = parser.parse_args()
    if not args.agent_name:
        parser.error("agent_name is required (pass as argument or set AGENT_NAME)")
    return args


def main() -> None:
    args = parse_args()

    base_url = os.environ.get("GATEWAY_BASE_URL")
    api_key = os.environ.get("TFY_API_KEY")
    if not base_url or not api_key:
        print("Set GATEWAY_BASE_URL and TFY_API_KEY environment variables.")
        sys.exit(1)

    client = TrueFoundryGateway(
        base_url=base_url,
        api_key=api_key,
    )
    session = client.agents.create_session(agent_name=args.agent_name)
    chat = ChatSession(session)

    print("Type a message, or Ctrl-D to exit.")
    while True:
        while chat.pending_next_turn_requests:
            chat.run_turn(chat.build_next_turn_input())

        try:
            text = input("\nyou: ").strip()
        except EOFError:
            print()
            break
        if not text:
            continue
        chat.run_turn([UserMessage(content=text)])


if __name__ == "__main__":
    main()

Sample run

$ python example.py support-bot
Type a message, or Ctrl-D to exit.

you: Refund invoice INV-2031 for tenant acme.
[mcp] connected to truefoundry-mcp
[sandbox] provisioned (my-tenant.a1b2c3d4)

assistant asks: Which refund reason should I record?
  1. Customer churn
  2. Billing error
  3. Goodwill credit
  Or type a free-form answer.
you: 2

assistant: Found invoice INV-2031 for $1,240.00 on 2026-04-12.

approval needed: process_refund
  arguments: {
  "invoice_id": "INV-2031",
  "amount": 1240.0,
  "reason": "Billing error"
}
  allow / deny / reason for denial: allow

[main tool result] {"refund_id":"RF-9914","status":"completed"}

assistant: Refund RF-9914 of $1,240.00 was processed for INV-2031.

you: ^D

Adapting the flow

A few common variations on top of the same skeleton:

Persisted history. Save session.id after the first Turn. On the next process start, call client.agents.get_session(session_id) and use list_turns() with turn.events() to rebuild UI state, or turn.stream() if the latest Turn is still running.
JSON output for piping. Replace the print calls inside _handle_event with json.dumps(event.model_dump()) to emit one event per line for downstream tools.
Generative UI. When you detect a fenced ```openui block in assembled model.message content, hand the block to the OpenUI React renderer instead of printing it. Everything else stays the same.
Browser / web UI. The same event shape is delivered over Server-Sent Events on POST /sessions/{session_id}/turns. Replace session.create_turn with an SSE consumer; the handlers above are unchanged.

Reference

create_session

client.agents.create_session(agent_name, title=None) -> Session

Create a conversation Session for a saved agent.

Param	Type	Required	Description
`agent_name`	string	Yes	Name of the saved agent to invoke.
`title`	string	No	Optional human-readable title for the session.

list_sessions

client.agents.list_sessions(agent_name, created_by_subject_slug=None, start_timestamp=None, end_timestamp=None, limit=100) -> Iterator[Session]

Iterate over the Sessions for a saved agent, newest-first, transparently paginating through all pages.

Param	Type	Required	Description
`agent_name`	string	Yes	Filter to sessions for a specific named agent.
`created_by_subject_slug`	string	No	Filter to sessions created by a specific user / service account.
`start_timestamp`	string	No	ISO-8601 lower bound on session creation time.
`end_timestamp`	string	No	ISO-8601 upper bound on session creation time.
`limit`	int	No	Number of sessions fetched per page (not a total cap). Defaults to `100`.

Session

The conversation context for a saved agent. Turns created within a Session are chained automatically, so each Turn sees the history of earlier ones. Key members:

Member	Type	Description
`id`	string	Unique session identifier. Persist it to resume the conversation later via `client.agents.get_session(session_id)`.
`create_turn(...)`	method	Start a Turn and stream its events, or pass `stream=False` to get a `Turn` back without streaming. See `create_turn`.
`list_turns(...)`	method	Iterate over the Turns in this session, newest first by default.
`get_turn(turn_id)`	method	Fetch a single Turn by ID.
`cancel()`	method	Cancel the session and any currently running Turn. Idempotent; after this, `create_turn()` raises.

create_turn

session.create_turn(input=None, previous_turn_id="auto", stream=True)
  -> Iterator[TurnStreamEvent]   # stream=True (default)
  -> Turn                        # stream=False

Start a Turn in the session. With stream=True (default) it streams the Turn’s events - the first event is always turn.created and the stream closes on a terminal event. With stream=False it returns a Turn without streaming.

Param	Type	Required	Description
`input`	`TurnInput`	No	Input items for this Turn. See Turn input. Omit or pass `[]` to resume after `mcp.auth_required`.
`previous_turn_id`	`string` \| `"auto"`	No	Turn chaining point. Defaults to `"auto"`, letting the server chain to the latest Turn.
`stream`	`bool`	No	Stream events as they arrive. Defaults to `True`. Set `False` to create the Turn and return a `Turn` without streaming - drive it with `wait()` / `state()` / `events()`.

Turn

A single request/response cycle within a Session. Transitions: running → done | cancelled | error.

Member	Type	Description
`id`	string	Unique turn identifier (UUIDv7).
`session`	`Session`	The Session this Turn belongs to.
`previous_turn_id`	string \| None	Turn ID this Turn chains from, or `None` for the first Turn.
`created_by`	string	Subject (user / service account) that created this Turn.
`created_at`	string	ISO-8601 timestamp of Turn creation.
`input`	`TurnInput` \| None	The Turn input that triggered this Turn, if any.
`state()`	method	Fetch the current lifecycle state from the server. Returns one of `TurnRunningState`, `TurnDoneState`, `TurnCancelledState`, or `TurnErrorState`.
`stream(...)`	method	Subscribe to the live SSE stream for this Turn. Handles reconnection and resumption transparently. Closes when the Turn reaches a terminal state. Raises if the Turn is not running — use `events()` for completed Turns.
`events(...)`	method	Return a point-in-time snapshot of events accumulated so far, paginating automatically. Works for both running and terminal Turns. `order="asc"` is natural for replaying history forward. `TurnCreated` and `TurnDone` are SSE-only and do not appear here.
`wait()`	method	Block until the Turn reaches a terminal state and return it. Returns immediately if already terminal.
`cancel()`	method	Request cancellation. Idempotent; safe to call on already-terminal Turns. Cancellation is asynchronous — a few more events may arrive before the stream closes.

To define or configure the agent itself, see Agent.

​Overview

​How interaction works

​Creating a Session

​Listing sessions for an agent

​Interacting with a Session by creating a Turn

​Non-streaming Turn

​Attaching images or files to a Turn

​Listing Turns

​Cancelling a Turn

​Streaming a running Turn’s events

​Getting events for a completed Turn

​Handling MCP Outbound Auth

​Merging the model message stream

​Handling tool approvals

​Answering agent’s questions

​Turn input

​UserMessage

​UserContentPart

​UserToolApproval

​UserToolResponse

​Turn Events

​TurnCreated

​ModelMessageDelta

​ModelMessage

​ToolResponse

​ThreadCreated

​ThreadDone

​McpInitialized

​SandboxCreated

​McpAuthRequired

​ToolApprovalRequired

​ToolResponseRequired

​TurnDone

​Complete usage example

​Sample run

​Adapting the flow

​Reference

​create_session

​list_sessions

​Session

​create_turn

​Turn

Overview

How interaction works

Creating a Session

Listing sessions for an agent

Interacting with a Session by creating a Turn

Non-streaming Turn

Attaching images or files to a Turn

Listing Turns

Cancelling a Turn

Streaming a running Turn’s events

Getting events for a completed Turn

Handling MCP Outbound Auth

Merging the model message stream

Handling tool approvals

Answering agent’s questions

Turn input

UserMessage

UserContentPart

UserToolApproval

UserToolResponse

Turn Events

TurnCreated

ModelMessageDelta

ModelMessage

ToolResponse

ThreadCreated

ThreadDone

McpInitialized

SandboxCreated

McpAuthRequired

ToolApprovalRequired

ToolResponseRequired

TurnDone

Complete usage example

Sample run

Adapting the flow

Reference

create_session

list_sessions

Session

create_turn

Turn