Skip to content

Models

Native providers: OpenAI (openai:), Anthropic (anthropic:), Azure OpenAI (azure:) and Amazon Bedrock (bedrock:). Also routed by name: a table of OpenAI-compatible prefixes, such as ollama:, vllm:, lmstudio:, groq:, together:, openrouter:, deepseek:, mistral: and gemini:. Each prefix has its own default endpoint and API-key variable. Any other OpenAI-compatible endpoint works through openai-compatible:<model> with an explicit base_url=. See OpenAI-compatible providers. A model is a string — the prefix before the colon selects the provider.

For the concepts, start with Model providers.

Registry

String factory — routes "openai:gpt-4o", "anthropic:claude-sonnet-4-6", etc. to the right client.

get_model 

get_model(model_string: str, **kwargs: Any) -> ModelProtocol

Get a model from a string identifier.

Format: "provider:model_name"

Examples:

  • "openai:gpt-4o"
  • "anthropic:claude-sonnet-4-6"

Parameters:

Name Type Description Default
model_string str

Model identifier in "provider:model" format

 required
**kwargs Any

Provider-specific configuration

 {}

Returns:

Type Description
ModelProtocol

Model instance

Raises:

Type Description
ValueError

If provider is unknown or model string is invalid
Source code in .sdk/src/tulip/models/registry.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_model(model_string: str, **kwargs: Any) -> ModelProtocol:
    """
    Get a model from a string identifier.

    Format: "provider:model_name"

    Examples:
        - "openai:gpt-4o"
        - "anthropic:claude-sonnet-4-6"

    Args:
        model_string: Model identifier in "provider:model" format
        **kwargs: Provider-specific configuration

    Returns:
        Model instance

    Raises:
        ValueError: If provider is unknown or model string is invalid
    """
    if ":" not in model_string:
        raise ValueError(
            f"Model string must be 'provider:model', got: {model_string}. "
            f"Available providers: {list(_PROVIDERS.keys())}"
        )

    provider, model_id = model_string.split(":", 1)

    if provider not in _PROVIDERS:
        raise ValueError(f"Unknown provider: {provider}. Available: {list(_PROVIDERS.keys())}")

    return _PROVIDERS[provider](model_id, **kwargs)

list_providers 

list_providers() -> list[str]

List available provider prefixes.

Source code in .sdk/src/tulip/models/registry.py 
64
65
66
def list_providers() -> list[str]:
    """List available provider prefixes."""
    return list(_PROVIDERS.keys())

register_provider 

register_provider(prefix: str, factory: Callable[..., ModelProtocol]) -> None

Register a model provider.

Parameters:

Name Type Description Default
prefix str

Provider prefix (e.g., "openai", "anthropic")

 required
factory Callable[..., ModelProtocol]

Factory function that takes model name and kwargs

 required
Source code in .sdk/src/tulip/models/registry.py 
19
20
21
22
23
24
25
26
27
def register_provider(prefix: str, factory: Callable[..., ModelProtocol]) -> None:
    """
    Register a model provider.

    Args:
        prefix: Provider prefix (e.g., "openai", "anthropic")
        factory: Factory function that takes model name and kwargs
    """
    _PROVIDERS[prefix] = factory

CompatibleProvider dataclass 

CompatibleProvider(prefix: str, base_url: str | None, env_key: str, label: str, env_base_url: str | None = None, requires_key: bool = True)

One OpenAI-compatible endpoint, addressable by prefix.

Attributes:

Name Type Description
prefix str

The provider: half of a model string.
base_url str | None

Default endpoint. None means the caller must supply one (used by the generic openai-compatible prefix).
env_key str

Environment variable holding the API key.
label str

Human-readable name, used in the provider table and errors.
env_base_url str | None

Provider-conventional environment variable that overrides base_url, if the vendor defines one.
requires_key bool

When False a placeholder key is used, for local servers that do not authenticate.

tulip_base_url_env property 

tulip_base_url_env: str

The TULIP_-namespaced base-URL override for this provider.

resolve_base_url 

resolve_base_url(explicit: str | None = None) -> str | None

Resolve the endpoint: explicit > TULIP_ override > vendor env > default.

Source code in .sdk/src/tulip/models/providers/__init__.py 
86
87
88
89
90
91
92
93
def resolve_base_url(self, explicit: str | None = None) -> str | None:
    """Resolve the endpoint: explicit > TULIP_ override > vendor env > default."""
    if explicit:
        return explicit
    for var in (self.tulip_base_url_env, self.env_base_url):
        if var and (value := os.environ.get(var)):
            return value
    return self.base_url

resolve_api_key 

resolve_api_key(explicit: str | None = None) -> str | None

Resolve the API key: explicit > provider env > placeholder-or-None.

Source code in .sdk/src/tulip/models/providers/__init__.py 
 95
 96
 97
 98
 99
