AgentChatAppEnvironment
Package: flyte.ai.chat
An flyte.app.AppEnvironment that spins up a FastAPI chat
interface backed by any object satisfying the flyte.ai.agents.AgentProtocol.
Parameters
class AgentChatAppEnvironment(
name: str,
depends_on: List[Environment] = <factory>,
pod_template: Optional[Union[str, PodTemplate]] = None,
description: Optional[str] = None,
secrets: Optional[SecretRequest] = None,
env_vars: Optional[Dict[str, str]] = None,
resources: Optional[Resources] = None,
interruptible: bool = False,
image: Union[str, Image, Literal['auto'], None] = 'auto',
include: Tuple[str, ...] = <factory>,
port: int | Port = 8080,
args: Optional[Union[List[str], str]] = None,
command: Optional[Union[List[str], str]] = None,
requires_auth: bool = True,
scaling: Scaling = <factory>,
domain: Domain | None = <factory>,
links: List[Link] = <factory>,
parameters: List[Parameter] = <factory>,
cluster_pool: str = 'default',
timeouts: Timeouts = <factory>,
type: str = 'AgentChat',
agent: Any = None,
title: str | None = None,
subtitle: str | None = None,
prompt_nudges: list[dict[str, str]] = <factory>,
theme: CustomTheme | None = None,
custom_css: str = '',
logo_url: str | None = None,
additional_buttons: list[dict[str, str]] = <factory>,
passthrough_auth: bool = False,
passthrough_auth_excluded_paths: frozenset[str] | None = None,
task_entrypoint: Any | None = None,
)| Parameter | Type | Description |
|---|---|---|
name |
str |
|
depends_on |
List[Environment] |
|
pod_template |
Optional[Union[str, PodTemplate]] |
|
description |
Optional[str] |
|
secrets |
Optional[SecretRequest] |
|
env_vars |
Optional[Dict[str, str]] |
|
resources |
Optional[Resources] |
|
interruptible |
bool |
|
image |
Union[str, Image, Literal['auto'], None] |
|
include |
Tuple[str, ...] |
|
port |
int | Port |
|
args |
Optional[Union[List[str], str]] |
|
command |
Optional[Union[List[str], str]] |
|
requires_auth |
bool |
|
scaling |
Scaling |
|
domain |
Domain | None |
|
links |
List[Link] |
|
parameters |
List[Parameter] |
|
cluster_pool |
str |
|
timeouts |
Timeouts |
|
type |
str |
|
agent |
Any |
Any object implementing the flyte.ai.agents.AgentProtocol. |
title |
str | None |
Title displayed in the UI header and browser tab. Defaults to the environment name. |
subtitle |
str | None |
Optional short subtitle displayed below the title in the header area. Use it to explain what the agent does. |
prompt_nudges |
list[dict[str, str]] |
Optional list of prompt-nudge cards shown before the first message. Each entry is a dict with "label" (short card title) and "prompt" (the query text sent when clicked). |
theme |
CustomTheme | None |
Optional flyte.ai.chat.CustomTheme instance that controls the UI accent colors via human-readable attributes. When provided, the theme CSS is generated automatically and prepended to any custom_css. |
custom_css |
str |
Optional CSS string appended after the default styles (and after theme CSS, if a theme is provided). Use this for fine-grained overrides beyond what flyte.ai.chat.CustomTheme exposes. |
logo_url |
str | None |
Optional URL to an image displayed to the left of the title in the header bar. When None (default), no logo is shown. |
additional_buttons |
list[dict[str, str]] |
Optional list of action-button dicts rendered to the right of the Send button. Each dict must have "button_text" and "button_url" keys. The first entry is displayed as a prominent primary button; any extra entries appear in a drop-up menu accessed via a chevron. |
passthrough_auth |
bool |
When True, the FastAPI app initializes flyte.init_passthrough at startup and adds FastAPIPassthroughAuthMiddleware so incoming Authorization / cookie headers are forwarded to Flyte remote calls. Enable this when using an agent with @env.task tools — nested task execution needs caller credentials (same pattern as FlyteWebhookAppEnvironment). |
passthrough_auth_excluded_paths |
frozenset[str] | None |
Paths skipped by passthrough middleware. When omitted, defaults include the HTML shell (/), /api/tools, /api/nudges, health, and docs routes so the sidebar and nudges load without Authorization headers; /api/chat still requires credentials. Only used when passthrough_auth is True. |
task_entrypoint |
Any | None |
Optional Flyte task used as the chat handler entrypoint. When set, /api/chat calls the task (via flyte.run.aio) instead of calling agent.run directly. This is useful for agents whose tool calls must run under a parent task context (e.g. an Agent in code_mode using durable @env.task tools). When streaming chat (stream: true), progress lines use flyte.remote.Run.watch on the returned run (first RUNNING → generating_code, next → executing). Fine-grained per-turn phases still require agent.run in the web process, or future worker-side signaling. The entrypoint may accept either: - (message: str, history: list[dict[str, str]]); or - (message: str). The return value may be a flyte.ai.agents.protocol.AgentResult, a dict with keys like summary/charts/code, or a plain string (treated as summary). |
Properties
| Property | Type | Description |
|---|---|---|
endpoint |
str |
Methods
| Method | Description |
|---|---|
add_dependency() |
Add one or more environment dependencies so they are deployed together. |
build_fastapi_app() |
Construct the FastAPI application (routes, HTML shell, optional auth). |
clone_with() |
|
container_args() |
|
container_cmd() |
|
container_command() |
|
get_port() |
|
on_shutdown() |
Decorator to define the shutdown function for the app environment. |
on_startup() |
Decorator to define the startup function for the app environment. |
server() |
Decorator to define the server function for the app environment. |
add_dependency()
def add_dependency(
*env: Environment,
)Add one or more environment dependencies so they are deployed together.
When you deploy this environment, any environments added via
add_dependency will also be deployed. This is an alternative to
passing depends_on=[...] at construction time, useful when the
dependency is defined after the environment is created.
Duplicate dependencies are silently ignored. An environment cannot depend on itself.
| Parameter | Type | Description |
|---|---|---|
*env |
Environment |
One or more Environment instances to add as dependencies. |
build_fastapi_app()
def build_fastapi_app()Construct the FastAPI application (routes, HTML shell, optional auth).
Useful for tests and advanced mounting; the deployed server uses this via
AgentChatAppEnvironment._fastapi_server.
clone_with()
def clone_with(
name: str,
image: Optional[Union[str, Image, Literal['auto']]] = None,
resources: Optional[Resources] = None,
env_vars: Optional[dict[str, str]] = None,
secrets: Optional[SecretRequest] = None,
depends_on: Optional[List[Environment]] = None,
description: Optional[str] = None,
interruptible: Optional[bool] = None,
**kwargs: Any,
) -> AppEnvironment| Parameter | Type | Description |
|---|---|---|
name |
str |
|
image |
Optional[Union[str, Image, Literal['auto']]] |
|
resources |
Optional[Resources] |
|
env_vars |
Optional[dict[str, str]] |
|
secrets |
Optional[SecretRequest] |
|
depends_on |
Optional[List[Environment]] |
|
description |
Optional[str] |
|
interruptible |
Optional[bool] |
|
**kwargs |
Any |
container_args()
def container_args(
serialize_context: SerializationContext,
) -> List[str]| Parameter | Type | Description |
|---|---|---|
serialize_context |
SerializationContext |
container_cmd()
def container_cmd(
serialize_context: SerializationContext,
parameter_overrides: list[Parameter] | None = None,
) -> List[str]| Parameter | Type | Description |
|---|---|---|
serialize_context |
SerializationContext |
|
parameter_overrides |
list[Parameter] | None |
container_command()
def container_command(
serialization_context: SerializationContext,
) -> list[str]| Parameter | Type | Description |
|---|---|---|
serialization_context |
SerializationContext |
get_port()
def get_port()on_shutdown()
def on_shutdown(
fn: F,
) -> FDecorator to define the shutdown function for the app environment.
This function is called after the server function is called.
This decorated function can be a sync or async function, and accepts input parameters based on the Parameters defined in the AppEnvironment definition.
| Parameter | Type | Description |
|---|---|---|
fn |
F |
on_startup()
def on_startup(
fn: F,
) -> FDecorator to define the startup function for the app environment.
This function is called before the server function is called.
The decorated function can be a sync or async function, and accepts input parameters based on the Parameters defined in the AppEnvironment definition.
| Parameter | Type | Description |
|---|---|---|
fn |
F |
server()
def server(
fn: F,
) -> FDecorator to define the server function for the app environment.
This decorated function can be a sync or async function, and accepts input parameters based on the Parameters defined in the AppEnvironment definition.
| Parameter | Type | Description |
|---|---|---|
fn |
F |