"""

Copilot Session - represents a single conversation session with the Copilot CLI.

This module provides the CopilotSession class for managing individual

conversation sessions with the Copilot CLI, along with all session-related

configuration and handler types.

"""

from __future__ import annotations

import asyncio

import functools

import inspect

import logging

import os

import pathlib

import threading

import time

from collections.abc import Awaitable, Callable

from dataclasses import dataclass

from datetime import UTC, datetime

from types import TracebackType

from typing import TYPE_CHECKING, Any, Literal, NotRequired, Required, TypedDict, cast

from ._diagnostics import log_timing

from ._jsonrpc import JsonRpcError, ProcessExitedError

from ._telemetry import get_trace_context, trace_context

from .canvas import CanvasError, CanvasHandler, OpenCanvasInstance

from .generated.rpc import (

CanvasHandler as RpcCanvasHandler,

)

from .generated.rpc import (

CanvasProviderCloseRequest,

CanvasProviderInvokeActionRequest,

CanvasProviderOpenRequest,

CanvasProviderOpenResult,

ClientSessionApiHandlers,

CommandsHandlePendingCommandRequest,

ExternalToolTextResultForLlm,

HandlePendingToolCallRequest,

LogRequest,

ModelSwitchToRequest,

PermissionDecision,

PermissionDecisionApproveOnce,

PermissionDecisionRequest,

PermissionDecisionUserNotAvailable,

SessionLogLevel,

SessionRpc,

UIElicitationRequest,

UIElicitationResponse,

UIElicitationResponseAction,

UIElicitationSchema,

UIElicitationSchemaProperty,

UIElicitationSchemaPropertyType,

UIElicitationSchemaType,

UIHandlePendingElicitationRequest,

)

from .generated.rpc import (

ContextTier as _RpcContextTier,

)

from .generated.rpc import ModelCapabilitiesOverride as _RpcModelCapabilitiesOverride

from .generated.session_events import (

AssistantMessageData,

CapabilitiesChangedData,

CommandExecuteData,

ElicitationRequestedData,

ExternalToolRequestedData,

PermissionRequest,

PermissionRequestedData,

SessionCanvasOpenedData,

SessionErrorData,

SessionEvent,

SessionIdleData,

session_event_from_dict,

)

from .generated.session_events import (

ReasoningSummary as _RpcReasoningSummary,

)

from .tools import Tool, ToolHandler, ToolInvocation, ToolResult

logger = logging.getLogger(__name__)

if TYPE_CHECKING:

from .client import ModelCapabilitiesOverride

from .session_fs_provider import SessionFsProvider

# Re-export SessionEvent under an alias used internally

SessionEventTypeAlias = SessionEvent

# ============================================================================

# Reasoning Effort

# ============================================================================

ReasoningEffort = Literal["low", "medium", "high", "xhigh"]

ReasoningSummary = Literal["none", "concise", "detailed"]

ContextTier = Literal["default", "long_context"]

SessionFsConventions = Literal["posix", "windows"]

class SessionFsCapabilities(TypedDict, total=False):

sqlite: bool

class SessionFsConfig(TypedDict):

initial_working_directory: str

session_state_path: str

conventions: SessionFsConventions

capabilities: NotRequired[SessionFsCapabilities]

# ============================================================================

# Attachment Types

# ============================================================================

class SelectionRange(TypedDict):

line: int

character: int

class Selection(TypedDict):

start: SelectionRange

end: SelectionRange

class FileAttachment(TypedDict):

"""File attachment."""

type: Literal["file"]

path: str

displayName: NotRequired[str]

class DirectoryAttachment(TypedDict):

"""Directory attachment."""

type: Literal["directory"]

path: str

displayName: NotRequired[str]

class SelectionAttachment(TypedDict):

"""Selection attachment with text from a file."""

type: Literal["selection"]

filePath: str

displayName: str

selection: NotRequired[Selection]

text: NotRequired[str]

class BlobAttachment(TypedDict):

"""Inline base64-encoded content attachment (e.g. images)."""

type: Literal["blob"]

data: str

"""Base64-encoded content"""

mimeType: str

"""MIME type of the inline data"""

displayName: NotRequired[str]

Attachment = FileAttachment | DirectoryAttachment | SelectionAttachment | BlobAttachment

# ============================================================================

# System Message Configuration

# ============================================================================

class SystemMessageAppendConfig(TypedDict, total=False):

"""

Append mode: Use CLI foundation with optional appended content.

"""

mode: NotRequired[Literal["append"]]