100
101
def resolve_api_key(self, explicit: str | None = None) -> str | None:
    """Resolve the API key: explicit > provider env > placeholder-or-None."""
    if explicit:
        return explicit
    if key := os.environ.get(self.env_key):
        return key
    return None if self.requires_key else _LOCAL_PLACEHOLDER_KEY

register_compatible_providers 

register_compatible_providers() -> None

Register every entry in :data:COMPATIBLE_PROVIDERS.

Called from the registry's default registration. Safe to call twice — registration is a dict assignment keyed by prefix.

Source code in .sdk/src/tulip/models/providers/__init__.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def register_compatible_providers() -> None:
    """Register every entry in :data:`COMPATIBLE_PROVIDERS`.

    Called from the registry's default registration. Safe to call twice —
    registration is a dict assignment keyed by prefix.
    """
    from tulip.models.registry import register_provider

    for spec in COMPATIBLE_PROVIDERS:
        # Bind ``spec`` per iteration; a bare closure would capture the loop
        # variable and give every provider the last table entry.
        def factory(model_id: str, _spec: CompatibleProvider = spec, **kw: Any) -> ModelProtocol:
            return _build(_spec, model_id, **kw)

        register_provider(spec.prefix, factory)

Base contract

Every model provider implements ModelProtocol. RequestBuilder and ResponseParser are the per-provider seams for translating between Tulip's ModelConfig / Message types and the provider's wire format.

ModelProtocol

Bases: Protocol

Protocol defining the model interface.

complete async 

