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 src/locus/models/registry.py

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 src/locus/models/registry.py

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 src/locus/models/registry.py

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.

locus_base_url_env property

locus_base_url_env: str

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

resolve_base_url

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

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

Source code in src/locus/models/providers/__init__.py

def resolve_base_url(self, explicit: str | None = None) -> str | None:
    """Resolve the endpoint: explicit > LOCUS_ override > vendor env > default."""
    if explicit:
        return explicit
    for var in (self.locus_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 src/locus/models/providers/__init__.py

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 src/locus/models/providers/__init__.py

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 locus.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 Locus'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 src/locus/models/base.py

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 src/locus/models/base.py

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 src/locus/models/base.py

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 src/locus/models/base.py

def parse(self, response: Any) -> ModelResponse:
    """Parse a provider-specific response."""
    ...

OCI Generative AI

Needs the oci extra (pip install "locus-sdk[oci]"). See OCI Generative AI and OCI GenAI models. get_model("oci:<model-id>") picks the transport automatically.

OpenAI-compatible transport (V1)

OCIChatCompletionsModel targets the /openai/v1 transport — real SSE streaming, day-0 support for new OpenAI / Meta / xAI / Mistral / Gemini releases. Use this for everything except Cohere R-series and Dedicated AI Cluster endpoints.

OCIChatCompletionsModel

OCIChatCompletionsModel(model: str, *, profile: str | None = None, auth_type: str | None = None, compartment_id: str | None = None, region: str = DEFAULT_OCI_GENAI_REGION, config_file: str = '~/.oci/config', base_url: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, **kwargs: Any)

Bases: OpenAIModel

OCI GenAI model accessed through the /openai/v1 endpoint.

Reuses :class:OpenAIModel for message conversion, tool handling, response parsing, and streaming. The only thing this class adds is the OCI-specific auth wiring.

Pass exactly one of profile, auth_type.

Initialize the OCI OpenAI-compat model.

Parameters:

Name	Type	Description	Default
model	str	OCI model identifier (e.g. openai.gpt-5, meta.llama-3.3-70b-instruct).	required
profile	str | None	OCI config profile name from config_file. Mutually exclusive with auth_type.	None
auth_type	str | None	"instance_principal" or "resource_principal". Mutually exclusive with profile. Requires compartment_id.	None
compartment_id	str | None	OCI compartment OCID, sent as opc-compartment-id. Auto-derived from the profile's tenancy under profile=. Must be supplied explicitly under auth_type=.	None
region	str	OCI region hosting the inference endpoint.	DEFAULT_OCI_GENAI_REGION
config_file	str	Path to the OCI config file (used with profile).	'~/.oci/config'
base_url	str | None	Override the derived endpoint URL (e.g. for a custom realm). Defaults to the OpenAI-compat URL for region.	None
max_tokens	int	Default max tokens. For o1/o3/gpt-5* this is automatically forwarded as max_completion_tokens — see :class:OpenAIModel.	4096
temperature	float	Default sampling temperature.	0.7
**kwargs	Any	Forwarded to :class:OCIChatCompletionsConfig (top_p, seed, frequency_penalty, presence_penalty, ...).	{}

Raises:

Type	Description
ValueError	If zero or both auth modes are set, if auth_type is invalid, or if auth_type is set without compartment_id.

Source code in src/locus/models/providers/oci/openai_compat.py

def __init__(
    self,
    model: str,
    *,
    profile: str | None = None,
    auth_type: str | None = None,
    compartment_id: str | None = None,
    region: str = DEFAULT_OCI_GENAI_REGION,
    config_file: str = "~/.oci/config",
    base_url: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
    **kwargs: Any,
) -> None:
    """Initialize the OCI OpenAI-compat model.

    Args:
        model: OCI model identifier (e.g. ``openai.gpt-5``,
            ``meta.llama-3.3-70b-instruct``).
        profile: OCI config profile name from ``config_file``. Mutually
            exclusive with ``auth_type``.
        auth_type: ``"instance_principal"`` or ``"resource_principal"``.
            Mutually exclusive with ``profile``. Requires
            ``compartment_id``.
        compartment_id: OCI compartment OCID, sent as
            ``opc-compartment-id``. Auto-derived from the profile's
            tenancy under ``profile=``. Must be supplied explicitly
            under ``auth_type=``.
        region: OCI region hosting the inference endpoint.
        config_file: Path to the OCI config file (used with ``profile``).
        base_url: Override the derived endpoint URL (e.g. for a custom
            realm). Defaults to the OpenAI-compat URL for ``region``.
        max_tokens: Default max tokens. For ``o1``/``o3``/``gpt-5*``
            this is automatically forwarded as
            ``max_completion_tokens`` — see :class:`OpenAIModel`.
        temperature: Default sampling temperature.
        **kwargs: Forwarded to :class:`OCIChatCompletionsConfig` (top_p, seed,
            frequency_penalty, presence_penalty, ...).

    Raises:
        ValueError: If zero or both auth modes are set, if ``auth_type``
            is invalid, or if ``auth_type`` is set without
            ``compartment_id``.
    """
    modes_set = sum(x is not None for x in (profile, auth_type))
    if modes_set != 1:
        msg = (
            "OCIChatCompletionsModel: specify exactly one auth mode. "
            "Either profile='<section_from_~/.oci/config>' (api-key path; "
            "use profile='DEFAULT' for the default section), "
            "or auth_type='instance_principal'|'resource_principal'|"
            "'security_token'|'delegation_token' together with compartment_id=. "
            f"Got profile={profile!r}, auth_type={auth_type!r}."
        )
        raise ValueError(msg)
    if auth_type is not None and auth_type not in _VALID_AUTH_TYPES:
        msg = f"auth_type must be one of {_VALID_AUTH_TYPES}, got {auth_type!r}"
        raise ValueError(msg)
    if auth_type is not None and compartment_id is None:
        msg = "compartment_id is required when auth_type= is set"
        raise ValueError(msg)

    # Pop fields we set explicitly to avoid duplicate-kwarg errors
    # when callers splat a config dict that includes the same keys.
    for explicit in (
        "model",
        "profile",
        "auth_type",
        "compartment_id",
        "region",
        "config_file",
        "base_url",
        "max_tokens",
        "temperature",
    ):
        kwargs.pop(explicit, None)

    # Resolve compartment_id with this precedence:
    #   1. explicit ``compartment_id=`` arg
    #   2. ``OCI_COMPARTMENT`` env var (matches OCIModel + the test
    #      conftest convention so a single export drives every test)
    #   3. profile's tenancy (last-resort default; may not have GenAI
    #      policy if the profile lives in a different home tenancy than
    #      the GenAI compartment, which is the MY_PROFILE / cross-tenancy case)
    if compartment_id is None:
        import os

        compartment_id = os.getenv("OCI_COMPARTMENT") or os.getenv("OCI_COMPARTMENT_ID")
    if compartment_id is None and profile is not None:
        try:
            profile_cfg = _load_profile_config(profile, config_file)
            compartment_id = profile_cfg.get("tenancy")
        except Exception:  # noqa: BLE001 — profile load may fail; keep None
            compartment_id = None

    config = OCIChatCompletionsConfig(
        model=model,
        api_key=None,
        base_url=base_url or build_oci_openai_base_url(region),
        region=region,
        profile=profile,
        auth_type=auth_type,
        compartment_id=compartment_id,
        config_file=config_file,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    # Skip OpenAIModel.__init__ — it would rebuild the config without
    # OCI fields. Go straight to the Pydantic BaseModel init.
    super(OpenAIModel, self).__init__(config=config)

client property

client: AsyncOpenAI

Build the AsyncOpenAI client wired with the OCI request signer.

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit - close client."""
    await self.close()

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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

OCIChatCompletionsConfig

Bases: OpenAIConfig

Configuration for :class:OCIChatCompletionsModel.

Inherits all OpenAI knobs from :class:OpenAIConfig. Adds region, profile, auth_type, compartment_id, and config_file for OCI auth selection.

Cohere R-series and Dedicated AI Cluster transport (SDK)

OCIModel targets the OCI SDK's /20231130/actions/v1 endpoint — required for cohere.command-r-* (Cohere's Chat API shape isn't served on the /openai/v1 transport) and for ocid1.generativeaiendpoint.… Dedicated AI Cluster endpoints.

OCIModel

OCIModel(model_id: str = 'cohere.command-r-plus', compartment_id: str | None = None, profile_name: str = 'DEFAULT', auth_type: str | OCIAuthType | None = None, config_file: str = '~/.oci/config', service_endpoint: str | None = None, region: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, profile: str | None = None, **kwargs: Any)

Bases: BaseModel

OCI GenAI model provider.

Automatically selects the appropriate provider based on model_id: - cohere.command-r-* → CohereProvider - cohere.command-a-* → GenericProvider (A series uses generic format) - meta., openai., google., xai., mistral.* → GenericProvider

Example

model = OCIModel( ... model_id="openai.gpt-5.1-chat-latest", ... profile_name="DEFAULT", ... auth_type="api_key", ... ) response = await model.complete([Message.user("Hello!")])

Initialize OCI GenAI model.

Parameters:

Name	Type	Description	Default
model_id	str	OCI model identifier (e.g., "openai.gpt-oss-20b", "cohere.command-r-plus")	'cohere.command-r-plus'
compartment_id	str | None	OCI compartment OCID (defaults to OCI_COMPARTMENT / OCI_COMPARTMENT_ID env var, then to the tenancy from profile)	None
profile_name	str	OCI config profile name from ~/.oci/config	'DEFAULT'
auth_type	str | OCIAuthType | None	Authentication type (api_key, security_token, instance_principal). When None (default), reads OCI_AUTH_TYPE from env, falling back to api_key.	None
config_file	str	Path to OCI config file	'~/.oci/config'
service_endpoint	str | None	Full OCI GenAI service endpoint URL. Takes precedence over region when both are set.	None
region	str | None	OCI region for the GenAI inference endpoint (e.g. "us-chicago-1"). When set and service_endpoint is not provided, the endpoint is built as https://inference.generativeai.{region}.oci.oraclecloud.com. Use this to target a model hosted outside your profile's home region without writing out the full URL.	None
max_tokens	int	Maximum tokens for response	4096
temperature	float	Model temperature (0.0-1.0)	0.7
profile	str | None	Alias for profile_name. Setting both at the same time (to different values) raises ValueError. Exists because profile= is the natural name users reach for — the previous behaviour silently swallowed it through **kwargs and fell through to the DEFAULT profile.	None
**kwargs	Any	Additional model parameters	{}

Source code in src/locus/models/providers/oci/__init__.py

def __init__(
    self,
    model_id: str = "cohere.command-r-plus",
    compartment_id: str | None = None,
    profile_name: str = "DEFAULT",
    auth_type: str | OCIAuthType | None = None,
    config_file: str = "~/.oci/config",
    service_endpoint: str | None = None,
    region: str | None = None,
    max_tokens: int = 4096,
    temperature: float = 0.7,
    profile: str | None = None,
    **kwargs: Any,
) -> None:
    """Initialize OCI GenAI model.

    Args:
        model_id: OCI model identifier (e.g., "openai.gpt-oss-20b", "cohere.command-r-plus")
        compartment_id: OCI compartment OCID (defaults to ``OCI_COMPARTMENT`` /
            ``OCI_COMPARTMENT_ID`` env var, then to the tenancy from profile)
        profile_name: OCI config profile name from ~/.oci/config
        auth_type: Authentication type (api_key, security_token,
            instance_principal). When ``None`` (default), reads
            ``OCI_AUTH_TYPE`` from env, falling back to ``api_key``.
        config_file: Path to OCI config file
        service_endpoint: Full OCI GenAI service endpoint URL. Takes
            precedence over ``region`` when both are set.
        region: OCI region for the GenAI inference endpoint (e.g.
            ``"us-chicago-1"``). When set and ``service_endpoint`` is
            not provided, the endpoint is built as
            ``https://inference.generativeai.{region}.oci.oraclecloud.com``.
            Use this to target a model hosted outside your profile's
            home region without writing out the full URL.
        max_tokens: Maximum tokens for response
        temperature: Model temperature (0.0-1.0)
        profile: Alias for ``profile_name``. Setting both at the same
            time (to different values) raises ``ValueError``. Exists
            because ``profile=`` is the natural name users reach for
            — the previous behaviour silently swallowed it through
            ``**kwargs`` and fell through to the ``DEFAULT`` profile.
        **kwargs: Additional model parameters
    """
    # ``profile=`` is the natural name; treat it as an alias.
    if profile is not None:
        if profile_name not in ("DEFAULT", profile):
            raise ValueError(
                "OCIModel: pass either `profile` or `profile_name`, not both "
                f"(got profile={profile!r}, profile_name={profile_name!r})."
            )
        profile_name = profile

    if auth_type is None:
        import os

        auth_type = os.getenv("OCI_AUTH_TYPE", "api_key")
    if isinstance(auth_type, str):
        auth_type = OCIAuthType(auth_type)

    # Resolve compartment_id with this precedence:
    #   1. explicit ``compartment_id=`` arg
    #   2. ``OCI_COMPARTMENT`` / ``OCI_COMPARTMENT_ID`` env var
    #   3. tenancy from the profile (handled inside OCIClient)
    if compartment_id is None:
        import os

        compartment_id = os.getenv("OCI_COMPARTMENT") or os.getenv("OCI_COMPARTMENT_ID")

    # Resolve the endpoint. Explicit ``service_endpoint`` wins; otherwise
    # build it from ``region`` so users don't have to type out the full
    # ``https://inference.generativeai.<region>.oci.oraclecloud.com`` URL
    # just to target a model hosted outside their profile's home region.
    if service_endpoint is None and region is not None:
        service_endpoint = f"https://inference.generativeai.{region}.oci.oraclecloud.com"

    config = OCIConfig(
        model_id=model_id,
        compartment_id=compartment_id,
        profile_name=profile_name,
        auth_type=auth_type,
        config_file=config_file,
        service_endpoint=service_endpoint,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    super().__init__(config=config)

supports_structured_output property

supports_structured_output: bool

OCI's native SDK transport (Cohere R-series) doesn't expose OpenAI-style response_format. Use the V1 transport (OCIChatCompletionsModel) for that.

client property

client: OCIClient

Get or create the OCI client.

provider property

provider: OCIModelProvider

Get the appropriate provider for this model.

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 (overrides bound tools if set)	None
**kwargs	Any	Additional OCI-specific options	{}

Returns:

Type	Description
ModelResponse	Model response with message and metadata

Source code in src/locus/models/providers/oci/__init__.py

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 (overrides bound tools if set)
        **kwargs: Additional OCI-specific options

    Returns:
        Model response with message and metadata
    """
    from oci.generative_ai_inference import models

    # Use explicitly-passed tools; fall back to tools bound via bind_tools()
    effective_tools = tools if tools is not None else getattr(self, "_bound_tools", None)

    # Convert messages and tools using the provider
    # Pass model_id for model-specific handling (e.g., Gemini parallel tool calls)
    converted_messages = self.provider.convert_messages(messages, self.config.model_id)
    converted_tools = self.provider.convert_tools(effective_tools)

    # Build request kwargs - remove duplicates
    request_kwargs = {
        "max_tokens": kwargs.pop("max_tokens", self.config.max_tokens),
        "temperature": kwargs.pop("temperature", self.config.temperature),
        **kwargs,
    }

    # Build the request. Pass model_id so the provider can pick the right
    # token-limit field (OpenAI wants max_completion_tokens, Meta wants
    # max_tokens, others accept either).
    request_kwargs["model_id"] = self.config.model_id
    if isinstance(converted_messages, dict):
        # Cohere returns a dict with special keys
        request_kwargs = {**converted_messages, **request_kwargs}
        chat_request = self.provider.build_request([], converted_tools, **request_kwargs)
    else:
        chat_request = self.provider.build_request(
            converted_messages,
            converted_tools,
            **request_kwargs,
        )

    # Create chat details
    chat_details = models.ChatDetails(
        compartment_id=self.client.compartment_id,
        serving_mode=self.client.get_serving_mode(self.config.model_id),
        chat_request=chat_request,
    )

    # Execute request with retry for empty responses.
    # OCI GenAI sometimes returns empty content, especially under
    # concurrent load. Retry up to 3 times with backoff.
    loop = asyncio.get_running_loop()
    max_retries = 3

    for attempt in range(max_retries):
        response = await loop.run_in_executor(
            None,
            lambda: self.client.chat(chat_details),
        )

        # Parse response
        content, tool_calls, stop_reason = self.provider.parse_response(response)
        usage = self.provider.parse_usage(response)

        # If we got content or tool calls, we're good
        if content or tool_calls:
            break

        # Backoff before retry (0.5s, 1.0s)
        if attempt < max_retries - 1:
            await asyncio.sleep(0.5 * (attempt + 1))

    return ModelResponse(
        message=Message.assistant(content=content, tool_calls=tool_calls),
        usage=usage,
        stop_reason=stop_reason,
    )

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 src/locus/models/providers/oci/__init__.py

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) -> OCIModel

LangChain-compatible bind_tools — stores tool schemas and passes them on complete().

Source code in src/locus/models/providers/oci/__init__.py

def bind_tools(self, tools: list[Any], **kwargs: Any) -> OCIModel:
    """LangChain-compatible bind_tools — stores tool schemas and passes them on complete()."""
    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 via the OCI GenAI SDK.

Sets is_stream=True on the chat request so the SDK returns an SSE event stream. Each data: event carries a JSON chunk with message.content deltas and (on the last event) finishReason. Works for both OnDemandServingMode (model id) and DedicatedServingMode (DAC endpoint OCID).

On any exception the stream falls back to the non-streaming complete() path and yields a single chunk with the full content — robust to providers that reject is_stream.

Source code in src/locus/models/providers/oci/__init__.py

async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a chat response via the OCI GenAI SDK.

    Sets ``is_stream=True`` on the chat request so the SDK returns
    an SSE event stream. Each ``data:`` event carries a JSON chunk
    with ``message.content`` deltas and (on the last event)
    ``finishReason``. Works for both ``OnDemandServingMode``
    (model id) and ``DedicatedServingMode`` (DAC endpoint OCID).

    On any exception the stream falls back to the non-streaming
    ``complete()`` path and yields a single chunk with the full
    content — robust to providers that reject ``is_stream``.
    """
    import json as _json

    from oci.generative_ai_inference import models

    # Build the same request shape as ``complete()`` but with
    # ``is_stream=True`` so the SDK returns a streaming response.
    converted_messages = self.provider.convert_messages(messages, model_id=self.config.model_id)
    converted_tools = self.provider.convert_tools(tools)
    request_kwargs = {
        "max_tokens": self.config.max_tokens,
        "temperature": self.config.temperature,
        "top_p": self.config.top_p,
        "model_id": self.config.model_id,
        "is_stream": True,
    }

    chat_request = self.provider.build_request(
        converted_messages,
        converted_tools,
        **request_kwargs,
    )
    # Some provider request builders (Cohere) take messages
    # under a different field — they may have ignored is_stream
    # in build_request. Set it on the resulting object as a
    # belt-and-braces step.
    if hasattr(chat_request, "is_stream"):
        chat_request.is_stream = True

    chat_details = models.ChatDetails(
        compartment_id=self.client.compartment_id,
        serving_mode=self.client.get_serving_mode(self.config.model_id),
        chat_request=chat_request,
    )

    loop = asyncio.get_running_loop()
    try:
        response = await loop.run_in_executor(None, lambda: self.client.chat(chat_details))
    except Exception:  # noqa: BLE001 — fall back on any provider error
        # Some DAC endpoints / model versions reject is_stream.
        # Hand the user a working stream by chunking the
        # non-streaming response.
        non_stream = await self.complete(messages, tools, **kwargs)
        if non_stream.content:
            yield ModelChunkEvent(content=non_stream.content)
        if non_stream.tool_calls:
            yield ModelChunkEvent(tool_calls=non_stream.tool_calls)
        yield ModelChunkEvent(done=True)
        return

    # ``response.data`` is the raw streaming body. Iterate the SSE
    # event stream synchronously in a worker thread so the asyncio
    # loop stays responsive — each event is a small JSON delta.
    events_iter = response.data.events()
    sentinel = object()

    def _next_event() -> Any:
        return next(events_iter, sentinel)

    while True:
        event = await loop.run_in_executor(None, _next_event)
        if event is sentinel:
            break
        data = getattr(event, "data", None)
        if not data:
            continue
        try:
            chunk = _json.loads(data)
        except (ValueError, TypeError):
            # Skip malformed deltas — keep the stream alive.
            continue
        content_delta, tool_calls_delta, _is_done = self.provider.parse_stream_chunk(chunk)
        if content_delta:
            yield ModelChunkEvent(content=content_delta)
        if tool_calls_delta:
            yield ModelChunkEvent(tool_calls=tool_calls_delta)

    yield ModelChunkEvent(done=True)

OCIConfig

Bases: ModelConfig

Configuration for OCI GenAI models.

OCIAuthType

Bases: StrEnum

OCI authentication types.

Responses API transport (opt-in)

OCIResponsesModel targets the server-stateful /v1/responses endpoint. See OCI Responses API.

OCIResponsesModel

OCIResponsesModel(model: str, *, profile: str | None = None, auth_type: str | None = None, compartment_id: str | None = None, project_ocid: str | None = None, region: str = DEFAULT_OCI_GENAI_REGION, config_file: str = '~/.oci/config', base_url: str | None = None, max_output_tokens: int = 4096, temperature: float = 0.7, request_timeout: float = 600.0, store: bool = True, **kwargs: Any)

Bases: BaseModel

OCI GenAI model via the Responses API.

Server-stateful transport — Locus's runtime loop sends only the input added since the last turn and threads previous_response_id through :attr:AgentState.provider_state so the OCI server can reattach to the existing thread.

Example

model = OCIResponsesModel( ... model="openai.gpt-5.5-pro", ... profile="LUIGI_FRA_API", ... region="us-chicago-1", ... ) agent = Agent(model=model, tools=[my_tool]) result = agent.run_sync("Plan a trip to Tokyo.")

Pass exactly one of profile, auth_type.

Initialize the OCI Responses model.

Parameters:

Name	Type	Description	Default
model	str	OCI model identifier (e.g. openai.gpt-5.5-pro).	required
profile	str | None	OCI config profile name in config_file. Mutually exclusive with auth_type.	None
auth_type	str | None	"instance_principal" or "resource_principal". Mutually exclusive with profile. Requires compartment_id.	None
compartment_id	str | None	OCI compartment OCID, sent as opc-compartment-id. Auto-derived from the profile's tenancy under profile=. Required under auth_type=.	None
project_ocid	str | None	Optional GenAI Project OCID. Locus does not require this — leave unset for the standard path. OCIProjectRequiredError is raised at request time if the model's response indicates a project is needed.	None
region	str	OCI region (default us-chicago-1).	DEFAULT_OCI_GENAI_REGION
config_file	str	Path to the OCI config file (with profile=).	'~/.oci/config'
base_url	str | None	Override the derived endpoint URL.	None
max_output_tokens	int	Default output token cap.	4096
temperature	float	Default sampling temperature.	0.7
request_timeout	float	HTTP request timeout in seconds.	600.0
store	bool	When True (default) the OCI server persists responses and the agent uses previous_response_id for cheap multi-turn continuation. Set False for tenancies with Zero Data Retention (ZDR) enabled — the server rejects previous_response_id otherwise. With store=False the model reports server_stateful=False to the agent runtime, so the agent sends the full message history each turn (like chat/completions) but still uses the Responses endpoint — useful for Responses-only models in ZDR tenants.	True

Raises:

Type	Description
ValueError	If zero or both auth modes are set, if auth_type is invalid, or if auth_type is set without compartment_id.

Source code in src/locus/models/providers/oci/responses.py

def __init__(
    self,
    model: str,
    *,
    profile: str | None = None,
    auth_type: str | None = None,
    compartment_id: str | None = None,
    project_ocid: str | None = None,
    region: str = DEFAULT_OCI_GENAI_REGION,
    config_file: str = "~/.oci/config",
    base_url: str | None = None,
    max_output_tokens: int = 4096,
    temperature: float = 0.7,
    request_timeout: float = 600.0,
    store: bool = True,
    **kwargs: Any,
) -> None:
    """Initialize the OCI Responses model.

    Args:
        model: OCI model identifier (e.g. ``openai.gpt-5.5-pro``).
        profile: OCI config profile name in ``config_file``.
            Mutually exclusive with ``auth_type``.
        auth_type: ``"instance_principal"`` or ``"resource_principal"``.
            Mutually exclusive with ``profile``. Requires
            ``compartment_id``.
        compartment_id: OCI compartment OCID, sent as
            ``opc-compartment-id``. Auto-derived from the profile's
            tenancy under ``profile=``. Required under ``auth_type=``.
        project_ocid: Optional GenAI Project OCID. Locus does **not**
            require this — leave unset for the standard path.
            ``OCIProjectRequiredError`` is raised at request time if
            the model's response indicates a project is needed.
        region: OCI region (default us-chicago-1).
        config_file: Path to the OCI config file (with ``profile=``).
        base_url: Override the derived endpoint URL.
        max_output_tokens: Default output token cap.
        temperature: Default sampling temperature.
        request_timeout: HTTP request timeout in seconds.
        store: When True (default) the OCI server persists responses
            and the agent uses ``previous_response_id`` for cheap
            multi-turn continuation. Set False for tenancies with
            Zero Data Retention (ZDR) enabled — the server rejects
            ``previous_response_id`` otherwise. With ``store=False``
            the model reports ``server_stateful=False`` to the agent
            runtime, so the agent sends the full message history
            each turn (like chat/completions) but still uses the
            Responses endpoint — useful for Responses-only models
            in ZDR tenants.

    Raises:
        ValueError: If zero or both auth modes are set, if
            ``auth_type`` is invalid, or if ``auth_type`` is set
            without ``compartment_id``.
    """
    modes_set = sum(x is not None for x in (profile, auth_type))
    if modes_set != 1:
        msg = "specify exactly one of profile=, auth_type="
        raise ValueError(msg)
    if auth_type is not None and auth_type not in _VALID_AUTH_TYPES:
        msg = f"auth_type must be one of {sorted(_VALID_AUTH_TYPES)}, got {auth_type!r}"
        raise ValueError(msg)
    if auth_type is not None and compartment_id is None:
        msg = "compartment_id is required when auth_type= is set"
        raise ValueError(msg)

    # Resolve compartment_id with same precedence as OCIChatCompletionsModel.
    if compartment_id is None:
        import os

        compartment_id = os.getenv("OCI_COMPARTMENT") or os.getenv("OCI_COMPARTMENT_ID")
    if compartment_id is None and profile is not None:
        try:
            profile_cfg = _load_profile_config(profile, config_file)
            compartment_id = profile_cfg.get("tenancy")
        except Exception:  # noqa: BLE001 — best-effort resolution
            compartment_id = None

    config = OCIResponsesConfig(
        model=model,
        region=region,
        profile=profile,
        auth_type=auth_type,
        config_file=config_file,
        compartment_id=compartment_id,
        project_ocid=project_ocid,
        store=store,
        base_url=base_url or build_oci_openai_base_url(region),
        max_output_tokens=max_output_tokens,
        temperature=temperature,
        request_timeout=request_timeout,
        **kwargs,
    )
    super().__init__(config=config)

server_stateful property

server_stateful: bool

Whether this model relies on OCI-side response persistence.

Reflects :attr:OCIResponsesConfig.store. The agent runtime reads this to decide between the latest-turn-slice + provider_state flow (True) and the stateless full-history flow (False). ZDR tenancies set store=False at construction so the agent automatically picks the stateless path and avoids OCIResponsesStateLostError on every turn.

complete async

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

Send a Responses request and return the parsed reply.

provider_state is the continuation token from :attr:AgentState.provider_state. On turn 1 it's None; on subsequent turns it carries {"previous_response_id": <id>}.

Source code in src/locus/models/providers/oci/responses.py

async def complete(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    *,
    provider_state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> ModelResponse:
    """Send a Responses request and return the parsed reply.

    ``provider_state`` is the continuation token from
    :attr:`AgentState.provider_state`. On turn 1 it's ``None``; on
    subsequent turns it carries ``{"previous_response_id": <id>}``.
    """
    previous_response_id = (
        provider_state.get("previous_response_id") if provider_state else None
    )

    # OpenAI reasoning families (gpt-5, gpt-5.5, o1, o3, o4) reject
    # ``temperature`` on the Responses endpoint. The Agent loop
    # always passes ``temperature=self.config.temperature`` as a
    # kwarg, so a "pop unless explicit" guard isn't enough — drop
    # it unconditionally on the Responses path. Callers who need
    # explicit sampling control can opt back in via the ``extra``
    # request-body escape hatch.
    kwargs.pop("temperature", None)
    body = build_request_body(
        messages,
        model=self.config.model,
        tools=tools,
        previous_response_id=previous_response_id,
        temperature=None,
        max_output_tokens=kwargs.pop("max_tokens", None)
        or kwargs.pop("max_output_tokens", self.config.max_output_tokens),
        stream=False,
        response_format=kwargs.pop("response_format", None),
        store=self.config.store,
    )

    client = self._http_client()
    try:
        resp = await client.post(
            "/responses",
            json=body,
            headers=self._extra_headers(),
        )
    except httpx.HTTPError as exc:
        msg = f"OCI Responses request failed: {exc}"
        raise RuntimeError(msg) from exc

    self._raise_for_error(resp)

    try:
        payload = resp.json()
    except ValueError as exc:
        msg = f"OCI Responses returned non-JSON body: {resp.text[:300]}"
        raise RuntimeError(msg) from exc

    message, usage, stop_reason, parsed_provider_state = parse_response(payload)
    return ModelResponse(
        message=message,
        usage=usage,
        stop_reason=stop_reason,
        provider_state=parsed_provider_state or None,
    )

stream async

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

Stream a Responses request as Locus ModelChunkEvents.

Accumulates function_call_arguments.delta events per call id and emits a single :class:ToolCall per id once arguments finish. Text deltas emit one event per chunk. done=True fires on response.completed carrying the continuation id in provider_state.

Source code in src/locus/models/providers/oci/responses.py

async def stream(
    self,
    messages: list[Message],
    tools: list[dict[str, Any]] | None = None,
    *,
    provider_state: dict[str, Any] | None = None,
    **kwargs: Any,
) -> AsyncIterator[ModelChunkEvent]:
    """Stream a Responses request as Locus ModelChunkEvents.

    Accumulates ``function_call_arguments.delta`` events per call
    id and emits a single :class:`ToolCall` per id once arguments
    finish. Text deltas emit one event per chunk. ``done=True``
    fires on ``response.completed`` carrying the continuation id
    in ``provider_state``.
    """
    previous_response_id = (
        provider_state.get("previous_response_id") if provider_state else None
    )
    # See complete() — temperature is always dropped on the
    # Responses path (reasoning models reject it).
    kwargs.pop("temperature", None)
    body = build_request_body(
        messages,
        model=self.config.model,
        tools=tools,
        previous_response_id=previous_response_id,
        temperature=None,
        max_output_tokens=kwargs.pop("max_tokens", None)
        or kwargs.pop("max_output_tokens", self.config.max_output_tokens),
        stream=True,
        response_format=kwargs.pop("response_format", None),
        store=self.config.store,
    )

    client = self._http_client()
    tool_args_buf: dict[str, str] = {}
    tool_names: dict[str, str] = {}

    async with client.stream(
        "POST", "/responses", json=body, headers=self._extra_headers()
    ) as resp:
        self._raise_for_error(resp)
        async for raw_event in _iter_sse_events(resp):
            parsed = parse_stream_event(raw_event)
            if not parsed:
                continue
            if "content" in parsed:
                yield ModelChunkEvent(content=parsed["content"])
            if "tool_calls" in parsed:
                for partial in parsed["tool_calls"]:
                    cid = partial.get("id", "")
                    delta = partial.get("arguments_delta", "")
                    tool_args_buf[cid] = tool_args_buf.get(cid, "") + delta
                    if partial.get("name"):
                        tool_names[cid] = partial["name"]
            if parsed.get("error"):
                msg = parsed["error"]
                raise RuntimeError(f"OCI Responses stream error: {msg}")
            if parsed.get("done"):
                # Emit accumulated tool calls before the done marker.
                if tool_args_buf:
                    import json

                    completed_calls: list[ToolCall] = []
                    for cid, raw in tool_args_buf.items():
                        try:
                            args = json.loads(raw) if raw else {}
                        except (ValueError, TypeError):
                            args = {}
                        completed_calls.append(
                            ToolCall(id=cid, name=tool_names.get(cid, ""), arguments=args)
                        )
                    yield ModelChunkEvent(tool_calls=completed_calls)
                yield ModelChunkEvent(done=True)

aclose async

aclose() -> None

Close the underlying httpx client.

Source code in src/locus/models/providers/oci/responses.py

async def aclose(self) -> None:
    """Close the underlying httpx client."""
    if self._client is not None:
        await self._client.aclose()
        object.__setattr__(self, "_client", None)

OCIResponsesConfig

Bases: BaseModel

Configuration for :class:OCIResponsesModel.

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 src/locus/models/native/openai.py

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:~locus.core.loop_bound.loop_bound.

close async

close() -> None

Close the OpenAI client and release resources.

Source code in src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 locus.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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/anthropic.py

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 src/locus/models/native/anthropic.py

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:locus.core.structured.parse_structured exactly as they would with native response_format providers.

Source code in src/locus/models/native/anthropic.py

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:`locus.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 src/locus/models/native/anthropic.py

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 src/locus/models/native/azure.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 locus.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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 src/locus/models/native/openai.py

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 "locus-sdk[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 src/locus/models/native/bedrock.py

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:locus.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 src/locus/models/native/bedrock.py

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 src/locus/models/native/bedrock.py

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 src/locus/models/native/bedrock.py

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.