content: NotRequired[str]

class SystemMessageReplaceConfig(TypedDict):

"""

Replace mode: Use caller-provided system message entirely.

Removes all SDK guardrails including security restrictions.

"""

mode: Literal["replace"]

content: str

# Known system message section identifiers for the "customize" mode.

SectionTransformFn = Callable[[str], str | Awaitable[str]]

"""Transform callback: receives current section content, returns new content."""

SectionOverrideAction = Literal["replace", "remove", "append", "prepend"] | SectionTransformFn

"""Override action: a string literal for static overrides, or a callback for transforms."""

SystemMessageSection = Literal[

"identity",

"tone",

"tool_efficiency",

"environment_context",

"code_change_rules",

"guidelines",

"safety",

"tool_instructions",

"custom_instructions",

"runtime_instructions",

"last_instructions",

]

SYSTEM_MESSAGE_SECTIONS: dict[SystemMessageSection, str] = {

"identity": "Agent identity preamble and mode statement",

"tone": "Response style, conciseness rules, output formatting preferences",

"tool_efficiency": "Tool usage patterns, parallel calling, batching guidelines",

"environment_context": "CWD, OS, git root, directory listing, available tools",

"code_change_rules": "Coding rules, linting/testing, ecosystem tools, style",

"guidelines": "Tips, behavioral best practices, behavioral guidelines",

"safety": "Environment limitations, prohibited actions, security policies",

"tool_instructions": "Per-tool usage instructions",

"custom_instructions": "Repository and organization custom instructions",

"runtime_instructions": (

"Runtime-provided context and instructions"

" (e.g. system notifications, memories, workspace context,"

" mode-specific instructions, content-exclusion policy)"

),

"last_instructions": (

"End-of-prompt instructions: parallel tool calling, persistence, task completion"

),

}

class SectionOverride(TypedDict, total=False):

"""Override operation for a single system message section."""

action: Required[SectionOverrideAction]

content: NotRequired[str]

class SystemMessageCustomizeConfig(TypedDict, total=False):

"""

Customize mode: Override individual sections of the system prompt.

Keeps the SDK-managed prompt structure while allowing targeted modifications.

"""

mode: Required[Literal["customize"]]

sections: NotRequired[dict[SystemMessageSection, SectionOverride]]

content: NotRequired[str]

SystemMessageConfig = (

SystemMessageAppendConfig | SystemMessageReplaceConfig | SystemMessageCustomizeConfig

)

# ============================================================================

# Permission Types

# ============================================================================

@dataclass

class PermissionNoResult:

"""Sentinel returned by a permission handler to leave the request unanswered.

Only meaningful against protocol-v1 servers. v2 servers reject ``no-result``

responses; the SDK raises :class:`ValueError` if a v2 server receives one.

Mirrors the ``{kind: "no-result"}`` extension TS adds to its ``PermissionDecision``

union (see ``nodejs/src/types.ts:883``).

"""

kind: Literal["no-result"] = "no-result"

# The decision returned by a permission handler. Identical shape to the wire

# ``PermissionDecision`` discriminated union, plus a :class:`PermissionNoResult`

# sentinel for v1 servers. Construct via the generated variant classes:

# ``PermissionDecisionApproveOnce()``, ``PermissionDecisionReject(feedback=...)``,

# etc. The ``kind`` discriminator is baked in as a ``ClassVar`` default by

# codegen, so callers must not pass it.

PermissionRequestResult = PermissionDecision | PermissionNoResult

_PermissionHandlerFn = Callable[

[PermissionRequest, dict[str, str]],

PermissionRequestResult | Awaitable[PermissionRequestResult],

]

class PermissionHandler:

@staticmethod

def approve_all(

request: PermissionRequest, invocation: dict[str, str]

) -> PermissionRequestResult:

return PermissionDecisionApproveOnce()

# ============================================================================

# User Input Request Types

# ============================================================================

class UserInputRequest(TypedDict, total=False):

"""Request for user input from the agent (enables ask_user tool)"""

question: str

choices: list[str]

allowFreeform: bool

class UserInputResponse(TypedDict):

"""Response to a user input request"""

answer: str

wasFreeform: bool

UserInputHandler = Callable[

[UserInputRequest, dict[str, str]],

UserInputResponse | Awaitable[UserInputResponse],

]

class ExitPlanModeRequest(TypedDict, total=False):

"""Request to exit plan mode and continue with a selected action."""

summary: Required[str]

planContent: NotRequired[str]

actions: Required[list[str]]

