Spaces:

openenv
/

chess_env

Running

App Files Files Community

chess_env / src /core /llm_client.py

sergiopaniego HF Staff

Upload folder using huggingface_hub

7aca0bc verified 8 days ago

Raw History Blame Contribute Delete

22.7 kB

# SPDX-License-Identifier: BSD-3-Clause

"""LLM client abstraction for calling LLM endpoints.
Provides a generic RPC abstraction: point it at an endpoint/port, tell it the
protocol, and it works. OpenAI-compatible API is the first implementation,
covering OpenAI, vLLM, TGI, Ollama, HuggingFace Inference API, etc.
Anthropic's native API is supported via `AnthropicClient`.
Examples:
    ```python
    client = OpenAIClient("http://localhost", 8000, model="meta-llama/...")
    response = await client.complete("What is 2+2?")
    # The endpoint may carry its own port and path prefix:
    client = OpenAIClient("http://localhost:8000/v1", port=None, model="meta-llama/...")
    # Or use the factory for hosted APIs:
    client = create_llm_client("openai", model="gpt-4", api_key="sk-...")
    response = await client.complete_with_tools(messages, tools)
    ```
"""

from __future__ import annotations

import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import urlsplit, urlunsplit

from openai import AsyncOpenAI

_OPENAI_API_PREFIX = "/v1"
_MIN_PORT, _MAX_PORT = 1, 65535


def _redact_userinfo(endpoint: str) -> str:
    """Drop `user:password@` from a URL so it is safe to echo in an error."""
    scheme, sep, rest = endpoint.partition("://")
    authority, slash, tail = rest.partition("/")
    if not sep or "@" not in authority:
        return endpoint
    return f"{scheme}{sep}***@{authority.rsplit('@', 1)[-1]}{slash}{tail}"


def _join_endpoint_port(endpoint: str, port: int | None) -> str:
    """Validate an endpoint URL and combine it with an optional port.
    Only `http`/`https` URLs with a host are accepted. Credentials, query
    strings and fragments are rejected: the endpoint is logged and persisted in
    rollout metadata, and the SDKs build request URLs by appending to the path,
    which corrupts a query string. `port` is appended only when the URL does
    not name one; an explicit port that differs from the one in the URL is
    rejected rather than silently overridden. A trailing slash on the path is
    dropped. Invalid endpoints raise `ValueError`.
    """
    safe = _redact_userinfo(endpoint)
    try:
        parts = urlsplit(endpoint)
        if parts.scheme not in ("http", "https"):
            raise ValueError("expected an http:// or https:// URL")
        if not parts.hostname:
            raise ValueError("missing host")
        if parts.username is not None or parts.password is not None:
            raise ValueError("credentials in the URL are not supported, use api_key")
        if parts.query or parts.fragment:
            raise ValueError("query strings and fragments are not supported")
        if parts.netloc.endswith(":"):
            raise ValueError("empty port")
        url_port = parts.port
        for candidate in (url_port, port):
            if candidate is not None and not _MIN_PORT <= candidate <= _MAX_PORT:
                raise ValueError(
                    f"port {candidate} is out of range {_MIN_PORT}-{_MAX_PORT}"
                )
    except ValueError as exc:
        raise ValueError(f"Invalid endpoint URL {safe!r}: {exc}") from exc

    if url_port is None:
        if port is not None:
            parts = parts._replace(netloc=f"{parts.netloc}:{port}")
    elif port is not None and port != url_port:
        raise ValueError(
            f"Endpoint URL {safe!r} already specifies port {url_port}, "
            f"which conflicts with port={port}"
        )
    return urlunsplit(parts._replace(path=parts.path.rstrip("/")))


def _openai_base_url(base_url: str) -> str:
    """Append the OpenAI `/v1` prefix when `base_url` has no path.
    A URL that already names a path (`/v1`, or a gateway prefix such as
    `/openai/v1`) is used as-is, matching the OpenAI SDK convention that
    `base_url` includes the API prefix.
    """
    parts = urlsplit(base_url)
    if parts.path in ("", "/"):
        return urlunsplit(parts._replace(path=_OPENAI_API_PREFIX))
    return base_url


@dataclass
class ToolCall:
    """A single tool/function call returned by the model."""

    id: str
    name: str
    args: dict[str, Any]


@dataclass
class LLMResponse:
    """Normalized response from an LLM, with optional tool calls."""

    content: str
    tool_calls: list[ToolCall] = field(default_factory=list)

    def to_message_dict(self) -> dict[str, Any]:
        """Convert to an OpenAI-format assistant message dict."""
        msg: dict[str, Any] = {"role": "assistant", "content": self.content}
        if self.tool_calls:
            msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.name,
                        "arguments": json.dumps(tc.args),
                    },
                }
                for tc in self.tool_calls
            ]
        return msg


