LiteLLM Client

Optional Dependency

LiteLLM is an optional dependency. Install with:

pip install stirrup[litellm]  # or: uv add stirrup[litellm]

max_tokens limits provider output. context_window_tokens (required) separately tells the agent when conversation history should be summarized.

stirrup.clients.litellm_client

LiteLLM-based LLM client for multi-provider support.

This client uses LiteLLM to provide a unified interface to multiple LLM providers (OpenAI, Anthropic, Google, etc.) with automatic retries for transient failures.

Requires the litellm extra: pip install stirrup[litellm]

__all__ module-attribute

__all__ = ['LiteLLMClient']

LOGGER module-attribute

LOGGER = getLogger(__name__)

AssistantBlock

AssistantBlock = Annotated[
    TextBlock
    | ReasoningBlock
    | SignedReasoningBlock
    | RedactedReasoningBlock
    | ReasoningRefBlock
    | EncryptedReasoningBlock
    | OpaqueBlock
    | ToolCall
    | ImageContentBlock
    | VideoContentBlock
    | AudioContentBlock,
    Field(discriminator=kind),
]

One block of an assistant turn, discriminated on kind.

ChatMessage

ChatMessage = Annotated[
    SystemMessage
    | UserRoleMessage
    | AssistantMessage
    | ToolMessage,
    Field(discriminator=role),
]

Discriminated union of all message types, automatically parsed based on role field.

ReasoningEffort

ReasoningEffort = Literal[
    "none",
    "minimal",
    "low",
    "medium",
    "high",
    "xhigh",
    "default",
]

ContextOverflowError

Bases: Exception

Raised when request input exceeds the model's context capacity.

OutputTokenLimitError

OutputTokenLimitError(
    *,
    model_slug: str,
    max_tokens: int,
    provider_reason: str,
)

Bases: Exception

Raised when a provider exhausts the configured response-token budget.

Source code in src/stirrup/core/exceptions.py

def __init__(self, *, model_slug: str, max_tokens: int, provider_reason: str) -> None:
    self.model_slug = model_slug
    self.max_tokens = max_tokens
    self.provider_reason = provider_reason
    super().__init__(
        f"Model '{model_slug}' exhausted its configured output budget "
        f"(max_tokens={max_tokens}; provider reason: {provider_reason}). "
        "Increase max_tokens or ask the model for a shorter response."
    )

AssistantMessage

Bases: BaseModel

LLM response message: an ordered sequence of assistant blocks.

blocks is the only stored content. The channel-era content and tool_calls attributes remain deprecated views; reasoning raises because an ordered reasoning block sequence has no faithful channel-shaped projection. Serialized v0.1 payloads upgrade to blocks during validation. Channel-shaped construction is not part of the v0.2 API; new code constructs blocks directly. Mixing blocks with non-empty legacy channel keys raises.

provider_response_id class-attribute instance-attribute

provider_response_id: str | None = None

Provider-attached continuation state, e.g. an OpenAI Responses resp_... id.