recommendedAction: Required[str]

class ExitPlanModeResult(TypedDict, total=False):

"""Response to an exit-plan-mode request."""

approved: Required[bool]

selectedAction: NotRequired[str]

feedback: NotRequired[str]

ExitPlanModeHandler = Callable[

[ExitPlanModeRequest, dict[str, str]],

ExitPlanModeResult | Awaitable[ExitPlanModeResult],

]

class AutoModeSwitchRequest(TypedDict, total=False):

"""Request to switch to auto mode after an eligible rate limit."""

errorCode: NotRequired[str]

retryAfterSeconds: NotRequired[float]

AutoModeSwitchResponse = Literal["yes", "yes_always", "no"]

AutoModeSwitchHandler = Callable[

[AutoModeSwitchRequest, dict[str, str]],

AutoModeSwitchResponse | Awaitable[AutoModeSwitchResponse],

]

# ============================================================================

# Command Types

# ============================================================================

@dataclass

class CommandContext:

"""Context passed to a command handler when a command is executed."""

session_id: str

"""Session ID where the command was invoked."""

command: str

"""The full command text (e.g. ``"/deploy production"``)."""

command_name: str

"""Command name without leading ``/``."""

args: str

"""Raw argument string after the command name."""

CommandHandler = Callable[[CommandContext], Awaitable[None] | None]

"""Handler invoked when a registered command is executed by a user."""

@dataclass

class CommandDefinition:

"""Definition of a slash command registered with the session.

When the CLI is running with a TUI, registered commands appear as

``/commandName`` for the user to invoke.

"""

name: str

"""Command name (without leading ``/``)."""

handler: CommandHandler

"""Handler invoked when the command is executed."""

description: str | None = None

"""Human-readable description shown in command completion UI."""

# ============================================================================

# Session Capabilities

# ============================================================================

class SessionUiCapabilities(TypedDict, total=False):

"""UI capabilities reported by the CLI host."""

elicitation: bool

"""Whether the host supports interactive elicitation dialogs."""

mcpApps: bool

"""**Experimental.** This capability is part of an experimental wire-protocol

surface (SEP-1865) and may change or be removed in a future release.

Whether the runtime has accepted the session's MCP Apps (SEP-1865) opt-in.

``True`` when the consumer set ``enable_mcp_apps=True`` on create/resume and

the runtime's ``MCP_APPS`` feature flag (or ``COPILOT_MCP_APPS=true`` env

override) is on. Otherwise absent or ``False``, indicating the runtime

silently dropped the opt-in."""

class SessionCapabilities(TypedDict, total=False):

"""Capabilities reported by the CLI host for this session."""

ui: SessionUiCapabilities

# ============================================================================

# Elicitation Types (client → server)

# ============================================================================

ElicitationFieldValue = str | float | bool | list[str]

"""Possible value types in elicitation form content."""

class ElicitationResult(TypedDict, total=False):

"""Result returned from an elicitation request."""

action: Required[Literal["accept", "decline", "cancel"]]

"""User action: ``"accept"`` (submitted), ``"decline"`` (rejected),

or ``"cancel"`` (dismissed)."""

content: dict[str, ElicitationFieldValue]

"""Form values submitted by the user (present when action is ``"accept"``)."""

class ElicitationParams(TypedDict):

"""Parameters for a raw elicitation request."""

message: str

"""Message describing what information is needed from the user."""

requestedSchema: dict[str, Any]

"""JSON Schema describing the form fields to present."""

class InputOptions(TypedDict, total=False):

"""Options for the ``input()`` convenience method."""

title: str

"""Title label for the input field."""

description: str

"""Descriptive text shown below the field."""

minLength: int

"""Minimum text length."""

maxLength: int

"""Maximum text length."""

format: str

"""Input format hint (e.g. ``"email"``, ``"uri"``, ``"date"``)."""

default: str

"""Default value for the input field."""

# ============================================================================

# Elicitation Types (server → client callback)

# ============================================================================

class ElicitationContext(TypedDict, total=False):

"""Context for an elicitation handler invocation, combining the request data

with session context. Mirrors the single-argument pattern of CommandContext."""

session_id: Required[str]

"""Identifier of the session that triggered the elicitation request."""

message: Required[str]

"""Message describing what information is needed from the user."""

requestedSchema: dict[str, Any]

"""JSON Schema describing the form fields to present."""

mode: Literal["form", "url"]

"""Elicitation mode: ``"form"`` for structured input, ``"url"`` for browser redirect."""

