lingma-openai-gateway/app/http/tooling_policy.py

from __future__ import annotations

from typing import Any

from fastapi import HTTPException

from ..anthropic_schema import AnthropicMessagesRequest
from ..config import Settings
from ..openai_schema import ChatCompletionsRequest
from .tool_bridge import (
    _anthropic_forced_tool_name,
    _anthropic_tool_name,
    _openai_forced_tool_name,
    _openai_tool_name,
)


def _tool_allowlist(settings: Settings) -> set[str]:
    return {name.strip() for name in settings.tool_allowlist if isinstance(name, str) and name.strip()}


def _filter_allowed_tools(
    tools: list[dict[str, Any]], *, provider: str, settings: Settings
) -> list[dict[str, Any]]:
    allowlist = _tool_allowlist(settings)
    if not allowlist:
        return tools
    name_fn = _openai_tool_name if provider == "openai" else _anthropic_tool_name
    return [tool for tool in tools if (name := name_fn(tool)) and name in allowlist]


def _ensure_tool_choice_allowed(tool_choice: Any, *, provider: str, settings: Settings) -> None:
    allowlist = _tool_allowlist(settings)
    if not allowlist:
        return
    forced_name = (
        _openai_forced_tool_name(tool_choice)
        if provider == "openai"
        else _anthropic_forced_tool_name(tool_choice)
    )
    if forced_name and forced_name not in allowlist:
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "type": "invalid_request_error",
                    "message": f"tool '{forced_name}' is not allowed",
                }
            },
        )


def _openai_tool_config(req: ChatCompletionsRequest, *, settings: Settings) -> dict[str, Any] | None:
    if not settings.tool_forward_enabled:
        return None
    has_tools = isinstance(req.tools, list) and len(req.tools) > 0
    has_choice = req.tool_choice is not None
    if not has_tools and not has_choice:
        return None
    _ensure_tool_choice_allowed(req.tool_choice, provider="openai", settings=settings)
    tools = _filter_allowed_tools(req.tools or [], provider="openai", settings=settings)
    return {
        "provider": "openai",
        "tools": tools,
        "tool_choice": req.tool_choice,
    }


def _anthropic_tool_config(
    req: AnthropicMessagesRequest, *, settings: Settings
) -> dict[str, Any] | None:
    if not settings.tool_forward_enabled:
        return None
    has_tools = isinstance(req.tools, list) and len(req.tools) > 0
    has_choice = req.tool_choice is not None
    if not has_tools and not has_choice:
        return None
    _ensure_tool_choice_allowed(req.tool_choice, provider="anthropic", settings=settings)
    tools = _filter_allowed_tools(req.tools or [], provider="anthropic", settings=settings)
    return {
        "provider": "anthropic",
        "tools": tools,
        "tool_choice": req.tool_choice,
    }


def _openai_has_tooling_context(req: ChatCompletionsRequest, messages: list[dict[str, Any]]) -> bool:
    if isinstance(req.tools, list) and len(req.tools) > 0:
        return True
    if req.tool_choice is not None:
        return True
    for m in messages:
        role = m.get("role")
        if role == "tool":
            return True
        if role == "assistant" and m.get("tool_calls"):
            return True
    return False


def _anthropic_content_has_tool_blocks(content: Any) -> bool:
    if not isinstance(content, list):
        return False
    for item in content:
        if isinstance(item, dict) and item.get("type") in {"tool_use", "tool_result"}:
            return True
    return False


def _anthropic_has_tooling_context(req: AnthropicMessagesRequest) -> bool:
    if isinstance(req.tools, list) and len(req.tools) > 0:
        return True
    if req.tool_choice is not None:
        return True
    if _anthropic_content_has_tool_blocks(req.system):
        return True
    for m in req.messages:
        if _anthropic_content_has_tool_blocks(m.content):
            return True
    return False