class LLMClient(ABC):
    """Abstract base for LLM endpoint clients.
    Subclass and implement `complete()` for your protocol.
    Args:
        endpoint (`str`):
            The `http(s)` base URL of the LLM service (e.g. "http://localhost").
            May include a port and a path (e.g. "http://localhost:8000/v1").
            Credentials, query strings and fragments are rejected.
        port (`int` or `None`):
            The port the service listens on. Appended to `endpoint` when the
            URL does not name one; must match the URL's port when both are given.
    """

    def __init__(self, endpoint: str, port: int | None):
        self.endpoint = endpoint
        self.port = port

    @abstractmethod
    async def complete(self, prompt: str, **kwargs) -> str:
        """Send a prompt, return the text response.
        Args:
            prompt (`str`):
                The user prompt to send.
            **kwargs:
                Override default parameters (temperature, max_tokens, etc.).
        Returns:
            The model's text response.
        """
        ...

    async def complete_with_tools(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        **kwargs: Any,
    ) -> LLMResponse:
        """Send messages with tool definitions, return a normalized response.
        Messages use OpenAI-format dicts (`{"role": "...", "content": "..."}`).
        Tools use MCP tool definitions; they are converted internally.
        Args:
            messages (`list[dict[str, Any]]`):
                Conversation history as OpenAI-format message dicts.
            tools (`list[dict[str, Any]]`):
                MCP tool definitions.
            **kwargs:
                Override default parameters (temperature, max_tokens, etc.).
        Returns:
            An [`LLMResponse`] with the model's text and any tool calls.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not support tool calling"
        )

    @property
    def base_url(self) -> str:
        """Base URL of the service: `endpoint` plus `port` when the URL names none."""
        return _join_endpoint_port(self.endpoint, self.port)


class OpenAIClient(LLMClient):
    """Client for OpenAI-compatible APIs.
    Works with: OpenAI, vLLM, TGI, Ollama, HuggingFace Inference API,
    or any endpoint that speaks the OpenAI chat completions format.
    Args:
        endpoint (`str`):
            The base URL (e.g. "http://localhost"). May include a port and a
            path (e.g. "http://localhost:8000/v1"). The `/v1` API prefix is
            appended when the URL has no path; a URL with a path is used as-is.
        port (`int` or `None`):
            The port number, appended when `endpoint` does not name one; must
            match the URL's port when both are given.
        model (`str`):
            Model name to pass to the API.
        api_key (`str`, *optional*):
            API key. Defaults to "not-needed" for local endpoints.
        system_prompt (`str`, *optional*):
            System message prepended to every request.
        temperature (`float`, *optional*, defaults to `0.0`):
            Default sampling temperature.
        max_tokens (`int`, *optional*, defaults to `256`):
            Default max tokens in the response.
        use_max_completion_tokens (`bool`, *optional*, defaults to `False`):
            Use max_completion_tokens instead of max_tokens. Required for newer OpenAI models
            (gpt-5-mini, o1, o3). Not supported by self-hosted OpenAI-compatible endpoints.
    """

    def __init__(
        self,
        endpoint: str,
        port: int | None,
        model: str,
        api_key: str | None = None,
        system_prompt: str | None = None,
        temperature: float = 0.0,
        max_tokens: int = 256,
        use_max_completion_tokens: bool = False,
    ):
        super().__init__(endpoint, port)
        self.model = model
        self.system_prompt = system_prompt
        self.temperature = temperature
        self.max_tokens = max_tokens
        self._tokens_param = (
            "max_completion_tokens" if use_max_completion_tokens else "max_tokens"
        )
        self._omit_temperature = use_max_completion_tokens

        self._client = AsyncOpenAI(
            base_url=_openai_base_url(self.base_url),
            api_key=api_key if api_key is not None else "not-needed",
        )

    def _chat_completion_kwargs(
        self, messages: list[dict[str, Any]], **kwargs: Any
    ) -> dict[str, Any]:
        create_kwargs: dict[str, Any] = {
            "model": self.model,
            "messages": messages,
            self._tokens_param: kwargs.get("max_tokens", self.max_tokens),
        }
        if not self._omit_temperature:
            create_kwargs["temperature"] = kwargs.get("temperature", self.temperature)
        return create_kwargs

    async def complete(self, prompt: str, **kwargs) -> str:
        """Send a chat completion request.
        Args:
            prompt (`str`):
                The user message.
            **kwargs:
                Overrides for temperature, max_tokens.
        Returns:
            The assistant's response text.
        """
        messages = []
        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})
        messages.append({"role": "user", "content": prompt})

        call_kwargs = self._chat_completion_kwargs(messages, **kwargs)
        response = await self._client.chat.completions.create(**call_kwargs)
        return response.choices[0].message.content or ""

    async def complete_with_tools(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        **kwargs: Any,
    ) -> LLMResponse:
        create_kwargs = self._chat_completion_kwargs(messages, **kwargs)
        openai_tools = _mcp_tools_to_openai(tools)
        if openai_tools:
            create_kwargs["tools"] = openai_tools

        response = await self._client.chat.completions.create(**create_kwargs)
        msg = response.choices[0].message

        tool_calls = []
        if msg.tool_calls:
            for tc in msg.tool_calls:
                tool_calls.append(
                    ToolCall(
                        id=tc.id,
                        name=tc.function.name,
                        args=json.loads(tc.function.arguments),
                    )
                )

        return LLMResponse(content=msg.content or "", tool_calls=tool_calls)


class AnthropicClient(LLMClient):
    """Client for Anthropic's Messages API.
    Requires the `anthropic` package (lazy-imported at construction time).
    Args:
        endpoint (`str`):
            The base URL (e.g. `https://api.anthropic.com`). May include a port.
        port (`int` or `None`):
            The port number, appended when `endpoint` does not name one; must
            match the URL's port when both are given.
        model (`str`):
            Model name (e.g. "claude-sonnet-4-20250514").
        api_key (`str`, *optional*):
            Anthropic API key.
        system_prompt (`str`, *optional*):
            System message prepended to every request.
        temperature (`float`, *optional*, defaults to `0.0`):
            Default sampling temperature.
        max_tokens (`int`, *optional*, defaults to `256`):
            Default max tokens in the response.
    """

    def __init__(
        self,
        endpoint: str,
        port: int | None,
        model: str,
        api_key: str | None = None,
        system_prompt: str | None = None,
        temperature: float = 0.0,
        max_tokens: int = 256,
    ):
        super().__init__(endpoint, port)
        self.model = model
        self.system_prompt = system_prompt
        self.temperature = temperature
        self.max_tokens = max_tokens

        try:
            from anthropic import AsyncAnthropic
        except ImportError as exc:
            raise ImportError(
                "AnthropicClient requires the 'anthropic' package. "
                "Install it with: pip install anthropic"
            ) from exc

        self._client = AsyncAnthropic(
            base_url=self.base_url,
            api_key=api_key if api_key is not None else "not-needed",
        )

    async def complete(self, prompt: str, **kwargs) -> str:
        create_kwargs: dict[str, Any] = {
            "model": self.model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": kwargs.get("temperature", self.temperature),
            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
        }
        if self.system_prompt:
            create_kwargs["system"] = self.system_prompt

        response = await self._client.messages.create(**create_kwargs)
        return "".join(block.text for block in response.content if block.type == "text")

    async def complete_with_tools(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
        **kwargs: Any,
    ) -> LLMResponse:
        system, anthropic_msgs = _openai_msgs_to_anthropic(messages)

        create_kwargs: dict[str, Any] = {
            "model": self.model,
            "messages": anthropic_msgs,
            "temperature": kwargs.get("temperature", self.temperature),
            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
        }
        system_text = system or self.system_prompt
        if system_text:
            create_kwargs["system"] = system_text
        anthropic_tools = _mcp_tools_to_anthropic(tools)
        if anthropic_tools:
            create_kwargs["tools"] = anthropic_tools

        response = await self._client.messages.create(**create_kwargs)

        content = ""
        tool_calls = []
        for block in response.content:
            if block.type == "text":
                content += block.text
            elif block.type == "tool_use":
                tool_calls.append(
                    ToolCall(id=block.id, name=block.name, args=block.input)
                )

        return LLMResponse(content=content, tool_calls=tool_calls)


# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------

_HOSTED_PROVIDERS: dict[str, tuple[str, int, type[LLMClient]]] = {
    "openai": ("https://api.openai.com", 443, OpenAIClient),
    "anthropic": ("https://api.anthropic.com", 443, AnthropicClient),
}

# Models that require max_completion_tokens instead of max_tokens and do not
# accept an explicit temperature parameter. Checked by prefix to cover versioned
# names such as "o1-2024-12-17" or "gpt-5-mini-2026-01-15".
_MAX_COMPLETION_TOKENS_PREFIXES: frozenset[str] = frozenset(
    {"gpt-5-mini", "o1", "o3", "o4-mini"}
)


def create_llm_client(
    provider: str,
    model: str,
    api_key: str,
    *,
    system_prompt: str | None = None,
    temperature: float = 0.0,
    max_tokens: int = 4096,
) -> LLMClient:
    """Create an LLM client for a hosted provider.
    Args:
        provider (`str`):
            Provider name ("openai" or "anthropic").
        model (`str`):
            Model identifier.
        api_key (`str`):
            API key for the provider.
        system_prompt (`str`, *optional*):
            System message prepended to every request.
        temperature (`float`, *optional*, defaults to `0.0`):
            Sampling temperature.
        max_tokens (`int`, *optional*, defaults to `4096`):
            Maximum tokens in the response.
    Returns:
        A configured [`LLMClient`] instance.
    """
    key = provider.lower()
    if key not in _HOSTED_PROVIDERS:
        raise ValueError(
            f"Unsupported provider: {provider!r}. "
            f"Supported: {sorted(_HOSTED_PROVIDERS)}"
        )
    endpoint, port, cls = _HOSTED_PROVIDERS[key]
    extra: dict[str, Any] = {}
    if cls is OpenAIClient and any(
        model.startswith(prefix) for prefix in _MAX_COMPLETION_TOKENS_PREFIXES
    ):
        extra["use_max_completion_tokens"] = True
    return cls(
        endpoint,
        port,
        model,
        api_key=api_key,
        system_prompt=system_prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        **extra,
    )


# ---------------------------------------------------------------------------
# MCP tool-schema helpers
# ---------------------------------------------------------------------------


def _clean_mcp_schema(schema: dict[str, Any]) -> dict[str, Any]:
    """Normalize an MCP tool `inputSchema` for LLM function-calling APIs."""
    if not isinstance(schema, dict):
        return {"type": "object", "properties": {}, "required": []}

    # Shallow copy to avoid mutating the caller's schema dict.
    schema = dict(schema)

    if "oneOf" in schema:
        for option in schema["oneOf"]:
            if isinstance(option, dict) and option.get("type") == "object":
                schema = option
                break
        else:
            return {"type": "object", "properties": {}, "required": []}

    if "allOf" in schema:
        merged: dict[str, Any] = {"type": "object", "properties": {}, "required": []}
        for sub in schema["allOf"]:
            if isinstance(sub, dict):
                if "properties" in sub:
                    merged["properties"].update(sub["properties"])
                if "required" in sub:
                    merged["required"].extend(sub["required"])
        schema = merged

    if "anyOf" in schema:
        for option in schema["anyOf"]:
            if isinstance(option, dict) and option.get("type") == "object":
                schema = option
                break
        else:
            return {"type": "object", "properties": {}, "required": []}

    schema.setdefault("type", "object")
    if schema.get("type") == "object" and "properties" not in schema:
        schema["properties"] = {}
    return schema


def _mcp_tools_to_openai(
    mcp_tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Convert MCP tool definitions to OpenAI function-calling format."""
    result = []
    for tool in mcp_tools:
        input_schema = tool.get(
            "inputSchema", {"type": "object", "properties": {}, "required": []}
        )
        result.append(
            {
                "type": "function",
                "function": {
                    "name": tool["name"],
                    "description": tool.get("description", ""),
                    "parameters": _clean_mcp_schema(input_schema),
                },
            }
        )
    return result