elicitationSource: str

"""The source that initiated the request (e.g. MCP server name)."""

url: str

"""URL to open in the browser (when mode is ``"url"``)."""

ElicitationHandler = Callable[

[ElicitationContext],

ElicitationResult | Awaitable[ElicitationResult],

]

"""Handler invoked when the server dispatches an elicitation request to this client."""

CreateSessionFsHandler = Callable[["CopilotSession"], "SessionFsProvider"]

# ============================================================================

# Session UI API

# ============================================================================

class SessionUiApi:

"""Interactive UI methods for showing dialogs to the user.

Only available when the CLI host supports elicitation

(``session.capabilities["ui"]["elicitation"] is True``).

Obtained via :attr:`CopilotSession.ui`.

"""

def __init__(self, session: CopilotSession) -> None:

self._session = session

async def elicitation(self, params: ElicitationParams) -> ElicitationResult:

"""Shows a generic elicitation dialog with a custom schema.

Args:

params: Elicitation parameters including message and requestedSchema.

Returns:

The user's response (action + optional content).

Raises:

RuntimeError: If the host does not support elicitation.

"""

self._session._assert_elicitation()

rpc_result = await self._session.rpc.ui.elicitation(

UIElicitationRequest(

message=params["message"],

requested_schema=UIElicitationSchema.from_dict(params["requestedSchema"]),

)

)

result: ElicitationResult = {"action": rpc_result.action.value}

if rpc_result.content is not None:

result["content"] = rpc_result.content

return result

async def confirm(self, message: str) -> bool:

"""Shows a confirmation dialog and returns the user's boolean answer.

Args:

message: The question to ask the user.

Returns:

``True`` if the user accepted, ``False`` otherwise.

Raises:

RuntimeError: If the host does not support elicitation.

"""

self._session._assert_elicitation()

rpc_result = await self._session.rpc.ui.elicitation(

UIElicitationRequest(

message=message,

requested_schema=UIElicitationSchema(

type=UIElicitationSchemaType.OBJECT,

properties={

"confirmed": UIElicitationSchemaProperty(

type=UIElicitationSchemaPropertyType.BOOLEAN,

default=True,

),

},

required=["confirmed"],

),

)

)

return (

rpc_result.action == UIElicitationResponseAction.ACCEPT

and rpc_result.content is not None

and rpc_result.content.get("confirmed") is True

)

async def select(self, message: str, options: list[str]) -> str | None:

"""Shows a selection dialog with a list of options.

Args:

message: Instruction to show the user.

options: List of choices the user can pick from.

Returns:

The selected string, or ``None`` if the user declined/cancelled.

Raises:

RuntimeError: If the host does not support elicitation.

"""

self._session._assert_elicitation()

rpc_result = await self._session.rpc.ui.elicitation(

UIElicitationRequest(

message=message,

requested_schema=UIElicitationSchema(

type=UIElicitationSchemaType.OBJECT,

properties={

"selection": UIElicitationSchemaProperty(

type=UIElicitationSchemaPropertyType.STRING,

enum=options,

),

},

required=["selection"],

),

)

)

if (

rpc_result.action == UIElicitationResponseAction.ACCEPT

and rpc_result.content is not None

and rpc_result.content.get("selection") is not None

):

return str(rpc_result.content["selection"])

return None

async def input(self, message: str, options: InputOptions | None = None) -> str | None:

"""Shows a text input dialog.

Args:

message: Instruction to show the user.

options: Optional constraints for the input field.

Returns:

The entered text, or ``None`` if the user declined/cancelled.

Raises:

RuntimeError: If the host does not support elicitation.

"""

self._session._assert_elicitation()

field: dict[str, Any] = {"type": "string"}

if options:

for key in ("title", "description", "minLength", "maxLength", "format", "default"):

if key in options:

field[key] = options[key]

rpc_result = await self._session.rpc.ui.elicitation(

UIElicitationRequest(

message=message,

requested_schema=UIElicitationSchema.from_dict(

{

"type": "object",

"properties": {"value": field},

"required": ["value"],

}

),

)

)

if (

rpc_result.action == UIElicitationResponseAction.ACCEPT

and rpc_result.content is not None

and rpc_result.content.get("value") is not None

):

return str(rpc_result.content["value"])

return None

# ============================================================================

# Hook Types

# ============================================================================

class PreToolUseHookInput(TypedDict):

