cubepi.providers
AssistantMessageâ
class
Contentâ
attribute
ImageContentâ
class
Messageâ
attribute
MessageStreamâ
class
MessageStream(self)
Modelâ
class
ModelCostâ
class
OnPayloadCallbackâ
attribute
Optional callback for inspecting/replacing provider payloads before sending. Return a dict to replace the payload, or None to keep unchanged.
OnResponseCallbackâ
attribute
Optional callback invoked after an HTTP response is received.
Providerâ
class
ProviderResponseâ
class
ProviderResponse(self, status: int, headers: dict[str, str] = dict())
HTTP response metadata exposed to on_response callbacks.
StreamEventâ
class
StreamOptionsâ
class
Options bag for Provider.stream(), transparent to the agent loop.
TextContentâ
class
ThinkingBudgetsâ
class
Token budgets for each thinking level.
ThinkingContentâ
class
ThinkingLevelâ
attribute
ToolCallâ
class
ToolDefinitionâ
class
ToolResultMessageâ
class
Usageâ
class
UserMessageâ
class
adjust_max_tokens_for_thinkingâ
function
adjust_max_tokens_for_thinking(base_max_tokens: int, model_max_tokens: int, reasoning_level: ThinkingLevel, custom_budgets: ThinkingBudgets | None = None) -> tuple[int, int]
Adjust max_tokens to reserve space for a thinking budget.
Given a base max_tokens (the desired output capacity), increases it to
accommodate the thinking budget while respecting the model's hard cap.
If the model cap is too small to fit both, the thinking budget is reduced
to leave at least min_output_tokens (1024) for output.
Returns
- A
(max_tokens, thinking_budget)tuple.
FauxProviderâ
class
FauxProvider(self, *, tokens_per_second: float | None = None, token_size_min: int = 3, token_size_max: int = 5)
faux_assistant_messageâ
function
faux_assistant_message(content: str | FauxContentBlock | list[FauxContentBlock], *, stop_reason: str = 'stop', error_message: str | None = None) -> AssistantMessage
faux_textâ
function
faux_text(text: str) -> TextContent
faux_thinkingâ
function
faux_thinking(thinking: str) -> ThinkingContent
faux_tool_callâ
function
faux_tool_call(name: str, arguments: dict[str, Any], *, id: str | None = None) -> ToolCall
THINKING_LEVELSâ
attribute
clamp_thinking_levelâ
function
clamp_thinking_level(model: Model, level: ThinkingLevel) -> ThinkingLevel
Clamp level to the nearest supported level for model.
If level is already supported, return it unchanged. Otherwise search upward first (higher intensity), then downward, through the ordered level list to find the closest available level.
get_supported_thinking_levelsâ
function
get_supported_thinking_levels(model: Model) -> list[ThinkingLevel]
Return the thinking levels supported by model.
- Non-reasoning models only support
["off"]. - For reasoning models, levels are filtered through the model's
thinking_level_map. A level mapped toNoneis unsupported."xhigh"is only included when it has an explicit (non-None) mapping. All other levels are included by default when the map omits them.
models_are_equalâ
function
models_are_equal(a: Model | None, b: Model | None) -> bool
Return True if a and b refer to the same model.
Comparison is by id and provider. Returns False when either
argument is None.