hermes-agent/agent/anthropic_endpoints.py

"""Endpoint-family detection for Anthropic-compatible base URLs.

A dozen services speak the Anthropic Messages API but differ in auth style, accepted beta
headers, and request quirks (MiniMax, Kimi/Moonshot, DeepSeek, OpenCode, Azure AI Foundry, Nous
Portal, Bedrock). Every such difference is decided from the configured base URL, so the
predicates live together here as pure functions (no I/O, SDK or credentials) that both
``agent/anthropic_adapter.py`` and ``agent/anthropic_message_convert.py`` can import without a
cycle.
"""

from urllib.parse import urlparse

from utils import base_url_host_matches, base_url_hostname

_MINIMAX_ANTHROPIC_PREFIXES = ("https://api.minimax.io/anthropic", "https://api.minimaxi.com/anthropic")


def _normalize_base_url_text(base_url) -> str:
    """Coerce a base URL (str or ``httpx.URL``) to a stripped string; "" when falsy."""
    return str(base_url).strip() if base_url else ""


def _normalized_lower(base_url) -> str:
    """``_normalize_base_url_text`` + rstrip("/") + lower(), the shape most predicates match on."""
    return _normalize_base_url_text(base_url).rstrip("/").lower()


def _is_third_party_anthropic_endpoint(base_url: str | None) -> bool:
    """Any non-anthropic.com endpoint (own x-api-key keys; skip OAuth detection). No base_url =
    direct Anthropic API."""
    normalized = _normalized_lower(base_url)
    return bool(normalized) and "anthropic.com" not in normalized


def _is_kimi_coding_endpoint(base_url: str | None) -> bool:
    """Kimi's /coding endpoint, which requires a claude-code User-Agent."""
    return _normalized_lower(base_url).startswith("https://api.kimi.com/coding")


def _is_opencode_endpoint(base_url: str | None) -> bool:
    """OpenCode's Zen/Go relay (opencode.ai)."""
    return base_url_host_matches(base_url or "", "opencode.ai")


# Kimi / Moonshot family model-name prefixes: official slugs (``kimi-k2.5``, ``kimi_thinking``,
# ``moonshot-v1-8k``) and release lines (``k1.5-…``, ``k2-thinking``, ``k25-…``, ``k3.x``/``k3-…``).
# Matched case-insensitively after stripping any ``vendor/`` prefix.
_KIMI_FAMILY_MODEL_PREFIXES = (
    "kimi-", "kimi_", "moonshot-", "moonshot_", "k1.", "k1-", "k2.", "k2-", "k25", "k2.5", "k3.", "k3-",
)
# Bare release slugs with no separator suffix (Kimi Coding Plan serves K3 as exactly ``k3``);
# exact-match so unrelated names sharing the prefix don't match.
_KIMI_FAMILY_EXACT_SLUGS = frozenset({"k3"})


def _model_name_is_kimi_family(model: str | None) -> bool:
    if not isinstance(model, str):
        return False
    m = model.strip().lower().rsplit("/", 1)[-1]  # ``moonshotai/kimi-k2.5`` -> ``kimi-k2.5``
    return bool(m) and (m in _KIMI_FAMILY_EXACT_SLUGS or m.startswith(_KIMI_FAMILY_MODEL_PREFIXES))


def _is_kimi_family_endpoint(base_url: str | None, model: str | None = None) -> bool:
    """Any Kimi / Moonshot Anthropic-Messages endpoint: the /coding endpoint, any api.kimi.com /
    moonshot.ai / moonshot.cn host, or any endpoint (e.g. a private gateway) whose *model* is in
    the Kimi family — the upstream enforces Kimi's thinking semantics regardless of hostname.
    Decides whether unsigned reasoning_content-derived thinking blocks are preserved on replay."""
    return (
        _is_kimi_coding_endpoint(base_url)
        or any(base_url_host_matches(base_url or "", d) for d in ("api.kimi.com", "moonshot.ai", "moonshot.cn"))
        or _model_name_is_kimi_family(model)
    )



_DEEPSEEK_THINKING_MODEL_PREFIXES = (
    "deepseek-r", "deepseek-v4", "deepseek_v4", "deepseek-pro",
    "deepseek_pro", "deepseek-flash", "deepseek_flash",
)