"""Input for pre-tool-use hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

toolName: str

toolArgs: Any

class PreToolUseHookOutput(TypedDict, total=False):

"""Output for pre-tool-use hook"""

permissionDecision: Literal["allow", "deny", "ask"]

permissionDecisionReason: str

modifiedArgs: Any

additionalContext: str

suppressOutput: bool

PreToolUseHandler = Callable[

[PreToolUseHookInput, dict[str, str]],

PreToolUseHookOutput | None | Awaitable[PreToolUseHookOutput | None],

]

class PreMcpToolCallHookInput(TypedDict):

"""Input for pre-MCP-tool-call hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

serverName: str

toolName: str

arguments: Any

toolCallId: NotRequired[str]

_meta: NotRequired[dict[str, Any]]

class PreMcpToolCallHookOutput(TypedDict, total=False):

"""Output for pre-MCP-tool-call hook.

metaToUse semantics:

- Key absent: preserve the current request _meta

- Key present with None value: omit _meta from the request

- Key present with dict value: use this dict as request _meta

"""

metaToUse: dict[str, Any] | None

PreMcpToolCallHandler = Callable[

[PreMcpToolCallHookInput, dict[str, str]],

PreMcpToolCallHookOutput | None | Awaitable[PreMcpToolCallHookOutput | None],

]

class PostToolUseHookInput(TypedDict):

"""Input for post-tool-use hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

toolName: str

toolArgs: Any

toolResult: Any

class PostToolUseHookOutput(TypedDict, total=False):

"""Output for post-tool-use hook"""

modifiedResult: Any

additionalContext: str

suppressOutput: bool

PostToolUseHandler = Callable[

[PostToolUseHookInput, dict[str, str]],

PostToolUseHookOutput | None | Awaitable[PostToolUseHookOutput | None],

]

class PostToolUseFailureHookInput(TypedDict):

"""Input for post-tool-use-failure hook.

Fires after a tool execution whose result was ``"failure"``. The CLI

extracts the failure message from the tool result and passes it as the

``error`` field (rather than passing the full result object).

"""

sessionId: str

timestamp: datetime

workingDirectory: str

toolName: str

toolArgs: Any

error: str

class PostToolUseFailureHookOutput(TypedDict, total=False):

"""Output for post-tool-use-failure hook.

Only ``additionalContext`` is consumed by the host CLI — it is appended

as hidden guidance to the model alongside the failed tool result.

"""

additionalContext: str

PostToolUseFailureHandler = Callable[

[PostToolUseFailureHookInput, dict[str, str]],

PostToolUseFailureHookOutput | None | Awaitable[PostToolUseFailureHookOutput | None],

]

class UserPromptSubmittedHookInput(TypedDict):

"""Input for user-prompt-submitted hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

prompt: str

class UserPromptSubmittedHookOutput(TypedDict, total=False):

"""Output for user-prompt-submitted hook"""

modifiedPrompt: str

additionalContext: str

suppressOutput: bool

UserPromptSubmittedHandler = Callable[

[UserPromptSubmittedHookInput, dict[str, str]],

UserPromptSubmittedHookOutput | None | Awaitable[UserPromptSubmittedHookOutput | None],

]

class SessionStartHookInput(TypedDict):

"""Input for session-start hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

source: Literal["startup", "resume", "new"]

initialPrompt: NotRequired[str]

class SessionStartHookOutput(TypedDict, total=False):

"""Output for session-start hook"""

additionalContext: str

modifiedConfig: dict[str, Any]

SessionStartHandler = Callable[

[SessionStartHookInput, dict[str, str]],

SessionStartHookOutput | None | Awaitable[SessionStartHookOutput | None],

]

class SessionEndHookInput(TypedDict):

"""Input for session-end hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

reason: Literal["complete", "error", "abort", "timeout", "user_exit"]

finalMessage: NotRequired[str]

error: NotRequired[str]

class SessionEndHookOutput(TypedDict, total=False):

"""Output for session-end hook"""

suppressOutput: bool

cleanupActions: list[str]

sessionSummary: str

SessionEndHandler = Callable[

[SessionEndHookInput, dict[str, str]],

SessionEndHookOutput | None | Awaitable[SessionEndHookOutput | None],

]

class ErrorOccurredHookInput(TypedDict):

"""Input for error-occurred hook"""

sessionId: str

timestamp: datetime

workingDirectory: str

error: str

errorContext: Literal["model_call", "tool_execution", "system", "user_input"]

recoverable: bool

class ErrorOccurredHookOutput(TypedDict, total=False):

"""Output for error-occurred hook"""

suppressOutput: bool