This is turn metadata rather than emitted assistant content, so it lives beside blocks instead of inside their emission order. It is distinct from id (Stirrup's message identity) and ReasoningRefBlock.id (an emitted reasoning item handle).

content property

content: list[AssistantBlock] | str

Bare text for one text block, empty text for no blocks, or the block list.

reasoning property

reasoning: Reasoning | None

Deprecated channel accessor retained only to fail with migration guidance.

tool_calls property

tool_calls: list[ToolCall]

Tool calls in emission order.

e2e_otps property

e2e_otps: float | None

End-to-end output tokens per second.

LLMClient

Bases: Protocol

Protocol defining the interface for LLM client implementations.

Any LLM client must implement this protocol to work with the Agent class. Provides text generation with tool support and model capability inspection.

ReasoningBlock

Bases: BaseModel

In-band reasoning text with no passback token.

E.g. reasoning_content on Chat Completions-compatible hosts, or -tag extraction.

RedactedReasoningBlock

Bases: BaseModel

Reasoning the provider withheld, replaced by an opaque payload.

E.g. Anthropic redacted_thinking: data must be re-emitted verbatim as a redacted_thinking block on passback. Carries no readable content.

SignedReasoningBlock

Bases: BaseModel

Reasoning bound to an opaque provider signature re-emitted verbatim on passback.

E.g. Anthropic signed thinking blocks.

TextBlock

Bases: BaseModel

One contiguous run of answer text in an assistant turn.

signature carries opaque passback state attached to this exact block, e.g. a Google thought signature emitted on a visible text part. A client that cannot re-emit the signature must reject passback rather than silently stripping it.

TokenUsage

Bases: BaseModel

Token counts for LLM usage.

Token terminology: output = reasoning + answer.

output property

output: int

Total output tokens (reasoning + answer).

total property

total: int

Total token count across input, answer, and reasoning.

__add__

__add__(other: TokenUsage) -> TokenUsage

Add two TokenUsage objects together, summing each field independently.

Source code in src/stirrup/core/models.py

def __add__(self, other: "TokenUsage") -> "TokenUsage":
    """Add two TokenUsage objects together, summing each field independently."""
    return TokenUsage(
        input=self.input + other.input,
        answer=self.answer + other.answer,
        reasoning=self.reasoning + other.reasoning,
    )

Tool

Bases: BaseModel

Tool definition with name, description, parameter schema, and executor function.

Generic over

P: Parameter model type (Pydantic BaseModel subclass, or EmptyParams for parameterless tools) M: Metadata type (should implement Addable for aggregation; use None for tools without metadata)

Tools are simple, stateless callables. For tools requiring lifecycle management (setup/teardown, resource pooling), use a ToolProvider instead.

Example with parameters

class CalcParams(BaseModel):
    expression: str

calc_tool = Tool[CalcParams, None](
    name="calc",
    description="Evaluate math",
    parameters=CalcParams,
    executor=lambda p: ToolResult(content=str(eval(p.expression))),
)

Example without parameters (uses EmptyParams by default):

time_tool = Tool[EmptyParams, None](
    name="time",
    description="Get current time",
    executor=lambda _: ToolResult(content=datetime.now().isoformat()),
)

ToolCall

Bases: BaseModel

Represents a tool invocation request from the LLM.

Also a member of the AssistantBlock union: the kind discriminator is defaulted, so legacy payloads without the key still validate anywhere ToolCall is used as a plain input, and new dumps always carry it.

Attributes:

Name	Type	Description
name	str	Name of the tool to invoke
arguments	str	JSON string containing tool parameters
tool_call_id	str	Unique identifier for tracking this tool call and its result

signature class-attribute instance-attribute

signature: str | None = None

Opaque passback state attached to this exact block, e.g. a Google thought signature.

has_provider_tool_call_id class-attribute instance-attribute

has_provider_tool_call_id: bool = True

Whether tool_call_id was present on the provider's original block.

A client may synthesize tool_call_id for internal call/result matching while retaining that it must be omitted from provider-attached passback.

from_provider classmethod

from_provider(
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self

Capture one provider call with a stable internal correlation ID.

Source code in src/stirrup/core/models.py

@classmethod
def from_provider(
    cls,
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self:
    """Capture one provider call with a stable internal correlation ID."""
    native_id = provider_id or None
    return cls(
        tool_call_id=native_id or uuid4().hex,
        has_provider_tool_call_id=native_id is not None,
        name=name,
        arguments=arguments,
        signature=signature,
    )

LiteLLMClient

LiteLLMClient(
    model: str | None = None,
    max_tokens: int = 64000,
    *,
    context_window_tokens: int,
    model_slug: str | None = None,
    api_key: str | None = None,
    reasoning_effort: ReasoningEffort | None = None,
    kwargs: dict[str, Any] | None = None,
)

Bases: LLMClient

LiteLLM-based client supporting multiple LLM providers with unified interface.

Includes automatic retries for transient failures and token usage tracking.

Initialize LiteLLM client with model configuration and capabilities.

Parameters:

Name	Type	Description	Default
model	str | None	Model identifier for LiteLLM (e.g., 'anthropic/claude-opus-5')	None
max_tokens	int	Maximum number of tokens the provider may generate	64000
context_window_tokens	int	Context capacity used to decide when Agent history should be summarized.	required
model_slug	str | None	Deprecated. Use model instead.	None
reasoning_effort	ReasoningEffort | None	Reasoning effort level for extended thinking models (e.g., 'medium', 'high')	None
kwargs	dict[str, Any] | None	Additional arguments to pass to LiteLLM completion calls. Keys that collide with the arguments this client sets (including max_tokens) raise TypeError at request time; other keys pass through unvalidated.	None

Raises:

Type	Description
ValueError	If no model is provided, context_window_tokens is not positive, or max_tokens exceeds it.

Source code in src/stirrup/clients/litellm_client.py

def __init__(
    self,
    model: str | None = None,
    max_tokens: int = 64_000,
    *,
    context_window_tokens: int,
    model_slug: str | None = None,
    api_key: str | None = None,
    reasoning_effort: ReasoningEffort | None = None,
    kwargs: dict[str, Any] | None = None,
) -> None:
    """Initialize LiteLLM client with model configuration and capabilities.

    Args:
        model: Model identifier for LiteLLM (e.g., 'anthropic/claude-opus-5')
        max_tokens: Maximum number of tokens the provider may generate
        context_window_tokens: Context capacity used to decide when Agent history
            should be summarized.
        model_slug: Deprecated. Use model instead.
        reasoning_effort: Reasoning effort level for extended thinking models (e.g., 'medium', 'high')
        kwargs: Additional arguments to pass to LiteLLM completion calls.
            Keys that collide with the arguments this client sets (including
            ``max_tokens``) raise TypeError at request time; other keys pass
            through unvalidated.

    Raises:
        ValueError: If no model is provided, ``context_window_tokens`` is not
            positive, or ``max_tokens`` exceeds it.
    """
    if model_slug is not None:
        warnings.warn(
            "The 'model_slug' parameter is deprecated. Use 'model' instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        if model is None:
            model = model_slug
    if model is None:
        raise ValueError("model is required")
    validate_token_budgets(max_tokens, context_window_tokens)

    self._model = model
    self._max_tokens = max_tokens
    self._context_window_tokens = context_window_tokens
    self._reasoning_effort: ReasoningEffort | None = reasoning_effort
    self._api_key = api_key
    self._kwargs = kwargs or {}

max_tokens property

max_tokens: int

Maximum number of tokens the provider may generate.

context_window_tokens property

context_window_tokens: int

Context capacity used by agents for history summarization.

model_slug property

model_slug: str

Model identifier used by LiteLLM.

generate async

generate(
    messages: list[ChatMessage], tools: dict[str, Tool]
) -> AssistantMessage

Generate assistant response with optional tool calls. Retries up to 3 times on timeout/connection errors.

Source code in src/stirrup/clients/litellm_client.py

@retry(
    retry=retry_if_exception_type((Timeout, APIConnectionError, RateLimitError)),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def generate(self, messages: list[ChatMessage], tools: dict[str, Tool]) -> AssistantMessage:
    """Generate assistant response with optional tool calls. Retries up to 3 times on timeout/connection errors."""
    request_start_time = perf_counter()
    try:
        r = await acompletion(
            model=self.model_slug,
            messages=to_openai_messages(messages, allow_tool_call_signatures=True),
            tools=to_openai_tools(tools) if tools else None,
            tool_choice="auto" if tools else None,
            max_tokens=self._max_tokens,
            reasoning_effort=self._reasoning_effort,
            api_key=self._api_key,
            **self._kwargs,
        )
    except ContextWindowExceededError as e:
        raise ContextOverflowError(str(e)) from e
    request_end_time = perf_counter()

    choice = r["choices"][0]

    if choice.finish_reason in ("max_tokens", "length"):
        raise OutputTokenLimitError(
            model_slug=self.model_slug,
            max_tokens=self._max_tokens,
            provider_reason=choice.finish_reason,
        )

    msg = choice["message"]

    # Chat Completions is channel-shaped on the wire — no ordering to preserve —
    # so blocks are built in canonical channel order: reasoning → text → tool calls.
    blocks: list[AssistantBlock] = list(_parse_thinking_blocks(getattr(msg, "thinking_blocks", None)))
    if not blocks and getattr(msg, "reasoning_content", None) is not None:
        blocks.append(ReasoningBlock(content=msg.reasoning_content))

    content = msg.get("content") or ""
    if content:
        blocks.append(TextBlock(text=content))

    blocks.extend(
        ToolCall.from_provider(
            provider_id=tc.get("id"),
            name=tc["function"]["name"],
            arguments=tc["function"].get("arguments", "") or "",
            signature=tc.get("provider_specific_fields", {}).get("thought_signature", None),
        )
        for tc in (msg.get("tool_calls") or [])
    )

    usage = r["usage"]
    input_tokens = usage.prompt_tokens
    reasoning_tokens = 0
    if usage.completion_tokens_details:
        reasoning_tokens = usage.completion_tokens_details.reasoning_tokens or 0
    output_tokens = usage.completion_tokens
    answer_tokens = output_tokens - reasoning_tokens

    return AssistantMessage(
        blocks=blocks,
        token_usage=TokenUsage(
            input=input_tokens,
            answer=answer_tokens,
            reasoning=reasoning_tokens,
        ),
        request_start_time=request_start_time,
        request_end_time=request_end_time,
    )

to_openai_messages

to_openai_messages(
    msgs: list[ChatMessage],
    *,
    allow_tool_call_signatures: bool = False,
) -> list[dict[str, Any]]

Convert ChatMessage list to OpenAI-compatible message dictionaries.

Handles all message types: SystemMessage, UserMessage, AssistantMessage, and ToolMessage. Preserves reasoning content and tool calls for assistant messages.

Parameters:

Name	Type	Description	Default
msgs	list[ChatMessage]	List of ChatMessage objects (System, User, Assistant, or Tool messages).	required

Returns:

Type	Description
list[dict[str, Any]]	List of message dictionaries ready for the OpenAI API.

Raises:

Type	Description
NotImplementedError	If an unsupported message type is encountered.

Source code in src/stirrup/clients/utils.py

def to_openai_messages(
    msgs: list[ChatMessage],
    *,
    allow_tool_call_signatures: bool = False,
) -> list[dict[str, Any]]:
    """Convert ChatMessage list to OpenAI-compatible message dictionaries.

    Handles all message types: SystemMessage, UserMessage, AssistantMessage,
    and ToolMessage. Preserves reasoning content and tool calls for assistant
    messages.

    Args:
        msgs: List of ChatMessage objects (System, User, Assistant, or Tool messages).

    Returns:
        List of message dictionaries ready for the OpenAI API.

    Raises:
        NotImplementedError: If an unsupported message type is encountered.
    """
    out: list[dict[str, Any]] = []
    for m in msgs:
        if isinstance(m, SystemMessage):
            out.append({"role": "system", "content": content_to_openai(m.content)})
        elif isinstance(m, UserMessage):
            out.append({"role": "user", "content": content_to_openai(m.content)})
        elif isinstance(m, AssistantMessage):
            _validate_chat_assistant_blocks(
                m.blocks,
                allow_tool_call_signatures=allow_tool_call_signatures,
            )
            # Note: message metadata is deliberately NOT sent on the wire — it is
            # integrator/user state, opaque to the framework.
            msg: dict[str, Any] = {"role": "assistant", "content": content_to_openai(_assistant_content(m.blocks))}

            rblocks = reasoning_blocks(m.blocks)
            reasoning_parts: list[str] = []
            for block in rblocks:
                if isinstance(block, RedactedReasoningBlock):
                    continue
                if block.content is not None:
                    reasoning_parts.append(block.content)
            reasoning_content = "".join(reasoning_parts)
            if reasoning_content:
                msg["reasoning_content"] = reasoning_content

            # Signed passback is per-block: one thinking entry per signed block,
            # redacted payloads re-emitted verbatim, in emission order.
            thinking_payload: list[dict[str, Any]] = []
            for block in rblocks:
                if isinstance(block, SignedReasoningBlock):
                    thinking_payload.append(
                        {"type": "thinking", "signature": block.signature, "thinking": block.content}
                    )
                elif isinstance(block, RedactedReasoningBlock):
                    thinking_payload.append({"type": "redacted_thinking", "data": block.data})
            if thinking_payload:
                msg["thinking_blocks"] = thinking_payload

            tool_calls = tool_call_blocks(m.blocks)
            if tool_calls:
                msg["tool_calls"] = []
                for tool in tool_calls:
                    tool_dict: dict[str, Any] = {
                        "id": tool.tool_call_id,
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "arguments": tool.arguments,
                        },
                    }
                    if tool.signature is not None:
                        tool_dict["provider_specific_fields"] = {
                            "thought_signature": tool.signature,
                        }
                    msg["tool_calls"].append(tool_dict)

            out.append(msg)
        elif isinstance(m, ToolMessage):
            out.append(
                {
                    "role": "tool",
                    "content": content_to_openai(m.content),
                    "tool_call_id": m.tool_call_id,
                    "name": m.name,
                }
            )
        else:
            raise NotImplementedError(f"Unsupported message type: {type(m)}")

    return out

to_openai_tools

to_openai_tools(
    tools: dict[str, Tool],
) -> list[dict[str, Any]]

Convert Tool objects to OpenAI function calling format.

Parameters:

Name	Type	Description	Default
tools	dict[str, Tool]	Dictionary mapping tool names to Tool objects.	required

Returns:

Type	Description
list[dict[str, Any]]	List of tool definitions in OpenAI's function calling format.

Example

tools = {"calculator": calculator_tool} openai_tools = to_openai_tools(tools)

Returns: [{"type": "function", "function": {"name": "calculator", ...}}]

Source code in src/stirrup/clients/utils.py

def to_openai_tools(tools: dict[str, Tool]) -> list[dict[str, Any]]:
    """Convert Tool objects to OpenAI function calling format.

    Args:
        tools: Dictionary mapping tool names to Tool objects.

    Returns:
        List of tool definitions in OpenAI's function calling format.

    Example:
        >>> tools = {"calculator": calculator_tool}
        >>> openai_tools = to_openai_tools(tools)
        >>> # Returns: [{"type": "function", "function": {"name": "calculator", ...}}]
    """
    out: list[dict[str, Any]] = []
    for t in tools.values():
        function: dict[str, Any] = {
            "name": t.name,
            "description": t.description,
        }
        if t.parameters is not EmptyParams:
            function["parameters"] = t.parameters.model_json_schema()
        tool_payload: dict[str, Any] = {
            "type": "function",
            "function": function,
        }
        out.append(tool_payload)
    return out

validate_token_budgets

validate_token_budgets(
    max_tokens: int, context_window_tokens: int
) -> None

Reject an invalid budget pair at client construction.

Raises:

Type	Description
ValueError	If context_window_tokens is not positive, or max_tokens exceeds it.

Source code in src/stirrup/clients/utils.py

def validate_token_budgets(max_tokens: int, context_window_tokens: int) -> None:
    """Reject an invalid budget pair at client construction.

    Raises:
        ValueError: If ``context_window_tokens`` is not positive, or
            ``max_tokens`` exceeds it.
    """
    if context_window_tokens <= 0:
        raise ValueError(f"context_window_tokens must be positive, got {context_window_tokens!r}")
    if max_tokens > context_window_tokens:
        raise ValueError(f"max_tokens ({max_tokens}) must not exceed context_window_tokens ({context_window_tokens})")

_parse_thinking_blocks

_parse_thinking_blocks(
    thinking_blocks: list[dict[str, Any]] | None,
) -> list[
    SignedReasoningBlock
    | RedactedReasoningBlock
    | ReasoningBlock
]

Parse LiteLLM thinking_blocks into reasoning blocks, one per entry, in order.

Source code in src/stirrup/clients/litellm_client.py

def _parse_thinking_blocks(
    thinking_blocks: list[dict[str, Any]] | None,
) -> list[SignedReasoningBlock | RedactedReasoningBlock | ReasoningBlock]:
    """Parse LiteLLM ``thinking_blocks`` into reasoning blocks, one per entry, in order."""
    parsed: list[SignedReasoningBlock | RedactedReasoningBlock | ReasoningBlock] = []
    for entry in thinking_blocks or []:
        if entry.get("type") == "redacted_thinking":
            parsed.append(RedactedReasoningBlock(data=entry["data"]))
            continue
        signature = entry.get("signature") or entry.get("thinking_signature")
        content = entry.get("thinking")
        if signature is not None:
            parsed.append(SignedReasoningBlock(signature=signature, content=content or ""))
        elif content is not None:
            parsed.append(ReasoningBlock(content=content))
        else:
            raise ValueError(f"Signature and content not found in the thinking block response: {entry!r}")
    return parsed