langgraph_agent_toolkit.core.models.transport module

Own HTTP connections for OpenAI and Azure model calls.

class langgraph_agent_toolkit.core.models.transport.LLMTransportConfig(*, async_transport='httpx', max_connections=1000, max_keepalive_connections=100, keepalive_expiry=5, connect_timeout=5, read_timeout=600, write_timeout=600, pool_timeout=600, max_retries=2, shutdown_timeout=10, max_pools=32)

Bases: BaseModel

Set connection limits for each endpoint and credential configuration.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

• async_transport (Literal['httpx', 'aiohttp'])

• max_connections (Annotated[int, Gt(gt=0)])

• max_keepalive_connections (Annotated[int, Ge(ge=0)])

• keepalive_expiry (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• connect_timeout (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• read_timeout (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• write_timeout (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• pool_timeout (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• max_retries (Annotated[int, Ge(ge=0)])

• shutdown_timeout (Annotated[float, Gt(gt=0), _PydanticGeneralMetadata(allow_inf_nan=False)])

• max_pools (Annotated[int, Gt(gt=0)])

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

async_transport: Literal['httpx', 'aiohttp']

max_connections: int

max_keepalive_connections: int

keepalive_expiry: float

connect_timeout: float

read_timeout: float

write_timeout: float

pool_timeout: float

max_retries: int

shutdown_timeout: float

max_pools: int

validate_limits()

Return type:

LLMTransportConfig

class langgraph_agent_toolkit.core.models.transport.LLMTransportManager(config=None)

Bases: object

Reuse HTTP clients inside one worker and one event loop.

Enter this manager before model construction. Bind it during graph startup and each request. Close it after all model calls finish. Explicit clients and explicit socket options keep the caller’s existing behavior.

Parameters:

config (LLMTransportConfig | None)

__init__(config=None)

Parameters:

config (LLMTransportConfig | None)

Return type:

None

classmethod from_settings(settings)

Parameters:

settings (Any)

Return type:

LLMTransportManager

bind()

Return type:

Iterator[LLMTransportManager]

configure(provider, params, *, chat=True)

Add owned clients without changing credentials or explicit clients.

Parameters:

• provider (str | None)

• params (dict[str, Any])

• chat (bool)

Return type:

dict[str, Any]

async aclose()

Close all owned clients within the configured shutdown deadline.

Return type:

None

langgraph_agent_toolkit.core.models.transport.current_llm_transport_manager()

Return the manager bound to the current request.

Return type:

LLMTransportManager | None

langgraph_agent_toolkit.core.models.transport.configure_model_transport(provider, params, *, chat=True)

Parameters:

• provider (str | None)

• params (dict[str, Any])

• chat (bool)

Return type:

dict[str, Any]