errorHandling: Literal["retry", "skip", "abort"]

retryCount: int

userNotification: str

ErrorOccurredHandler = Callable[

[ErrorOccurredHookInput, dict[str, str]],

ErrorOccurredHookOutput | None | Awaitable[ErrorOccurredHookOutput | None],

]

class SessionHooks(TypedDict, total=False):

"""Configuration for session hooks"""

on_pre_tool_use: PreToolUseHandler

on_pre_mcp_tool_call: PreMcpToolCallHandler

on_post_tool_use: PostToolUseHandler

on_post_tool_use_failure: PostToolUseFailureHandler

on_user_prompt_submitted: UserPromptSubmittedHandler

on_session_start: SessionStartHandler

on_session_end: SessionEndHandler

on_error_occurred: ErrorOccurredHandler

# ============================================================================

# MCP Server Configuration Types

# ============================================================================

class MCPStdioServerConfig(TypedDict, total=False):

"""Configuration for a local/stdio MCP server."""

tools: list[str] # List of tools to include. [] means none. "*" means all.

type: NotRequired[Literal["local", "stdio"]] # Server type

timeout: NotRequired[int] # Timeout in milliseconds

command: str # Command to run

args: NotRequired[list[str]] # Command arguments

env: NotRequired[dict[str, str]] # Environment variables

working_directory: NotRequired[str] # Working directory

class MCPHTTPServerConfig(TypedDict, total=False):

"""Configuration for a remote MCP server (HTTP or SSE)."""

tools: list[str] # List of tools to include. [] means none. "*" means all.

type: Literal["http", "sse"] # Server type

timeout: NotRequired[int] # Timeout in milliseconds

url: str # URL of the remote server

headers: NotRequired[dict[str, str]] # HTTP headers

MCPServerConfig = MCPStdioServerConfig | MCPHTTPServerConfig

# ============================================================================

# Custom Agent Configuration Types

# ============================================================================

class CustomAgentConfig(TypedDict, total=False):

"""Configuration for a custom agent."""

name: str # Unique name of the custom agent

display_name: NotRequired[str] # Display name for UI purposes

description: NotRequired[str] # Description of what the agent does

# List of tool names the agent can use

tools: NotRequired[list[str] | None]

prompt: str # The prompt content for the agent

# MCP servers specific to agent

mcp_servers: NotRequired[dict[str, MCPServerConfig]]

infer: NotRequired[bool] # Whether agent is available for model inference

# Skill names to preload into this agent's context at startup (opt-in; omit for none)

skills: NotRequired[list[str]]

# Model identifier (e.g. "claude-haiku-4.5"); runtime falls back to parent model if unavailable

model: NotRequired[str]

class DefaultAgentConfig(TypedDict, total=False):

"""Configuration for the default agent.

The default agent is the built-in agent that handles turns

when no custom agent is selected.

"""

# List of tool names to exclude from the default agent.

# These tools remain available to custom sub-agents that reference them.

excluded_tools: list[str]

class InfiniteSessionConfig(TypedDict, total=False):

"""

Configuration for infinite sessions with automatic context compaction

and workspace persistence.

When enabled, sessions automatically manage context window limits through

background compaction and persist state to a workspace directory.

"""

# Whether infinite sessions are enabled (default: True)

enabled: bool

# Context utilization threshold (0.0-1.0) at which background compaction starts.

# Compaction runs asynchronously, allowing the session to continue processing.

# Default: 0.80

background_compaction_threshold: float

# Context utilization threshold (0.0-1.0) at which the session blocks until

# compaction completes. This prevents context overflow when compaction hasn't

# finished in time. Default: 0.95

buffer_exhaustion_threshold: float

class LargeToolOutputConfig(TypedDict, total=False):

"""

Configuration for handling large tool outputs.

When a tool produces output exceeding the configured size, the output is

written to a temp file and a reference is returned to the model instead of

the full payload.

"""

# Whether large output handling is enabled. Default True.

enabled: bool

# Maximum size in bytes before output is written to a temp file. Default 50KB.

max_size_bytes: int

# Directory to write temp files to. Defaults to the OS temp directory.

output_directory: str

# ============================================================================

# Session Configuration

# ============================================================================

class AzureProviderOptions(TypedDict, total=False):

"""Azure-specific provider configuration"""

api_version: str # Azure API version. Defaults to "2024-10-21".

class ProviderConfig(TypedDict, total=False):

"""Configuration for a custom API provider"""

type: Literal["openai", "azure", "anthropic"]