refactor: extract tooling policy helpers
Move tool allowlist, tool_config, and tooling-context helpers into app/http/tooling_policy.py while keeping route behavior unchanged. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
120
app/http/tooling_policy.py
Normal file
120
app/http/tooling_policy.py
Normal file
@@ -0,0 +1,120 @@
|
||||
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
|
||||
Reference in New Issue
Block a user