complete(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> ModelResponse

Complete a chat request.

Source code in .sdk/src/tulip/models/base.py 
24
25
26
27
28
29
30
31
async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """Complete a chat request."""
    ...

stream 

stream(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncIterator[ModelChunkEvent]

Stream a chat response.

Source code in .sdk/src/tulip/models/base.py 
38
39
40
41
42
43
44
45
def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a chat response."""
    ...

ModelConfig

Bases: BaseModel

Base configuration for models.

ModelResponse

Bases: BaseModel

Response from a model completion.

content property 

content: str | None

Get response content.

tool_calls property 

tool_calls: list[Any]

Get tool calls.

prompt_tokens property 

prompt_tokens: int

Get prompt token count.

completion_tokens property 

completion_tokens: int

Get completion token count.

total_tokens property 

total_tokens: int

Get total token count.

RequestBuilder

Bases: Protocol

Protocol for building provider-specific requests.

build 

build(messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None, **kwargs: Any) -> Any

Build a provider-specific request.

Source code in .sdk/src/tulip/models/base.py 
52
53
54
55
56
57
58
59
def build(
    self,
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None,
    **kwargs: Any,
) -> Any:
    """Build a provider-specific request."""
    ...

ResponseParser

Bases: Protocol

Protocol for parsing provider-specific responses.

parse 

parse(response: Any) -> ModelResponse

Parse a provider-specific response.

Source code in .sdk/src/tulip/models/base.py 
66
67
68
def parse(self, response: Any) -> ModelResponse:
    """Parse a provider-specific response."""
    ...

OpenAI

OpenAIModel 

OpenAIModel(model: str = 'gpt-4o', api_key: str | None = None, base_url: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, **kwargs: Any)

Bases: BaseModel

OpenAI model provider.

Supports GPT-4o, GPT-4, o1, o3, gpt-5.x models with streaming and tool calling. Speaks both OpenAI wire APIs: chat-completions (the default) and the Responses API, selected via api= on the config or automatically for model families that require it (gpt-5.6-*, which reject function tools on chat-completions whenever reasoning is on).

Example

model = OpenAIModel(model="gpt-4o") response = await model.complete([Message.user("Hello!")])

Initialize OpenAI model.

Source code in .sdk/src/tulip/models/native/openai.py 
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
def __init__(
    self,
    model: str = "gpt-4o",
    api_key: str | None = None,
    base_url: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
    **kwargs: Any,
) -> None:
    """Initialize OpenAI model."""
    config = OpenAIConfig(
        model=model,
        api_key=api_key,
        base_url=base_url,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    super().__init__(config=config)

supports_structured_output property 

supports_structured_output: bool

Native response_format={"type":"json_schema",...} support.

OpenAI's chat-completions API accepts a JSON-schema response_format and guarantees a parseable instance. The agent loop uses this property to skip the prompted-JSON fallback when the provider ships native structured output.

client property 

client: AsyncOpenAI

Get or create the OpenAI client.

The client is configured with explicit max_retries and timeout from :class:OpenAIConfig so transient errors (429, 5xx, network resets) don't kill the agent loop on first try. The openai SDK retries with exponential backoff between attempts.

Bound to the event loop that built it: AsyncOpenAI wraps an httpx pool, so a client cached across two loops fails on the second with APIConnectionError: Connection error — which reads as a provider outage and sends you to check your key and their status page. See :func:~tulip.core.loop_bound.loop_bound.

close async 

close() -> None

Close the OpenAI client and release resources.

Source code in .sdk/src/tulip/models/native/openai.py 
772
773
774
775
776
async def close(self) -> None:
    """Close the OpenAI client and release resources."""
    if self._client is not None:
        await self._client.close()
        self._client = None

__aenter__ async 

__aenter__() -> OpenAIModel

Async context manager entry.

Source code in .sdk/src/tulip/models/native/openai.py 
778
779
780
async def __aenter__(self) -> OpenAIModel:
    """Async context manager entry."""
    return self

__aexit__ async 

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit - close client.

Source code in .sdk/src/tulip/models/native/openai.py 
782
783
784
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit - close client."""
    await self.close()

complete async 

complete(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> ModelResponse

Complete a chat request.

Parameters:

Name Type Description Default
messages list[Message]

Conversation history

 required
tools list[dict[str, Any]] | None

Tool schemas in OpenAI format

 None
**kwargs Any

Additional OpenAI-specific options

 {}

Returns:

Type Description
ModelResponse

Model response with message and metadata
Source code in .sdk/src/tulip/models/native/openai.py 
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """
    Complete a chat request.

    Args:
        messages: Conversation history
        tools: Tool schemas in OpenAI format
        **kwargs: Additional OpenAI-specific options

    Returns:
        Model response with message and metadata
    """
    if self._use_responses_api():
        return await self._complete_responses(messages, tools, **kwargs)

    openai_messages = self._convert_messages(messages)
    openai_tools = self._convert_tools(tools)

    uses_completion_tokens = self._uses_max_completion_tokens(self.config.model)
    rejects_sampling = self._rejects_sampling_params(self.config.model)

    max_tokens_value = kwargs.get("max_tokens")
    if max_tokens_value is None:
        max_tokens_value = self.config.max_tokens

    request_kwargs: dict[str, Any] = {
        "model": self.config.model,
        "messages": openai_messages,
    }

    # Use appropriate token parameter based on model
    if uses_completion_tokens:
        request_kwargs["max_completion_tokens"] = max_tokens_value
    else:
        request_kwargs["max_tokens"] = max_tokens_value
        if not rejects_sampling:
            self._apply_sampling(request_kwargs, kwargs)

    if openai_tools:
        request_kwargs["tools"] = openai_tools

    if self.config.seed is not None:
        request_kwargs["seed"] = self.config.seed

    if self.config.stop_sequences and not uses_completion_tokens:
        request_kwargs["stop"] = self.config.stop_sequences

    # Forward ``response_format`` for structured output. Caller is expected
    # to pass a fully-formed dict (see tulip.core.structured.build_response_format).
    response_format = kwargs.get("response_format")
    if response_format is not None:
        request_kwargs["response_format"] = response_format

    self._apply_passthrough(request_kwargs, kwargs)
    self._apply_extra_body(request_kwargs, kwargs)

    response = await self.client.chat.completions.create(**request_kwargs)
    return self._parse_response(response)

ainvoke async 

ainvoke(messages: list[Any], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> Any

LangChain-compatible alias — returns Message (AIMessage equivalent).

Source code in .sdk/src/tulip/models/native/openai.py 
1618
1619
1620
1621
1622
1623
1624
1625
1626
async def ainvoke(
    self,
    messages: list[Any],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Any:
    """LangChain-compatible alias — returns Message (AIMessage equivalent)."""
    response = await self.complete(messages, tools=tools, **kwargs)
    return response.message if hasattr(response, "message") else response

bind_tools 

bind_tools(tools: list[Any], **kwargs: Any) -> OpenAIModel

LangChain-compatible bind_tools.

Source code in .sdk/src/tulip/models/native/openai.py 
1628
1629
1630
1631
1632
1633
1634
1635
1636
def bind_tools(self, tools: list[Any], **kwargs: Any) -> OpenAIModel:
    """LangChain-compatible bind_tools."""
    bound = self.model_copy()
    object.__setattr__(
        bound,
        "_bound_tools",
        [t.to_openai_schema() if hasattr(t, "to_openai_schema") else t for t in (tools or [])],
    )
    return bound

stream async 

stream(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncIterator[ModelChunkEvent]

Stream a chat response.

With stream_reconnects (0 by default) each turn is read whole before any of it is yielded, and a turn whose connection drops part way is asked again, up to that many times. Partial output is never yielded, so the caller never sees half a turn twice. Inside an NVIDIA OpenShell box a connection is cut whenever the box's policy generation moves on (its first settings poll after start does so), and the OpenAI client retries a request that failed but never a stream that broke after it began. A refusal (an HTTP status, e.g. a guard's token_budget_exhausted) is not retried.

Source code in .sdk/src/tulip/models/native/openai.py 
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a chat response.

    With ``stream_reconnects`` (0 by default) each turn is read whole before any of it
    is yielded, and a turn whose connection drops part way is asked again, up to that
    many times. Partial output is never yielded, so the caller never sees half a turn
    twice. Inside an NVIDIA OpenShell box a connection is cut whenever the box's policy
    generation moves on (its first settings poll after start does so), and the OpenAI
    client retries a request that failed but never a stream that broke after it began.
    A refusal (an HTTP status, e.g. a guard's ``token_budget_exhausted``) is not retried.
    """
    if self.config.stream_reconnects <= 0:
        async for event in self._stream_once(messages, tools, **kwargs):
            yield event
        return
    attempt = 0
    while True:
        turn: list[ModelChunkEvent] = []
        try:
            async for event in self._stream_once(messages, tools, **kwargs):
                turn.append(event)
        except Exception as exc:
            if attempt >= self.config.stream_reconnects or not _is_disconnect(exc):
                raise
            attempt += 1
            delay = min(STREAM_RECONNECT_BASE_DELAY * (2 ** (attempt - 1)), 4.0)
            logger.warning(
                "model stream dropped (%s); asking again in %.1fs (%d/%d)",
                type(exc).__name__,
                delay,
                attempt,
                self.config.stream_reconnects,
            )
            await asyncio.sleep(delay)
            continue
        for event in turn:
            yield event
        return

OpenAIConfig

Bases: ModelConfig

Configuration for OpenAI models.

Anthropic

AnthropicModel 

AnthropicModel(model: str = 'claude-sonnet-4-6', api_key: str | None = None, base_url: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, prompt_cache: bool = False, default_headers: dict[str, str] | None = None, auth_token: str | None = None, **kwargs: Any)

Bases: BaseModel

Anthropic model provider.

Supports Claude 4.6, 4.5, 3.5 models with streaming and tool calling.

Example

model = AnthropicModel(model="claude-sonnet-4-6") response = await model.complete([Message.user("Hello!")])

Source code in .sdk/src/tulip/models/native/anthropic.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
def __init__(
    self,
    model: str = "claude-sonnet-4-6",
    api_key: str | None = None,
    base_url: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
    prompt_cache: bool = False,
    default_headers: dict[str, str] | None = None,
    auth_token: str | None = None,
    **kwargs: Any,
) -> None:
    config = AnthropicConfig(
        model=model,
        api_key=api_key,
        base_url=base_url,
        auth_token=auth_token,
        max_tokens=max_tokens,
        temperature=temperature,
        prompt_cache=prompt_cache,
        default_headers=default_headers,
        **kwargs,
    )
    super().__init__(config=config)

supports_structured_output property 

supports_structured_output: bool

Anthropic doesn't ship OpenAI-style response_format.

The agent loop falls back to the prompted-JSON path with post-hoc parsing for Anthropic models.

client property 

client: AsyncAnthropic

Get or create the Anthropic client.

Configured with explicit max_retries + timeout so a transient 529 (overloaded) / 5xx / connection reset doesn't kill the agent loop on the first try. Retries use exponential backoff inside the anthropic SDK.

base_url property 

base_url: str | None

The endpoint requests go to: the configured one, else ANTHROPIC_BASE_URL.

wire_model property 

wire_model: str

The model id as the endpoint names it.

OpenRouter's Messages API takes its own slugs, so a Claude id is translated there; an id already written as a slug is sent unchanged.

close async 

close() -> None

Close the underlying httpx client.

Agent.run_sync calls this in a finally block so the loop-bound httpx connections are shut down inside the same event loop that opened them. Without this, the next asyncio.run invocation closes the prior loop and the leftover client's __del__ later tries to aclose against it, raising RuntimeError: Event loop is closed.

Source code in .sdk/src/tulip/models/native/anthropic.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
async def close(self) -> None:
    """Close the underlying httpx client.

    ``Agent.run_sync`` calls this in a ``finally`` block so the
    loop-bound httpx connections are shut down inside the same
    event loop that opened them. Without this, the next
    ``asyncio.run`` invocation closes the prior loop and the
    leftover client's ``__del__`` later tries to ``aclose`` against
    it, raising ``RuntimeError: Event loop is closed``.
    """
    if self._client is not None:
        try:
            await self._client.close()
        finally:
            self._client = None

complete async 

complete(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> ModelResponse

Complete a chat request.

Recognises an OpenAI-style response_format={"type": "json_schema", ...} kwarg and translates it into Anthropic's tool-use mechanism: a synthetic respond_with_schema tool is appended to the call and tool_choice is pinned to it. The tool arguments are then surfaced as the message content (canonical JSON) so callers can parse them with :func:tulip.core.structured.parse_structured exactly as they would with native response_format providers.

Source code in .sdk/src/tulip/models/native/anthropic.py 
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """Complete a chat request.

    Recognises an OpenAI-style ``response_format={"type": "json_schema", ...}``
    kwarg and translates it into Anthropic's tool-use mechanism: a synthetic
    ``respond_with_schema`` tool is appended to the call and ``tool_choice``
    is pinned to it. The tool arguments are then surfaced as the message
    content (canonical JSON) so callers can parse them with
    :func:`tulip.core.structured.parse_structured` exactly as they would
    with native ``response_format`` providers.
    """
    import json as _json

    params, structured_mode = self._request_params(messages, tools, kwargs)
    response = await self.client.messages.create(**params)

    # Parse response
    content: str | None = None
    tool_calls: list[ToolCall] = []
    structured_payload: dict[str, Any] | None = None

    for block in response.content:
        if block.type == "text":
            content = (content or "") + block.text
        elif block.type == "tool_use":
            if structured_mode and block.name == self._STRUCTURED_TOOL_NAME:
                structured_payload = block.input if isinstance(block.input, dict) else {}
                continue
            tool_calls.append(
                ToolCall(
                    id=block.id,
                    name=block.name,
                    arguments=block.input if isinstance(block.input, dict) else {},
                )
            )

    # In structured mode, surface the tool's arguments as the message
    # content so downstream ``parse_structured`` can validate it.
    if structured_mode and structured_payload is not None:
        content = _json.dumps(structured_payload)

    usage: dict[str, int] = _usage_dict(response.usage) if response.usage else {}

    return ModelResponse(
        message=Message.assistant(content=content, tool_calls=tool_calls),
        usage=usage,
        stop_reason=response.stop_reason,
        cost_usd=_reported_cost(response.usage) if response.usage else None,
    )

stream async 

stream(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncIterator[ModelChunkEvent]

Stream a chat response.

Sends the same request :meth:complete does — temperature, prompt caching, native-tool betas and response_format included — and reports the same usage, cache counters included.

Source code in .sdk/src/tulip/models/native/anthropic.py 
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a chat response.

    Sends the same request :meth:`complete` does — temperature, prompt
    caching, native-tool betas and ``response_format`` included — and
    reports the same usage, cache counters included.
    """
    import json as _json

    params, structured_mode = self._request_params(messages, tools, kwargs)

    async with self.client.messages.stream(**params) as stream:
        async for text in stream.text_stream:
            yield ModelChunkEvent(content=text)

        # ``text_stream`` yields text and nothing else — tool_use blocks,
        # usage and the stop reason live on the assembled message. Without
        # reading it, a streaming tool-using agent silently loses every
        # tool call and cannot be metered.
        final = await stream.get_final_message()

    # ``final.content`` is a union of block types (text, thinking, tool_use,
    # …); only tool_use carries id/name/input, so read them defensively
    # rather than narrowing against a shape that grows with the API.
    tool_calls: list[ToolCall] = []
    for block in final.content or []:
        if getattr(block, "type", None) != "tool_use":
            continue
        block_input = getattr(block, "input", None)
        arguments = block_input if isinstance(block_input, dict) else {}
        name = str(getattr(block, "name", "") or "")
        if structured_mode and name == self._STRUCTURED_TOOL_NAME:
            # Same contract as complete(): the schema tool's arguments
            # are the answer, delivered as content.
            yield ModelChunkEvent(content=_json.dumps(arguments))
            continue
        tool_calls.append(
            ToolCall(
                id=str(getattr(block, "id", "") or ""),
                name=name,
                arguments=arguments,
            )
        )
    if tool_calls:
        yield ModelChunkEvent(tool_calls=tool_calls)

    usage: dict[str, int] | None = None
    if final.usage is not None:
        usage = _usage_dict(final.usage)

    yield ModelChunkEvent(
        done=True,
        usage=usage,
        stop_reason=final.stop_reason,
        cost_usd=_reported_cost(final.usage) if final.usage is not None else None,
    )

AnthropicConfig

Bases: ModelConfig

Configuration for Anthropic models.

Azure OpenAI

See Azure OpenAI.

AzureOpenAIModel 

AzureOpenAIModel(model: str, endpoint: str | None = None, api_version: str | None = None, api_key: str | None = None, **kwargs: Any)

Bases: OpenAIModel

Azure OpenAI provider.

Example

model = AzureOpenAIModel(model="gpt4o-prod") # a deployment name response = await model.complete([Message.user("Hello!")])

Source code in .sdk/src/tulip/models/native/azure.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self,
    model: str,
    endpoint: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    **kwargs: Any,
) -> None:
    config = AzureOpenAIConfig(
        model=model,
        endpoint=endpoint or os.environ.get("AZURE_OPENAI_ENDPOINT"),
        api_version=api_version
        or os.environ.get("AZURE_OPENAI_API_VERSION")
        or DEFAULT_API_VERSION,
        api_key=api_key or os.environ.get("AZURE_OPENAI_API_KEY"),
        **kwargs,
    )
    # Skip OpenAIModel.__init__, which would rebuild a plain OpenAIConfig
    # and drop the Azure fields.
    super(OpenAIModel, self).__init__(config=config)

client property 

client: AsyncOpenAI

An AsyncAzureOpenAI, which is an AsyncOpenAI subclass.

Loop-bound for the same reason as the OpenAI client: it wraps an httpx pool, and one cached across two event loops fails on the second with APIConnectionError — which reads as a provider outage rather than the caching bug it is.

supports_structured_output property 

supports_structured_output: bool

Native response_format={"type":"json_schema",...} support.

OpenAI's chat-completions API accepts a JSON-schema response_format and guarantees a parseable instance. The agent loop uses this property to skip the prompted-JSON fallback when the provider ships native structured output.

close async 

close() -> None

Close the OpenAI client and release resources.

Source code in .sdk/src/tulip/models/native/openai.py 
772
773
774
775
776
async def close(self) -> None:
    """Close the OpenAI client and release resources."""
    if self._client is not None:
        await self._client.close()
        self._client = None

__aenter__ async 

__aenter__() -> OpenAIModel

Async context manager entry.

Source code in .sdk/src/tulip/models/native/openai.py 
778
779
780
async def __aenter__(self) -> OpenAIModel:
    """Async context manager entry."""
    return self

__aexit__ async 

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit - close client.

Source code in .sdk/src/tulip/models/native/openai.py 
782
783
784
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit - close client."""
    await self.close()

complete async 

complete(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> ModelResponse

Complete a chat request.

Parameters:

Name Type Description Default
messages list[Message]

Conversation history

 required
tools list[dict[str, Any]] | None

Tool schemas in OpenAI format

 None
**kwargs Any

Additional OpenAI-specific options

 {}

Returns:

Type Description
ModelResponse

Model response with message and metadata
Source code in .sdk/src/tulip/models/native/openai.py 
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """
    Complete a chat request.

    Args:
        messages: Conversation history
        tools: Tool schemas in OpenAI format
        **kwargs: Additional OpenAI-specific options

    Returns:
        Model response with message and metadata
    """
    if self._use_responses_api():
        return await self._complete_responses(messages, tools, **kwargs)

    openai_messages = self._convert_messages(messages)
    openai_tools = self._convert_tools(tools)

    uses_completion_tokens = self._uses_max_completion_tokens(self.config.model)
    rejects_sampling = self._rejects_sampling_params(self.config.model)

    max_tokens_value = kwargs.get("max_tokens")
    if max_tokens_value is None:
        max_tokens_value = self.config.max_tokens

    request_kwargs: dict[str, Any] = {
        "model": self.config.model,
        "messages": openai_messages,
    }

    # Use appropriate token parameter based on model
    if uses_completion_tokens:
        request_kwargs["max_completion_tokens"] = max_tokens_value
    else:
        request_kwargs["max_tokens"] = max_tokens_value
        if not rejects_sampling:
            self._apply_sampling(request_kwargs, kwargs)

    if openai_tools:
        request_kwargs["tools"] = openai_tools

    if self.config.seed is not None:
        request_kwargs["seed"] = self.config.seed

    if self.config.stop_sequences and not uses_completion_tokens:
        request_kwargs["stop"] = self.config.stop_sequences

    # Forward ``response_format`` for structured output. Caller is expected
    # to pass a fully-formed dict (see tulip.core.structured.build_response_format).
    response_format = kwargs.get("response_format")
    if response_format is not None:
        request_kwargs["response_format"] = response_format

    self._apply_passthrough(request_kwargs, kwargs)
    self._apply_extra_body(request_kwargs, kwargs)

    response = await self.client.chat.completions.create(**request_kwargs)
    return self._parse_response(response)

ainvoke async 

ainvoke(messages: list[Any], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> Any

LangChain-compatible alias — returns Message (AIMessage equivalent).

Source code in .sdk/src/tulip/models/native/openai.py 
1618
1619
1620
1621
1622
1623
1624
1625
1626
async def ainvoke(
    self,
    messages: list[Any],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Any:
    """LangChain-compatible alias — returns Message (AIMessage equivalent)."""
    response = await self.complete(messages, tools=tools, **kwargs)
    return response.message if hasattr(response, "message") else response

bind_tools 

bind_tools(tools: list[Any], **kwargs: Any) -> OpenAIModel

LangChain-compatible bind_tools.

Source code in .sdk/src/tulip/models/native/openai.py 
1628
1629
1630
1631
1632
1633
1634
1635
1636
def bind_tools(self, tools: list[Any], **kwargs: Any) -> OpenAIModel:
    """LangChain-compatible bind_tools."""
    bound = self.model_copy()
    object.__setattr__(
        bound,
        "_bound_tools",
        [t.to_openai_schema() if hasattr(t, "to_openai_schema") else t for t in (tools or [])],
    )
    return bound

stream async 

stream(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncIterator[ModelChunkEvent]

Stream a chat response.

With stream_reconnects (0 by default) each turn is read whole before any of it is yielded, and a turn whose connection drops part way is asked again, up to that many times. Partial output is never yielded, so the caller never sees half a turn twice. Inside an NVIDIA OpenShell box a connection is cut whenever the box's policy generation moves on (its first settings poll after start does so), and the OpenAI client retries a request that failed but never a stream that broke after it began. A refusal (an HTTP status, e.g. a guard's token_budget_exhausted) is not retried.

Source code in .sdk/src/tulip/models/native/openai.py 
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a chat response.

    With ``stream_reconnects`` (0 by default) each turn is read whole before any of it
    is yielded, and a turn whose connection drops part way is asked again, up to that
    many times. Partial output is never yielded, so the caller never sees half a turn
    twice. Inside an NVIDIA OpenShell box a connection is cut whenever the box's policy
    generation moves on (its first settings poll after start does so), and the OpenAI
    client retries a request that failed but never a stream that broke after it began.
    A refusal (an HTTP status, e.g. a guard's ``token_budget_exhausted``) is not retried.
    """
    if self.config.stream_reconnects <= 0:
        async for event in self._stream_once(messages, tools, **kwargs):
            yield event
        return
    attempt = 0
    while True:
        turn: list[ModelChunkEvent] = []
        try:
            async for event in self._stream_once(messages, tools, **kwargs):
                turn.append(event)
        except Exception as exc:
            if attempt >= self.config.stream_reconnects or not _is_disconnect(exc):
                raise
            attempt += 1
            delay = min(STREAM_RECONNECT_BASE_DELAY * (2 ** (attempt - 1)), 4.0)
            logger.warning(
                "model stream dropped (%s); asking again in %.1fs (%d/%d)",
                type(exc).__name__,
                delay,
                attempt,
                self.config.stream_reconnects,
            )
            await asyncio.sleep(delay)
            continue
        for event in turn:
            yield event
        return

AzureOpenAIConfig

Bases: OpenAIConfig

Configuration for Azure-hosted OpenAI deployments.

Amazon Bedrock

Needs the bedrock extra (pip install "tulip-agents[bedrock]"). See Amazon Bedrock.

BedrockModel 

BedrockModel(model: str = 'us.amazon.nova-lite-v1:0', region: str | None = None, profile: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, **kwargs: Any)

Bases: BaseModel

Amazon Bedrock provider, via the Converse API.

Example

model = BedrockModel(model="us.amazon.nova-micro-v1:0") response = await model.complete([Message.user("Hello!")])

Source code in .sdk/src/tulip/models/native/bedrock.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def __init__(
    self,
    model: str = "us.amazon.nova-lite-v1:0",
    region: str | None = None,
    profile: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
    **kwargs: Any,
) -> None:
    config = BedrockConfig(
        model=model,
        region=region,
        profile=profile,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    super().__init__(config=config)

supports_structured_output property 

supports_structured_output: bool

Converse has no response_format.

It could be emulated by forcing a single tool via toolChoice, the way the Anthropic provider does — but toolChoice support varies by model on Bedrock, and a structured-output path that works on Claude and silently fails on Llama is worse than not claiming it. The agent loop falls back to prompted JSON, which every Bedrock model handles.

client property 

client: Any

The bedrock-runtime client, built once and reused.

Deliberately not wrapped in :func:tulip.core.loop_bound. That helper exists for clients whose connection pool binds to the event loop that created it — httpx, redis.asyncio, asyncpg. A botocore client is synchronous and holds no loop reference, so it survives being used from whatever thread :func:asyncio.to_thread happens to hand it to, and rebuilding it per loop would only pay the (slow) client construction cost again.

complete async 

complete(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> ModelResponse

Complete a chat request through converse.

botocore is synchronous, so the call runs on a worker thread. Without that, every Bedrock request would block the event loop for its whole duration — which for an agent means no concurrent tool execution, no streaming consumer, and no timeout firing.

Source code in .sdk/src/tulip/models/native/bedrock.py 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """Complete a chat request through ``converse``.

    botocore is synchronous, so the call runs on a worker thread. Without
    that, every Bedrock request would block the event loop for its whole
    duration — which for an agent means no concurrent tool execution, no
    streaming consumer, and no timeout firing.
    """
    params = self._params(messages, tools, **kwargs)
    response = await asyncio.to_thread(lambda: self.client.converse(**params))

    content: str | None = None
    tool_calls: list[ToolCall] = []
    for block in response.get("output", {}).get("message", {}).get("content", []):
        if "text" in block:
            content = (content or "") + block["text"]
        elif "toolUse" in block:
            use = block["toolUse"]
            raw_input = use.get("input")
            tool_calls.append(
                ToolCall(
                    id=use.get("toolUseId", ""),
                    name=use.get("name", ""),
                    arguments=raw_input if isinstance(raw_input, dict) else {},
                )
            )

    return ModelResponse(
        message=Message.assistant(content=content, tool_calls=tool_calls),
        usage=self._usage(response),
        stop_reason=response.get("stopReason"),
    )

stream async 

stream(messages: list[Message], tools: list[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncIterator[ModelChunkEvent]

Stream a response through converse_stream.

botocore's event stream is a synchronous iterator, so it is drained on a worker thread into a bounded thread-safe queue and read back with :func:asyncio.to_thread. An asyncio.Queue would be the wrong tool: its put_nowait is not safe to call from another thread, and the resulting corruption shows up as dropped or duplicated chunks under load rather than as an error.

Source code in .sdk/src/tulip/models/native/bedrock.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a response through ``converse_stream``.

    botocore's event stream is a *synchronous* iterator, so it is drained on
    a worker thread into a bounded thread-safe queue and read back with
    :func:`asyncio.to_thread`. An ``asyncio.Queue`` would be the wrong tool:
    its ``put_nowait`` is not safe to call from another thread, and the
    resulting corruption shows up as dropped or duplicated chunks under
    load rather than as an error.
    """
    params = self._params(messages, tools, **kwargs)
    events: queue.Queue[Any] = queue.Queue(maxsize=_STREAM_QUEUE_MAX)

    def pump() -> None:
        try:
            for event in self.client.converse_stream(**params)["stream"]:
                events.put(event)
        except BaseException as exc:  # noqa: BLE001 — re-raised on the consumer side
            events.put(exc)
        finally:
            events.put(_DONE)

    pump_task = asyncio.create_task(asyncio.to_thread(pump))
    # Tool arguments arrive as partial JSON strings across many deltas,
    # keyed by content-block index; they are only parseable once the block
    # closes.
    partial_tools: dict[int, dict[str, Any]] = {}
    usage: dict[str, int] = {}
    stop_reason: str | None = None

    try:
        while True:
            item = await asyncio.to_thread(events.get)
            if item is _DONE:
                break
            if isinstance(item, BaseException):
                raise item

            if "contentBlockStart" in item:
                start = item["contentBlockStart"].get("start", {})
                if "toolUse" in start:
                    index = item["contentBlockStart"].get("contentBlockIndex", 0)
                    partial_tools[index] = {
                        "id": start["toolUse"].get("toolUseId", ""),
                        "name": start["toolUse"].get("name", ""),
                        "buffer": "",
                    }

            elif "contentBlockDelta" in item:
                delta = item["contentBlockDelta"].get("delta", {})
                index = item["contentBlockDelta"].get("contentBlockIndex", 0)
                if "text" in delta:
                    yield ModelChunkEvent(content=delta["text"])
                elif "toolUse" in delta and index in partial_tools:
                    partial_tools[index]["buffer"] += delta["toolUse"].get("input", "")

            elif "contentBlockStop" in item:
                index = item["contentBlockStop"].get("contentBlockIndex", 0)
                if pending := partial_tools.pop(index, None):
                    yield ModelChunkEvent(
                        tool_calls=[
                            ToolCall(
                                id=pending["id"],
                                name=pending["name"],
                                arguments=_loads_or_empty(pending["buffer"]),
                            )
                        ]
                    )

            elif "messageStop" in item:
                stop_reason = item["messageStop"].get("stopReason")

            elif "metadata" in item:
                usage = self._usage(item["metadata"])

        yield ModelChunkEvent(done=True, usage=usage or None, stop_reason=stop_reason)
    finally:
        # A consumer that breaks out early leaves the reader thread parked
        # on a full queue. Drain what is buffered so it can finish and the
        # HTTP connection is released rather than held to timeout.
        if not pump_task.done():
            while True:
                try:
                    if events.get_nowait() is _DONE:
                        break
                except queue.Empty:  # noqa: PERF203 — drain loop, not hot path
                    break
        await asyncio.gather(pump_task, return_exceptions=True)

close async 

close() -> None

Release the underlying client.

Source code in .sdk/src/tulip/models/native/bedrock.py 
510
511
512
513
514
515
516
async def close(self) -> None:
    """Release the underlying client."""
    if self._client is not None:
        close = getattr(self._client, "close", None)
        if close is not None:
            await asyncio.to_thread(close)
        self._client = None

BedrockConfig

Bases: ModelConfig

Configuration for Bedrock models.