def _mcp_tools_to_anthropic(
    mcp_tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Convert MCP tool definitions to Anthropic tool format."""
    result = []
    for tool in mcp_tools:
        input_schema = tool.get(
            "inputSchema", {"type": "object", "properties": {}, "required": []}
        )
        result.append(
            {
                "name": tool["name"],
                "description": tool.get("description", ""),
                "input_schema": _clean_mcp_schema(input_schema),
            }
        )
    return result


def _openai_msgs_to_anthropic(
    messages: list[dict[str, Any]],
) -> tuple[str, list[dict[str, Any]]]:
    """Convert OpenAI-format messages to Anthropic format.
    Returns `(system_text, anthropic_messages)`. System-role messages are
    extracted and concatenated; tool-result messages are converted to
    Anthropic's `tool_result` content blocks inside user turns.
    """
    system_parts: list[str] = []
    anthropic_msgs: list[dict[str, Any]] = []

    for msg in messages:
        role = msg["role"]

        if role == "system":
            system_parts.append(msg["content"])

        elif role == "user":
            anthropic_msgs.append({"role": "user", "content": msg["content"]})

        elif role == "assistant":
            if msg.get("tool_calls"):
                content: list[dict[str, Any]] = []
                if msg.get("content"):
                    content.append({"type": "text", "text": msg["content"]})
                for tc in msg["tool_calls"]:
                    args = tc["function"]["arguments"]
                    if isinstance(args, str):
                        args = json.loads(args)
                    content.append(
                        {
                            "type": "tool_use",
                            "id": tc["id"],
                            "name": tc["function"]["name"],
                            "input": args,
                        }
                    )
                anthropic_msgs.append({"role": "assistant", "content": content})
            else:
                anthropic_msgs.append(
                    {"role": "assistant", "content": msg.get("content", "")}
                )

        elif role == "tool":
            tool_result = {
                "type": "tool_result",
                "tool_use_id": msg["tool_call_id"],
                "content": msg["content"],
            }
            # Anthropic requires tool results in user turns; merge if possible.
            if (
                anthropic_msgs
                and anthropic_msgs[-1]["role"] == "user"
                and isinstance(anthropic_msgs[-1]["content"], list)
            ):
                anthropic_msgs[-1]["content"].append(tool_result)
            else:
                anthropic_msgs.append({"role": "user", "content": [tool_result]})

    system = "\n\n".join(system_parts)
    return system, anthropic_msgs

	# SPDX-License-Identifier: BSD-3-Clause

	"""LLM client abstraction for calling LLM endpoints.

	Provides a generic RPC abstraction: point it at an endpoint/port, tell it the
	protocol, and it works. OpenAI-compatible API is the first implementation,
	covering OpenAI, vLLM, TGI, Ollama, HuggingFace Inference API, etc.
	Anthropic's native API is supported via `AnthropicClient`.

	Examples:

	```python
	client = OpenAIClient("http://localhost", 8000, model="meta-llama/...")
	response = await client.complete("What is 2+2?")

	# The endpoint may carry its own port and path prefix:
	client = OpenAIClient("http://localhost:8000/v1", port=None, model="meta-llama/...")

	# Or use the factory for hosted APIs:
	client = create_llm_client("openai", model="gpt-4", api_key="sk-...")
	response = await client.complete_with_tools(messages, tools)
	```
	"""

	from __future__ import annotations

	import json
	from abc import ABC, abstractmethod
	from dataclasses import dataclass, field
	from typing import Any
	from urllib.parse import urlsplit, urlunsplit

	from openai import AsyncOpenAI

	_OPENAI_API_PREFIX = "/v1"
	_MIN_PORT, _MAX_PORT = 1, 65535


	def _redact_userinfo(endpoint: str) -> str:
	"""Drop `user:password@` from a URL so it is safe to echo in an error."""
	scheme, sep, rest = endpoint.partition("://")
	authority, slash, tail = rest.partition("/")
	if not sep or "@" not in authority:
	return endpoint
	return f"{scheme}{sep}***@{authority.rsplit('@', 1)[-1]}{slash}{tail}"


	def _join_endpoint_port(endpoint: str, port: int \| None) -> str:
	"""Validate an endpoint URL and combine it with an optional port.

	Only `http`/`https` URLs with a host are accepted. Credentials, query
	strings and fragments are rejected: the endpoint is logged and persisted in
	rollout metadata, and the SDKs build request URLs by appending to the path,
	which corrupts a query string. `port` is appended only when the URL does
	not name one; an explicit port that differs from the one in the URL is
	rejected rather than silently overridden. A trailing slash on the path is
	dropped. Invalid endpoints raise `ValueError`.
	"""
	safe = _redact_userinfo(endpoint)
	try:
	parts = urlsplit(endpoint)
	if parts.scheme not in ("http", "https"):
	raise ValueError("expected an http:// or https:// URL")
	if not parts.hostname:
	raise ValueError("missing host")
	if parts.username is not None or parts.password is not None:
	raise ValueError("credentials in the URL are not supported, use api_key")
	if parts.query or parts.fragment:
	raise ValueError("query strings and fragments are not supported")
	if parts.netloc.endswith(":"):
	raise ValueError("empty port")
	url_port = parts.port
	for candidate in (url_port, port):
	if candidate is not None and not _MIN_PORT <= candidate <= _MAX_PORT:
	raise ValueError(
	f"port {candidate} is out of range {_MIN_PORT}-{_MAX_PORT}"
	)
	except ValueError as exc:
	raise ValueError(f"Invalid endpoint URL {safe!r}: {exc}") from exc

	if url_port is None:
	if port is not None:
	parts = parts._replace(netloc=f"{parts.netloc}:{port}")
	elif port is not None and port != url_port:
	raise ValueError(
	f"Endpoint URL {safe!r} already specifies port {url_port}, "
	f"which conflicts with port={port}"
	)
	return urlunsplit(parts._replace(path=parts.path.rstrip("/")))


	def _openai_base_url(base_url: str) -> str:
	"""Append the OpenAI `/v1` prefix when `base_url` has no path.

	A URL that already names a path (`/v1`, or a gateway prefix such as
	`/openai/v1`) is used as-is, matching the OpenAI SDK convention that
	`base_url` includes the API prefix.
	"""
	parts = urlsplit(base_url)
	if parts.path in ("", "/"):
	return urlunsplit(parts._replace(path=_OPENAI_API_PREFIX))
	return base_url


	@dataclass
	class ToolCall:
	"""A single tool/function call returned by the model."""

	id: str
	name: str
	args: dict[str, Any]


	@dataclass
	class LLMResponse:
	"""Normalized response from an LLM, with optional tool calls."""

	content: str
	tool_calls: list[ToolCall] = field(default_factory=list)

	def to_message_dict(self) -> dict[str, Any]:
	"""Convert to an OpenAI-format assistant message dict."""
	msg: dict[str, Any] = {"role": "assistant", "content": self.content}
	if self.tool_calls:
	msg["tool_calls"] = [
	{
	"id": tc.id,
	"type": "function",
	"function": {
	"name": tc.name,
	"arguments": json.dumps(tc.args),
	},
	}
	for tc in self.tool_calls
	]
	return msg


	class LLMClient(ABC):
	"""Abstract base for LLM endpoint clients.

	Subclass and implement `complete()` for your protocol.

	Args:
	endpoint (`str`):
	The `http(s)` base URL of the LLM service (e.g. "http://localhost").
	May include a port and a path (e.g. "http://localhost:8000/v1").
	Credentials, query strings and fragments are rejected.
	port (`int` or `None`):
	The port the service listens on. Appended to `endpoint` when the
	URL does not name one; must match the URL's port when both are given.
	"""

	def __init__(self, endpoint: str, port: int \| None):
	self.endpoint = endpoint
	self.port = port

	@abstractmethod
	async def complete(self, prompt: str, **kwargs) -> str:
	"""Send a prompt, return the text response.

	Args:
	prompt (`str`):
	The user prompt to send.
	**kwargs:
	Override default parameters (temperature, max_tokens, etc.).

	Returns:
	The model's text response.
	"""
	...

	async def complete_with_tools(
	self,
	messages: list[dict[str, Any]],
	tools: list[dict[str, Any]],
	**kwargs: Any,
	) -> LLMResponse:
	"""Send messages with tool definitions, return a normalized response.

	Messages use OpenAI-format dicts (`{"role": "...", "content": "..."}`).
	Tools use MCP tool definitions; they are converted internally.

	Args:
	messages (`list[dict[str, Any]]`):
	Conversation history as OpenAI-format message dicts.
	tools (`list[dict[str, Any]]`):
	MCP tool definitions.
	**kwargs:
	Override default parameters (temperature, max_tokens, etc.).

	Returns:
	An [`LLMResponse`] with the model's text and any tool calls.
	"""
	raise NotImplementedError(
	f"{type(self).__name__} does not support tool calling"
	)

	@property
	def base_url(self) -> str:
	"""Base URL of the service: `endpoint` plus `port` when the URL names none."""
	return _join_endpoint_port(self.endpoint, self.port)


	class OpenAIClient(LLMClient):
	"""Client for OpenAI-compatible APIs.

	Works with: OpenAI, vLLM, TGI, Ollama, HuggingFace Inference API,
	or any endpoint that speaks the OpenAI chat completions format.

	Args:
	endpoint (`str`):
	The base URL (e.g. "http://localhost"). May include a port and a
	path (e.g. "http://localhost:8000/v1"). The `/v1` API prefix is
	appended when the URL has no path; a URL with a path is used as-is.
	port (`int` or `None`):
	The port number, appended when `endpoint` does not name one; must
	match the URL's port when both are given.
	model (`str`):
	Model name to pass to the API.
	api_key (`str`, optional):
	API key. Defaults to "not-needed" for local endpoints.
	system_prompt (`str`, optional):
	System message prepended to every request.
	temperature (`float`, optional, defaults to `0.0`):
	Default sampling temperature.
	max_tokens (`int`, optional, defaults to `256`):
	Default max tokens in the response.
	use_max_completion_tokens (`bool`, optional, defaults to `False`):
	Use max_completion_tokens instead of max_tokens. Required for newer OpenAI models
	(gpt-5-mini, o1, o3). Not supported by self-hosted OpenAI-compatible endpoints.
	"""

	def __init__(
	self,
	endpoint: str,
	port: int \| None,
	model: str,
	api_key: str \| None = None,
	system_prompt: str \| None = None,
	temperature: float = 0.0,
	max_tokens: int = 256,
	use_max_completion_tokens: bool = False,
	):
	super().__init__(endpoint, port)
	self.model = model
	self.system_prompt = system_prompt
	self.temperature = temperature
	self.max_tokens = max_tokens
	self._tokens_param = (
	"max_completion_tokens" if use_max_completion_tokens else "max_tokens"
	)
	self._omit_temperature = use_max_completion_tokens

	self._client = AsyncOpenAI(
	base_url=_openai_base_url(self.base_url),
	api_key=api_key if api_key is not None else "not-needed",
	)

	def _chat_completion_kwargs(
	self, messages: list[dict[str, Any]], **kwargs: Any
	) -> dict[str, Any]:
	create_kwargs: dict[str, Any] = {
	"model": self.model,
	"messages": messages,
	self._tokens_param: kwargs.get("max_tokens", self.max_tokens),
	}
	if not self._omit_temperature:
	create_kwargs["temperature"] = kwargs.get("temperature", self.temperature)
	return create_kwargs

	async def complete(self, prompt: str, **kwargs) -> str:
	"""Send a chat completion request.

	Args:
	prompt (`str`):
	The user message.
	**kwargs:
	Overrides for temperature, max_tokens.

	Returns:
	The assistant's response text.
	"""
	messages = []
	if self.system_prompt:
	messages.append({"role": "system", "content": self.system_prompt})
	messages.append({"role": "user", "content": prompt})

	call_kwargs = self._chat_completion_kwargs(messages, **kwargs)
	response = await self._client.chat.completions.create(**call_kwargs)
	return response.choices[0].message.content or ""

	async def complete_with_tools(
	self,
	messages: list[dict[str, Any]],
	tools: list[dict[str, Any]],
	**kwargs: Any,
	) -> LLMResponse:
	create_kwargs = self._chat_completion_kwargs(messages, **kwargs)
	openai_tools = _mcp_tools_to_openai(tools)
	if openai_tools:
	create_kwargs["tools"] = openai_tools

	response = await self._client.chat.completions.create(**create_kwargs)
	msg = response.choices[0].message

	tool_calls = []
	if msg.tool_calls:
	for tc in msg.tool_calls:
	tool_calls.append(
	ToolCall(
	id=tc.id,
	name=tc.function.name,
	args=json.loads(tc.function.arguments),
	)
	)

	return LLMResponse(content=msg.content or "", tool_calls=tool_calls)


	class AnthropicClient(LLMClient):
	"""Client for Anthropic's Messages API.

	Requires the `anthropic` package (lazy-imported at construction time).

	Args:
	endpoint (`str`):
	The base URL (e.g. `https://api.anthropic.com`). May include a port.
	port (`int` or `None`):
	The port number, appended when `endpoint` does not name one; must
	match the URL's port when both are given.
	model (`str`):
	Model name (e.g. "claude-sonnet-4-20250514").
	api_key (`str`, optional):
	Anthropic API key.
	system_prompt (`str`, optional):
	System message prepended to every request.
	temperature (`float`, optional, defaults to `0.0`):
	Default sampling temperature.
	max_tokens (`int`, optional, defaults to `256`):
	Default max tokens in the response.
	"""

	def __init__(
	self,
	endpoint: str,
	port: int \| None,
	model: str,
	api_key: str \| None = None,
	system_prompt: str \| None = None,
	temperature: float = 0.0,
	max_tokens: int = 256,
	):
	super().__init__(endpoint, port)
	self.model = model
	self.system_prompt = system_prompt
	self.temperature = temperature
	self.max_tokens = max_tokens

	try:
	from anthropic import AsyncAnthropic
	except ImportError as exc:
	raise ImportError(
	"AnthropicClient requires the 'anthropic' package. "
	"Install it with: pip install anthropic"
	) from exc

	self._client = AsyncAnthropic(
	base_url=self.base_url,
	api_key=api_key if api_key is not None else "not-needed",
	)

	async def complete(self, prompt: str, **kwargs) -> str:
	create_kwargs: dict[str, Any] = {
	"model": self.model,
	"messages": [{"role": "user", "content": prompt}],
	"temperature": kwargs.get("temperature", self.temperature),
	"max_tokens": kwargs.get("max_tokens", self.max_tokens),
	}
	if self.system_prompt:
	create_kwargs["system"] = self.system_prompt

	response = await self._client.messages.create(**create_kwargs)
	return "".join(block.text for block in response.content if block.type == "text")

	async def complete_with_tools(
	self,
	messages: list[dict[str, Any]],
	tools: list[dict[str, Any]],
	**kwargs: Any,
	) -> LLMResponse:
	system, anthropic_msgs = _openai_msgs_to_anthropic(messages)

	create_kwargs: dict[str, Any] = {
	"model": self.model,
	"messages": anthropic_msgs,
	"temperature": kwargs.get("temperature", self.temperature),
	"max_tokens": kwargs.get("max_tokens", self.max_tokens),
	}
	system_text = system or self.system_prompt
	if system_text:
	create_kwargs["system"] = system_text
	anthropic_tools = _mcp_tools_to_anthropic(tools)
	if anthropic_tools:
	create_kwargs["tools"] = anthropic_tools

	response = await self._client.messages.create(**create_kwargs)

	content = ""
	tool_calls = []
	for block in response.content:
	if block.type == "text":
	content += block.text
	elif block.type == "tool_use":
	tool_calls.append(
	ToolCall(id=block.id, name=block.name, args=block.input)
	)

	return LLMResponse(content=content, tool_calls=tool_calls)


	# ---------------------------------------------------------------------------
	# Factory
	# ---------------------------------------------------------------------------

	_HOSTED_PROVIDERS: dict[str, tuple[str, int, type[LLMClient]]] = {
	"openai": ("https://api.openai.com", 443, OpenAIClient),
	"anthropic": ("https://api.anthropic.com", 443, AnthropicClient),
	}

	# Models that require max_completion_tokens instead of max_tokens and do not
	# accept an explicit temperature parameter. Checked by prefix to cover versioned
	# names such as "o1-2024-12-17" or "gpt-5-mini-2026-01-15".
	_MAX_COMPLETION_TOKENS_PREFIXES: frozenset[str] = frozenset(
	{"gpt-5-mini", "o1", "o3", "o4-mini"}
	)


	def create_llm_client(
	provider: str,
	model: str,
	api_key: str,
	*,
	system_prompt: str \| None = None,
	temperature: float = 0.0,
	max_tokens: int = 4096,
	) -> LLMClient:
	"""Create an LLM client for a hosted provider.

	Args:
	provider (`str`):
	Provider name ("openai" or "anthropic").
	model (`str`):
	Model identifier.
	api_key (`str`):
	API key for the provider.
	system_prompt (`str`, optional):
	System message prepended to every request.
	temperature (`float`, optional, defaults to `0.0`):
	Sampling temperature.
	max_tokens (`int`, optional, defaults to `4096`):
	Maximum tokens in the response.

	Returns:
	A configured [`LLMClient`] instance.
	"""
	key = provider.lower()
	if key not in _HOSTED_PROVIDERS:
	raise ValueError(
	f"Unsupported provider: {provider!r}. "
	f"Supported: {sorted(_HOSTED_PROVIDERS)}"
	)
	endpoint, port, cls = _HOSTED_PROVIDERS[key]
	extra: dict[str, Any] = {}
	if cls is OpenAIClient and any(
	model.startswith(prefix) for prefix in _MAX_COMPLETION_TOKENS_PREFIXES
	):
	extra["use_max_completion_tokens"] = True
	return cls(
	endpoint,
	port,
	model,
	api_key=api_key,
	system_prompt=system_prompt,
	temperature=temperature,
	max_tokens=max_tokens,
	**extra,
	)


	# ---------------------------------------------------------------------------
	# MCP tool-schema helpers
	# ---------------------------------------------------------------------------


	def _clean_mcp_schema(schema: dict[str, Any]) -> dict[str, Any]:
	"""Normalize an MCP tool `inputSchema` for LLM function-calling APIs."""
	if not isinstance(schema, dict):
	return {"type": "object", "properties": {}, "required": []}

	# Shallow copy to avoid mutating the caller's schema dict.
	schema = dict(schema)

	if "oneOf" in schema:
	for option in schema["oneOf"]:
	if isinstance(option, dict) and option.get("type") == "object":
	schema = option
	break
	else:
	return {"type": "object", "properties": {}, "required": []}

	if "allOf" in schema:
	merged: dict[str, Any] = {"type": "object", "properties": {}, "required": []}
	for sub in schema["allOf"]:
	if isinstance(sub, dict):
	if "properties" in sub:
	merged["properties"].update(sub["properties"])
	if "required" in sub:
	merged["required"].extend(sub["required"])
	schema = merged

	if "anyOf" in schema:
	for option in schema["anyOf"]:
	if isinstance(option, dict) and option.get("type") == "object":
	schema = option
	break
	else:
	return {"type": "object", "properties": {}, "required": []}

	schema.setdefault("type", "object")
	if schema.get("type") == "object" and "properties" not in schema:
	schema["properties"] = {}
	return schema


	def _mcp_tools_to_openai(
	mcp_tools: list[dict[str, Any]],
	) -> list[dict[str, Any]]:
	"""Convert MCP tool definitions to OpenAI function-calling format."""
	result = []
	for tool in mcp_tools:
	input_schema = tool.get(
	"inputSchema", {"type": "object", "properties": {}, "required": []}
	)
	result.append(
	{
	"type": "function",
	"function": {
	"name": tool["name"],
	"description": tool.get("description", ""),
	"parameters": _clean_mcp_schema(input_schema),
	},
	}
	)
	return result


	def _mcp_tools_to_anthropic(
	mcp_tools: list[dict[str, Any]],
	) -> list[dict[str, Any]]:
	"""Convert MCP tool definitions to Anthropic tool format."""
	result = []
	for tool in mcp_tools:
	input_schema = tool.get(
	"inputSchema", {"type": "object", "properties": {}, "required": []}
	)
	result.append(
	{
	"name": tool["name"],
	"description": tool.get("description", ""),
	"input_schema": _clean_mcp_schema(input_schema),
	}
	)
	return result


	def _openai_msgs_to_anthropic(
	messages: list[dict[str, Any]],
	) -> tuple[str, list[dict[str, Any]]]:
	"""Convert OpenAI-format messages to Anthropic format.

	Returns `(system_text, anthropic_messages)`. System-role messages are
	extracted and concatenated; tool-result messages are converted to
	Anthropic's `tool_result` content blocks inside user turns.
	"""
	system_parts: list[str] = []
	anthropic_msgs: list[dict[str, Any]] = []

	for msg in messages:
	role = msg["role"]

	if role == "system":
	system_parts.append(msg["content"])

	elif role == "user":
	anthropic_msgs.append({"role": "user", "content": msg["content"]})

	elif role == "assistant":
	if msg.get("tool_calls"):
	content: list[dict[str, Any]] = []
	if msg.get("content"):
	content.append({"type": "text", "text": msg["content"]})
	for tc in msg["tool_calls"]:
	args = tc["function"]["arguments"]
	if isinstance(args, str):
	args = json.loads(args)
	content.append(
	{
	"type": "tool_use",
	"id": tc["id"],
	"name": tc["function"]["name"],
	"input": args,
	}
	)
	anthropic_msgs.append({"role": "assistant", "content": content})
	else:
	anthropic_msgs.append(
	{"role": "assistant", "content": msg.get("content", "")}
	)

	elif role == "tool":
	tool_result = {
	"type": "tool_result",
	"tool_use_id": msg["tool_call_id"],
	"content": msg["content"],
	}
	# Anthropic requires tool results in user turns; merge if possible.
	if (
	anthropic_msgs
	and anthropic_msgs[-1]["role"] == "user"
	and isinstance(anthropic_msgs[-1]["content"], list)
	):
	anthropic_msgs[-1]["content"].append(tool_result)
	else:
	anthropic_msgs.append({"role": "user", "content": [tool_result]})

	system = "\n\n".join(system_parts)
	return system, anthropic_msgs