orxtra v0.13.0 /transport.src.orxtra.transport._transport
Updated
On this page

Transport: drives the send/resume state machine over a Provider -- streams API calls, executes tool_use blocks, suspends on suspending tools, retries transient errors, tracks liveness.

#transport.src.orxtra.transport._transport

#transport.src.orxtra.transport._transport

#_StepContext

Mutable state carried across step() calls within a single send().

#Transport

#inject_history

python
def inject_history(self, session_id: str, messages: list[dict[str, Any]]) -> None

Inject pre-existing conversation history for a session.

Used on session resume to restore conversation context from persisted transcripts. Messages should be simple role/content dicts (e.g. {"role": "user", "content": "..."}).

#_compact_history

python
def _compact_history(history: list[dict[str, Any]], max_turns: int) -> None

Trim history in-place, keeping only the last max_turns turns.

A "turn" starts at each user message. All non-user messages following a user message belong to the same turn. We count turns from the end and drop earlier ones.

#send

python
async def send(self, message: str, *, model: str, system_prompt: str, tools: list[Tool], session_id: str | None=None) -> AsyncIterator[TransportEvent]

#resume

python
async def resume(self, continuation: Continuation, _await_result: str, *, model: str, system_prompt: str, tools: list[Tool]) -> AsyncIterator[TransportEvent]

Resume from suspension.

Executes remaining tools, then continues the API loop.

#step

python
async def step(self, state: TransportState, ctx: _StepContext) -> tuple[TransportState, list[TransportEvent]]

Execute one state transition. Returns (next_state, events).

#_accumulate_usage

python
def _accumulate_usage(ctx: _StepContext, usage: Usage) -> None

Update cumulative token counters on the step context.

#_categorize_blocks

python
def _categorize_blocks(blocks: list[ContentBlock]) -> tuple[list[ContentBlock], list[ContentBlock], list[ContentBlock], list[ContentBlock]]

Split content blocks into text, thinking, tool_use, and unknown lists.

#_build_finish_events

python
def _build_finish_events(ctx: _StepContext, usage: Usage, text_blocks: list[ContentBlock]) -> list[TransportEvent]

Build StepFinish and Result events for a text-only response.

#_parse_rate_limit_headers

python
def _parse_rate_limit_headers(headers: httpx.Headers) -> RateLimit | None

Extract rate limit info from response headers (429 responses).

#_send_streaming_with_retry

python
async def _send_streaming_with_retry(self, *, url: str, headers: dict[str, str], json_body: dict[str, Any], health_threshold: float=30.0, stuck_threshold: float=120.0) -> tuple[list[TransportEvent] | None, list[TransportEvent]]

Send a streaming request and collect parsed events.

Returns (stream_events, retry_events) where stream_events is a list of events from parse_stream, or None on failure. retry_events contains any ApiRetry, Error, LivenessWarning, or StuckDetected events from the retry loop.

Search