fix: migrate http clients to httpx2

livekit-agents/livekit/agents/cli/log.py

@@ -16,6 +16,8 @@
 NOISY_LOGGERS = [
     "httpx",
     "httpcore",
+    "httpx2",
+    "httpcore2",
     "openai",
     "watchfiles",
     "anthropic",

livekit-agents/livekit/agents/inference/llm.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from typing import Any, Literal, cast
 
-import httpx
+import httpx2
 import openai
 from openai.types.chat import (
     ChatCompletionChunk,
@@ -270,10 +270,10 @@ def __init__(
             api_key=create_access_token(self._opts.api_key, self._opts.api_secret),
             base_url=self._opts.base_url,
             max_retries=0,
-            http_client=httpx.AsyncClient(
-                timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
+            http_client=openai.DefaultAsyncHttpx2Client(
+                timeout=httpx2.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
                 follow_redirects=True,
-                limits=httpx.Limits(
+                limits=httpx2.Limits(
                     max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
                 ),
             ),
@@ -451,7 +451,7 @@ async def _run(self) -> None:
                 model=self._model,
                 stream_options={"include_usage": True},
                 stream=True,
-                timeout=httpx.Timeout(self._conn_options.timeout),
+                timeout=self._conn_options.timeout,
                 **self._extra_kwargs,
             )
 
@@ -486,9 +486,9 @@ async def _run(self) -> None:
 
         except openai.APITimeoutError:
             raise APITimeoutError(retryable=retryable) from None
-        except httpx.TimeoutException as e:
+        except httpx2.TimeoutException as e:
             # Only the request call runs inside the openai client's error mapping, so a
-            # timeout waiting on the stream body arrives as the raw httpx exception.
+            # timeout waiting on the stream body arrives as the raw httpx2 exception.
             raise APITimeoutError(retryable=retryable) from e
         except openai.APIStatusError as e:
             if e.status_code == 429:

livekit-agents/livekit/agents/llm/mcp.py

@@ -1,30 +1,28 @@
-# mypy: disable-error-code=unused-ignore
-
 from __future__ import annotations
 
 import asyncio
 import json
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
-from datetime import timedelta
 from pathlib import Path
 from typing import Any, Literal
 from urllib.parse import urlparse
 
-from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+import httpx2
 from typing_extensions import Self, TypedDict
 
 from ..log import logger
+from ..utils import httpx_compat
 
 try:
-    import httpx
     import mcp.types
     from mcp import ClientSession, stdio_client
     from mcp.client.sse import sse_client
     from mcp.client.stdio import StdioServerParameters
-    from mcp.client.streamable_http import GetSessionIdCallback, streamable_http_client
+    from mcp.client.streamable_http import streamable_http_client
+    from mcp.shared._stream_protocols import ReadStream, WriteStream
     from mcp.shared.message import SessionMessage
 except ImportError as e:
     raise ImportError(
@@ -155,9 +153,7 @@ async def _run_client(self, ready_fut: asyncio.Future[None]) -> None:
                 async with ClientSession(
                     receive_stream,
                     send_stream,
-                    read_timeout_seconds=timedelta(seconds=self._read_timeout)
-                    if self._read_timeout
-                    else None,
+                    read_timeout_seconds=self._read_timeout or None,
                 ) as client:
                     await client.initialize()
                     self._client = client
@@ -196,7 +192,7 @@ async def list_tools(
             self._make_function_tool(
                 tool.name,
                 tool.description,
-                tool.inputSchema,
+                tool.input_schema,
                 tool.meta,
                 options=_resolve_tool_options(options.get(tool.name)),
             )
@@ -215,7 +211,7 @@ def _make_function_tool(
         async def _resolve(
             tool_result: mcp.types.CallToolResult, raw_arguments: dict[str, Any]
         ) -> Any:
-            if tool_result.isError:
+            if tool_result.is_error:
                 error_str = "\n".join(
                     part.text if hasattr(part, "text") else str(part)
                     for part in tool_result.content
@@ -306,13 +302,8 @@ def client_streams(
         self,
     ) -> AbstractAsyncContextManager[
         tuple[
-            MemoryObjectReceiveStream[SessionMessage | Exception],
-            MemoryObjectSendStream[SessionMessage],
-        ]
-        | tuple[
-            MemoryObjectReceiveStream[SessionMessage | Exception],
-            MemoryObjectSendStream[SessionMessage],
-            GetSessionIdCallback,
+            ReadStream[SessionMessage | Exception],
+            WriteStream[SessionMessage],
         ]
     ]: ...
 
@@ -345,7 +336,7 @@ def __init__(
         url: str,
         transport_type: Literal["sse", "streamable_http"] | None = None,
         allowed_tools: list[str] | None = None,
-        headers: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
         timeout: float = 5,
         sse_read_timeout: float = 60 * 5,
         client_session_timeout_seconds: float = 5,
@@ -373,35 +364,35 @@ def __init__(
             # Fall back to URL-based detection for backward compatibility
             self._use_streamable_http = self._should_use_streamable_http(url)
 
-        self._http_client: httpx.AsyncClient | None = None
+        self._http_client: httpx2.AsyncClient | None = None
 
     @property
-    def headers(self) -> dict[str, Any]:
+    def headers(self) -> dict[str, str]:
         return self._headers
 
     @headers.setter
-    def headers(self, headers: dict[str, Any]) -> None:
+    def headers(self, headers: dict[str, str]) -> None:
         self._headers = headers
         if self._http_client is not None:
             self._http_client.headers = headers
 
     def _create_http_client(
         self,
-        headers: dict[str, Any] | None = None,
-        timeout: httpx.Timeout | None = None,
-        auth: httpx.Auth | None = None,
-    ) -> httpx.AsyncClient:
+        headers: dict[str, str] | None = None,
+        timeout: httpx_compat.HTTPXTimeout | None = None,
+        auth: httpx2.Auth | None = None,
+    ) -> httpx2.AsyncClient:
         # ported from mcp.shared._httpx_utils.create_mcp_http_client
         kwargs: dict[str, Any] = {
             "follow_redirects": True,
-            "timeout": timeout
+            "timeout": httpx_compat.to_httpx2_timeout(timeout)
             if timeout is not None
-            else httpx.Timeout(self._timeout, read=self._sse_read_timeout),
+            else httpx2.Timeout(self._timeout, read=self._sse_read_timeout),
             "headers": headers if headers is not None else self._headers,
         }
         if auth is not None:
             kwargs["auth"] = auth
-        self._http_client = httpx.AsyncClient(**kwargs)
+        self._http_client = httpx2.AsyncClient(**kwargs)
         return self._http_client
 
     def _should_use_streamable_http(self, url: str) -> bool:
@@ -419,28 +410,29 @@ def client_streams(
         self,
     ) -> AbstractAsyncContextManager[
         tuple[
-            MemoryObjectReceiveStream[SessionMessage | Exception],
-            MemoryObjectSendStream[SessionMessage],
-        ]
-        | tuple[
-            MemoryObjectReceiveStream[SessionMessage | Exception],
-            MemoryObjectSendStream[SessionMessage],
-            GetSessionIdCallback,
+            ReadStream[SessionMessage | Exception],
+            WriteStream[SessionMessage],
         ]
     ]:
         if self._use_streamable_http:
 
             @asynccontextmanager
-            async def _streamable_http_with_client():  # type: ignore[no-untyped-def]
+            async def _streamable_http_with_client() -> AsyncIterator[
+                tuple[
+                    ReadStream[SessionMessage | Exception],
+                    WriteStream[SessionMessage],
+                ]
+            ]:
                 async with self._create_http_client() as http_client:
                     async with streamable_http_client(
-                        url=self.url, http_client=http_client
+                        url=self.url,
+                        http_client=http_client,
                     ) as streams:
                         yield streams
 
-            return _streamable_http_with_client()  # type: ignore[return-value]
+            return _streamable_http_with_client()
         else:
-            return sse_client(  # type: ignore[no-any-return]
+            return sse_client(
                 url=self.url,
                 headers=self._headers,
                 timeout=self._timeout,
@@ -516,11 +508,11 @@ def client_streams(
         self,
     ) -> AbstractAsyncContextManager[
         tuple[
-            MemoryObjectReceiveStream[SessionMessage | Exception],
-            MemoryObjectSendStream[SessionMessage],
+            ReadStream[SessionMessage | Exception],
+            WriteStream[SessionMessage],
         ]
     ]:
-        return stdio_client(  # type: ignore[no-any-return]
+        return stdio_client(
             StdioServerParameters(command=self.command, args=self.args, env=self.env, cwd=self.cwd)
         )
 

livekit-agents/livekit/agents/utils/httpx_compat.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import warnings
+from collections.abc import Mapping
+from typing import TypeAlias
+
+import httpx
+import httpx2
+
+HTTPXTimeout: TypeAlias = httpx2.Timeout | httpx.Timeout
+HTTPXLimits: TypeAlias = httpx2.Limits | httpx.Limits
+
+LegacyTimeoutException = httpx.TimeoutException
+
+_DEPRECATION_MESSAGE = (
+    "httpx.Timeout inputs are deprecated and will no longer be supported in LiveKit Agents 2.0. "
+    "Use httpx2.Timeout instead."
+)
+
+
+def warn_on_legacy_timeout(timeout: HTTPXTimeout | None) -> None:
+    if isinstance(timeout, httpx.Timeout):
+        warnings.warn(_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=3)
+
+
+def to_httpx2_timeout(timeout: HTTPXTimeout | None) -> httpx2.Timeout | None:
+    if timeout is None or isinstance(timeout, httpx2.Timeout):
+        return timeout
+
+    return httpx2.Timeout(
+        connect=timeout.connect,
+        read=timeout.read,
+        write=timeout.write,
+        pool=timeout.pool,
+    )
+
+
+def to_legacy_timeout(timeout: HTTPXTimeout) -> httpx.Timeout:
+    if isinstance(timeout, httpx.Timeout):
+        return timeout
+
+    return httpx.Timeout(
+        connect=timeout.connect,
+        read=timeout.read,
+        write=timeout.write,
+        pool=timeout.pool,
+    )
+
+
+def legacy_async_client(
+    *,
+    timeout: HTTPXTimeout,
+    limits: HTTPXLimits,
+    headers: Mapping[str, str] | None = None,
+    follow_redirects: bool = False,
+) -> httpx.AsyncClient:
+    if isinstance(limits, httpx2.Limits):
+        resolved_limits = httpx.Limits(
+            max_connections=limits.max_connections,
+            max_keepalive_connections=limits.max_keepalive_connections,
+            keepalive_expiry=limits.keepalive_expiry,
+        )
+    else:
+        resolved_limits = limits
+
+    return httpx.AsyncClient(
+        timeout=to_legacy_timeout(timeout),
+        limits=resolved_limits,
+        headers=headers,
+        follow_redirects=follow_redirects,
+    )

livekit-agents/pyproject.toml

@@ -49,7 +49,10 @@ dependencies = [
     "opentelemetry-sdk>=1.39.0,<1.45",
     "opentelemetry-exporter-otlp>=1.39.0,<1.45",
     "prometheus-client>=0.22",
-    "openai>=2,<3",
+    # Keep legacy HTTPX for compatibility with user inputs and upstream SDKs until 2.0.
+    "httpx>=0.27,<1",
+    "httpx2>=2.7,<3",
+    "openai>=3,<4",
     "aiofiles>=24",
     "json-repair==0.60.1",
     "pyyaml>=6.0.3",
@@ -61,7 +64,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-mcp = ["mcp>=1.24.0, <2"]
+mcp = ["mcp>=2,<3"]
 codecs = ["numpy>=1.26.0"]
 images = ["pillow>=10.3.0"]
 anam = ["livekit-plugins-anam>=1.6.10"]