def _model_name_is_deepseek_thinking(model: str | None) -> bool:
    """Known DeepSeek thinking families behind an Anthropic-compatible relay.

    Strip vendor namespaces, but do not treat arbitrary DeepSeek chat/distill
    names as evidence of the thinking replay contract.
    """
    if not isinstance(model, str):
        return False
    name = model.strip().lower().rsplit("/", 1)[-1]
    return bool(name) and name.startswith(_DEEPSEEK_THINKING_MODEL_PREFIXES)


def _is_deepseek_anthropic_endpoint(base_url: str | None) -> bool:
    """DeepSeek's ``/anthropic`` route. In thinking mode DeepSeek requires prior-turn ``thinking``
    blocks to round-trip while the generic third-party path strips them; its blocks are unsigned,
    so it gets the same strip-signed / keep-unsigned policy as Kimi. Pinned to the ``/anthropic``
    path so the OpenAI-compatible base URL is not misclassified.

    Per DeepSeek's published compatibility matrix the blocks are unsigned (no Anthropic-proprietary
    signature, no ``redacted_thinking`` support), so this endpoint is handled with the same strip-signed /
    keep-unsigned policy used for Kimi's ``/coding`` endpoint. See hermes-agent#16748.
    """
    return base_url_host_matches(base_url or "", "api.deepseek.com") and "/anthropic" in _normalized_lower(base_url)


def _is_nous_portal_endpoint(base_url: str | None) -> bool:
    """Nous Portal's Anthropic Messages route (Bearer JWT, verbatim catalog ids, native
    thinking-signature replay). Trusted hosts only: prod ``inference-api.nousresearch.com`` or the
    operator-set ``NOUS_INFERENCE_BASE_URL`` host (exact hostname equality, so neither lookalike
    domains nor sibling hosts of the override match)."""
    if base_url_host_matches(base_url or "", "inference-api.nousresearch.com"):
        return True
    try:
        from hermes_cli.auth import _nous_inference_env_override
        override = _nous_inference_env_override()
    except Exception:
        return False
    override_host = base_url_hostname(override) if override else ""
    return bool(override_host) and base_url_hostname(base_url or "") == override_host


def _requires_bearer_auth(base_url: str | None) -> bool:
    """Providers needing ``Authorization: Bearer`` instead of ``x-api-key``: MiniMax, Azure AI
    Foundry, Palantir Foundry's LLM proxy, CommandCode, Nous Portal. Palantir/CommandCode use
    hostname matching (not substring) so ``evil.com/palantirfoundry`` paths don't trigger it."""
    normalized = _normalized_lower(base_url)
    return (
        _is_nous_portal_endpoint(base_url)
        or normalized.startswith(_MINIMAX_ANTHROPIC_PREFIXES)
        or "azure.com" in normalized
        or base_url_host_matches(normalized, "palantirfoundry.com")
        or base_url_host_matches(normalized, "api.commandcode.ai")
    )


def _base_url_needs_context_1m_beta(base_url: str | None) -> bool:
    """Endpoints that still gate 1M context behind a beta (Azure)."""
    return "azure.com" in _normalize_base_url_text(base_url).lower()


def _is_minimax_anthropic_endpoint(base_url: str | None) -> bool:
    """MiniMax's Anthropic-compatible endpoints, which reject the fine-grained-tool-streaming and
    context-1m betas (stripped even though MiniMax also uses Bearer auth)."""
    return _normalized_lower(base_url).startswith(_MINIMAX_ANTHROPIC_PREFIXES)


def _is_azure_anthropic_endpoint(base_url: str | None) -> bool:
    """Azure-hosted Anthropic Messages endpoints serving ``/anthropic``: modern Foundry
    (``*.services.ai.azure.*``) and legacy Azure OpenAI (``*.openai.azure.*``) hosts; opts them
    into ``api-version`` query plumbing. Deliberately no finite TLD allow-list, so
    sovereign/private clouds work."""
    normalized = _normalize_base_url_text(base_url)
    if not normalized:
        return False
    parsed = urlparse(normalized)
    host_padded = f".{(parsed.hostname or '').lower().rstrip('.')}."
    is_azure_host = ".services.ai.azure." in host_padded or ".openai.azure." in host_padded
    return is_azure_host and "/anthropic" in (parsed.path or "").lower()