diff --git a/litellm/constants.py b/litellm/constants.py index d78c124d71d..9539cf25a10 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -513,6 +513,7 @@ ANTHROPIC_WEB_SEARCH_TOOL_MAX_USES = { # LiteLLM standard web search tool name # Used for web search interception across providers LITELLM_WEB_SEARCH_TOOL_NAME = "litellm_web_search" +LITELLM_WEB_FETCH_TOOL_NAME = "litellm_web_fetch" DEFAULT_IMAGE_ENDPOINT_MODEL = "dall-e-2" DEFAULT_VIDEO_ENDPOINT_MODEL = "sora-2" diff --git a/litellm/integrations/webfetch_interception/__init__.py b/litellm/integrations/webfetch_interception/__init__.py new file mode 100644 index 00000000000..827d59be5c9 --- /dev/null +++ b/litellm/integrations/webfetch_interception/__init__.py @@ -0,0 +1,20 @@ +""" +WebFetch Interception Module + +Provides server-side WebFetch tool execution for models that don't natively +support server-side tool calling (e.g., Bedrock/Claude). +""" + +from litellm.integrations.webfetch_interception.handler import ( + WebFetchInterceptionLogger, +) +from litellm.integrations.webfetch_interception.tools import ( + get_litellm_web_fetch_tool, + is_web_fetch_tool, +) + +__all__ = [ + "WebFetchInterceptionLogger", + "get_litellm_web_fetch_tool", + "is_web_fetch_tool", +] diff --git a/litellm/integrations/webfetch_interception/handler.py b/litellm/integrations/webfetch_interception/handler.py new file mode 100644 index 00000000000..3996b2e8d07 --- /dev/null +++ b/litellm/integrations/webfetch_interception/handler.py @@ -0,0 +1,1038 @@ +""" +WebFetch Interception Handler + +CustomLogger that intercepts WebFetch tool calls for models that don't +natively support web fetch (e.g., Bedrock/Claude) and executes them +server-side using litellm router's fetch tools. +""" + +import asyncio +import math +import uuid +from typing import Any, Dict, List, Optional, Tuple, Union, cast + +import litellm +from litellm._logging import verbose_logger +from litellm.anthropic_interface import messages as anthropic_messages +from litellm.constants import LITELLM_WEB_FETCH_TOOL_NAME +from litellm.integrations.custom_logger import CustomLogger +from litellm.integrations.webfetch_interception.tools import ( + get_litellm_web_fetch_tool, + get_litellm_web_fetch_tool_openai, + is_web_fetch_tool, + is_web_fetch_tool_chat_completion, +) +from litellm.integrations.webfetch_interception.transformation import ( + WebFetchTransformation, +) +from litellm.llms.base_llm.fetch.transformation import BaseFetchConfig, WebFetchResponse +from litellm.types.integrations.custom_logger import ( + AgenticLoopPlan, + AgenticLoopRequestPatch, +) +from litellm.types.llms.openai import AllMessageValues +from litellm.types.utils import LlmProviders +from litellm.utils import ProviderConfigManager + + +class WebFetchInterceptionLogger(CustomLogger): + """ + CustomLogger that intercepts WebFetch tool calls for models that don't + natively support web fetch. + + Implements agentic loop: + 1. Detects WebFetch tool_use in model response + 2. Executes fetch using router's fetch tools + 3. Makes follow-up request with fetch results + 4. Returns final response + """ + + def __init__( + self, + enabled_providers: Optional[List[Union[LlmProviders, str]]] = None, + fetch_tool_name: Optional[str] = None, + ): + """ + Args: + enabled_providers: List of LLM providers to enable interception for. + Use LlmProviders enum values (e.g., [LlmProviders.BEDROCK]) + If None or empty list, enables for ALL providers. + Default: None (all providers enabled) + fetch_tool_name: Name of fetch tool configured in router's fetch_tools. + If None, will attempt to use first available fetch tool. + """ + super().__init__() + # Convert enum values to strings for comparison + if enabled_providers is None: + self.enabled_providers = [LlmProviders.BEDROCK.value] + else: + self.enabled_providers = [ + p.value if isinstance(p, LlmProviders) else p for p in enabled_providers + ] + self.fetch_tool_name = fetch_tool_name + self._request_has_webfetch = False # Track if current request has web fetch + + async def async_pre_call_deployment_hook( + self, kwargs: Dict[str, Any], call_type: Optional[Any] + ) -> Optional[dict]: + """ + Pre-call hook to convert native Anthropic web_fetch tools to regular tools. + + This prevents Bedrock from trying to execute web_fetch server-side (which fails). + Instead, we convert it to a regular tool so the model returns tool_use blocks + that we can intercept and execute ourselves. + """ + # Check if this is for an enabled provider + custom_llm_provider = kwargs.get("custom_llm_provider", "") or kwargs.get( + "litellm_params", {} + ).get("custom_llm_provider", "") + if not custom_llm_provider: + try: + _, custom_llm_provider, _, _ = litellm.get_llm_provider( + model=kwargs.get("model", "") + ) + except Exception: + custom_llm_provider = "" + if custom_llm_provider not in self.enabled_providers: + return None + + # Check if request has tools with native web_fetch + tools = kwargs.get("tools") + if not tools: + return None + + # Check if any tool is a web fetch tool (native or already LiteLLM standard) + has_webfetch = any(is_web_fetch_tool(t) for t in tools) + + if not has_webfetch: + return None + + verbose_logger.debug( + "WebFetchInterception: Converting native web_fetch tools to LiteLLM standard" + ) + + # Convert native/custom web_fetch tools to LiteLLM standard + converted_tools = [] + for tool in tools: + if is_web_fetch_tool(tool): + # Convert to LiteLLM standard web fetch tool + converted_tool = get_litellm_web_fetch_tool_openai() + converted_tools.append(converted_tool) + verbose_logger.debug( + f"WebFetchInterception: Converted {tool.get('name', 'unknown')} " + f"(type={tool.get('type', 'none')}) to {LITELLM_WEB_FETCH_TOOL_NAME}" + ) + else: + # Keep other tools as-is + converted_tools.append(tool) + + kwargs["tools"] = converted_tools + + if kwargs.get("stream"): + verbose_logger.debug( + "WebFetchInterception: deployment hook converting stream=True to stream=False" + ) + kwargs["stream"] = False + kwargs["_webfetch_interception_converted_stream"] = True + + return kwargs + + @classmethod + def from_config_yaml(cls, config: Dict[str, Any]) -> "WebFetchInterceptionLogger": + """ + Initialize WebFetchInterceptionLogger from proxy config.yaml parameters. + + Args: + config: Configuration dictionary from litellm_settings.webfetch_interception_params + + Returns: + Configured WebFetchInterceptionLogger instance + + Example: + From proxy_config.yaml: + litellm_settings: + webfetch_interception_params: + enabled_providers: ["bedrock"] + fetch_tool_name: "my-firecrawl-fetch" + + Usage: + config = litellm_settings.get("webfetch_interception_params", {}) + logger = WebFetchInterceptionLogger.from_config_yaml(config) + """ + # Extract parameters from config + enabled_providers_str = config.get("enabled_providers", None) + fetch_tool_name = config.get("fetch_tool_name", None) + + # Convert string provider names to LlmProviders enum values + enabled_providers: Optional[List[Union[LlmProviders, str]]] = None + if enabled_providers_str is not None: + enabled_providers = [] + for provider in enabled_providers_str: + try: + # Try to convert string to LlmProviders enum + provider_enum = LlmProviders(provider) + enabled_providers.append(provider_enum) + except ValueError: + # If conversion fails, keep as string + enabled_providers.append(provider) + + return cls( + enabled_providers=enabled_providers, + fetch_tool_name=fetch_tool_name, + ) + + async def async_pre_request_hook( + self, model: str, messages: List[Dict], kwargs: Dict + ) -> Optional[Dict]: + """ + Pre-request hook to convert native web fetch tools to LiteLLM standard. + + This hook is called before the API request is made, allowing us to: + 1. Detect native web fetch tools (web_fetch_20250305, etc.) + 2. Convert them to LiteLLM standard format (litellm_web_fetch) + 3. Convert stream=True to stream=False for interception + + This prevents providers like Bedrock from trying to execute web_fetch + natively (which fails), and ensures our agentic loop can intercept tool_use. + + Returns: + Modified kwargs dict with converted tools, or None if no modifications needed + """ + # Check if this request is for an enabled provider + custom_llm_provider = kwargs.get("litellm_params", {}).get( + "custom_llm_provider", "" + ) + + verbose_logger.debug( + f"WebFetchInterception: Pre-request hook called" + f" - custom_llm_provider={custom_llm_provider}" + f" - enabled_providers={self.enabled_providers or 'ALL'}" + ) + + if ( + self.enabled_providers is not None + and custom_llm_provider not in self.enabled_providers + ): + verbose_logger.debug( + f"WebFetchInterception: Skipping - provider {custom_llm_provider} not in {self.enabled_providers}" + ) + return None + + # Check if request has tools + tools = kwargs.get("tools") + if not tools: + return None + + # Check if any tool is a web fetch tool + has_webfetch = any(is_web_fetch_tool(t) for t in tools) + if not has_webfetch: + return None + + verbose_logger.debug( + f"WebFetchInterception: Pre-request hook triggered for provider={custom_llm_provider}" + ) + + # Convert native web fetch tools to LiteLLM standard + converted_tools = [] + for tool in tools: + if is_web_fetch_tool(tool): + standard_tool = get_litellm_web_fetch_tool() + converted_tools.append(standard_tool) + verbose_logger.debug( + f"WebFetchInterception: Converted {tool.get('name', 'unknown')} " + f"(type={tool.get('type', 'none')}) to {LITELLM_WEB_FETCH_TOOL_NAME}" + ) + else: + converted_tools.append(tool) + + kwargs["tools"] = converted_tools + verbose_logger.debug( + f"WebFetchInterception: Tools after conversion: {[t.get('name') for t in converted_tools]}" + ) + + # Also convert here for direct callers that bypass the deployment hook. + if kwargs.get("stream"): + verbose_logger.debug( + "WebFetchInterception: Converting stream=True to stream=False" + ) + kwargs["stream"] = False + kwargs["_webfetch_interception_converted_stream"] = True + + return kwargs + + async def async_should_run_agentic_loop( + self, + response: Any, + model: str, + messages: List[Dict], + tools: Optional[List[Dict]], + stream: bool, + custom_llm_provider: str, + kwargs: Dict, + ) -> Tuple[bool, Dict]: + """ + Check if WebFetch tool interception is needed for Anthropic Messages API. + """ + + verbose_logger.debug( + f"WebFetchInterception: Hook called! provider={custom_llm_provider}, stream={stream}" + ) + verbose_logger.debug(f"WebFetchInterception: Response type: {type(response)}") + + # Check if provider should be intercepted + if ( + self.enabled_providers is not None + and custom_llm_provider not in self.enabled_providers + ): + verbose_logger.debug( + f"WebFetchInterception: Skipping provider {custom_llm_provider} (not in enabled list: {self.enabled_providers})" + ) + return False, {} + + # Check if tools include any web fetch tool (LiteLLM standard or native) + has_webfetch_tool = any(is_web_fetch_tool(t) for t in (tools or [])) + if not has_webfetch_tool: + verbose_logger.debug("WebFetchInterception: No web fetch tool in request") + return False, {} + + # Detect WebFetch tool_use in response (Anthropic format) + should_intercept, tool_calls = WebFetchTransformation.transform_request( + response=response, + stream=stream, + response_format="anthropic", + ) + + if not should_intercept: + verbose_logger.debug( + "WebFetchInterception: No WebFetch tool_use detected in response" + ) + return False, {} + + verbose_logger.debug( + f"WebFetchInterception: Detected {len(tool_calls)} WebFetch tool call(s), executing agentic loop" + ) + + # Extract thinking blocks from response content. + thinking_blocks: List[Dict] = [] + if isinstance(response, dict): + content = response.get("content", []) + else: + content = getattr(response, "content", []) or [] + + for block in content: + if isinstance(block, dict): + block_type = block.get("type") + else: + block_type = getattr(block, "type", None) + + if block_type in ("thinking", "redacted_thinking"): + if isinstance(block, dict): + thinking_blocks.append(block) + else: + thinking_block_dict: Dict = {"type": block_type} + if block_type == "thinking": + thinking_block_dict["thinking"] = getattr(block, "thinking", "") + thinking_block_dict["signature"] = getattr( + block, "signature", "" + ) + else: # redacted_thinking + thinking_block_dict["data"] = getattr(block, "data", "") + thinking_blocks.append(thinking_block_dict) + + if thinking_blocks: + verbose_logger.debug( + f"WebFetchInterception: Extracted {len(thinking_blocks)} thinking block(s) from response" + ) + + # Return tools dict with tool calls and thinking blocks + tools_dict = { + "tool_calls": tool_calls, + "tool_type": "webfetch", + "provider": custom_llm_provider, + "response_format": "anthropic", + "thinking_blocks": thinking_blocks, + } + return True, tools_dict + + async def async_should_run_chat_completion_agentic_loop( + self, + response: Any, + model: str, + messages: List[Dict], + tools: Optional[List[Dict]], + stream: bool, + custom_llm_provider: str, + kwargs: Dict, + ) -> Tuple[bool, Dict]: + """ + Check if WebFetch tool interception is needed for Chat Completions API. + """ + + verbose_logger.debug( + f"WebFetchInterception: Chat completion hook called! provider={custom_llm_provider}, stream={stream}" + ) + verbose_logger.debug(f"WebFetchInterception: Response type: {type(response)}") + + # Check if provider should be intercepted + if ( + self.enabled_providers is not None + and custom_llm_provider not in self.enabled_providers + ): + verbose_logger.debug( + f"WebFetchInterception: Skipping provider {custom_llm_provider} (not in enabled list: {self.enabled_providers})" + ) + return False, {} + + # Check if tools include any web fetch tool (strict check for chat completions) + has_webfetch_tool = any( + is_web_fetch_tool_chat_completion(t) for t in (tools or []) + ) + if not has_webfetch_tool: + verbose_logger.debug( + "WebFetchInterception: No litellm_web_fetch tool in request" + ) + return False, {} + + # Detect WebFetch tool_calls in response (OpenAI format) + should_intercept, tool_calls = WebFetchTransformation.transform_request( + response=response, + stream=stream, + response_format="openai", + ) + + if not should_intercept: + verbose_logger.debug( + "WebFetchInterception: No WebFetch tool_calls detected in response" + ) + return False, {} + + verbose_logger.debug( + f"WebFetchInterception: Detected {len(tool_calls)} WebFetch tool call(s), executing agentic loop" + ) + + # Return tools dict with tool calls + tools_dict = { + "tool_calls": tool_calls, + "tool_type": "webfetch", + "provider": custom_llm_provider, + "response_format": "openai", + } + return True, tools_dict + + async def async_run_agentic_loop( + self, + tools: Dict, + model: str, + messages: List[Dict], + response: Any, + anthropic_messages_provider_config: Any, + anthropic_messages_optional_request_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + ) -> Any: + """ + Execute agentic loop with WebFetch execution for Anthropic Messages API. + """ + + tool_calls = tools["tool_calls"] + thinking_blocks = tools.get("thinking_blocks", []) + + verbose_logger.debug( + f"WebFetchInterception: Executing agentic loop for {len(tool_calls)} fetch(es)" + ) + + return await self._execute_agentic_loop( + model=model, + messages=messages, + tool_calls=tool_calls, + thinking_blocks=thinking_blocks, + anthropic_messages_optional_request_params=anthropic_messages_optional_request_params, + logging_obj=logging_obj, + stream=stream, + kwargs=kwargs, + ) + + async def async_build_agentic_loop_plan( + self, + tools: Dict, + model: str, + messages: List[Dict], + response: Any, + anthropic_messages_provider_config: Any, + anthropic_messages_optional_request_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + ) -> AgenticLoopPlan: + tool_calls = tools["tool_calls"] + thinking_blocks = tools.get("thinking_blocks", []) + request_patch = await self._build_anthropic_request_patch( + model=model, + messages=messages, + tool_calls=tool_calls, + thinking_blocks=thinking_blocks, + anthropic_messages_optional_request_params=anthropic_messages_optional_request_params, + logging_obj=logging_obj, + kwargs=kwargs, + ) + return AgenticLoopPlan( + run_agentic_loop=True, + request_patch=request_patch, + metadata={"tool_type": "webfetch", "response_format": "anthropic"}, + ) + + async def async_run_chat_completion_agentic_loop( + self, + tools: Dict, + model: str, + messages: List[Dict], + response: Any, + optional_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + ) -> Any: + """ + Execute agentic loop with WebFetch execution for Chat Completions API. + """ + + tool_calls = tools["tool_calls"] + response_format = tools.get("response_format", "openai") + + verbose_logger.debug( + f"WebFetchInterception: Executing chat completion agentic loop for {len(tool_calls)} fetch(es)" + ) + + return await self._execute_chat_completion_agentic_loop( + model=model, + messages=messages, + tool_calls=tool_calls, + optional_params=optional_params, + logging_obj=logging_obj, + stream=stream, + kwargs=kwargs, + response_format=response_format, + ) + + async def async_build_chat_completion_agentic_loop_plan( + self, + tools: Dict, + model: str, + messages: List[Dict], + response: Any, + optional_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + ) -> AgenticLoopPlan: + tool_calls = tools["tool_calls"] + response_format = tools.get("response_format", "openai") + request_patch = await self._build_chat_completion_request_patch( + model=model, + messages=messages, + tool_calls=tool_calls, + optional_params=optional_params, + kwargs=kwargs, + response_format=response_format, + ) + return AgenticLoopPlan( + run_agentic_loop=True, + request_patch=request_patch, + metadata={"tool_type": "webfetch", "response_format": response_format}, + ) + + async def _get_fetch_config(self) -> BaseFetchConfig: + """Get the fetch config for the configured fetch tool.""" + # Import router from proxy_server + try: + from litellm.proxy.proxy_server import llm_router + except ImportError: + verbose_logger.debug( + "WebFetchInterception: Could not import llm_router from proxy_server, " + "falling back to direct fetch" + ) + llm_router = None + + # Determine fetch provider from router's fetch_tools + fetch_provider: Optional[str] = None + api_key: Optional[str] = None + api_base: Optional[str] = None + + if llm_router is not None and hasattr(llm_router, "fetch_tools"): + if self.fetch_tool_name: + # Find specific fetch tool by name + matching_tools = [ + tool + for tool in llm_router.fetch_tools + if tool.get("fetch_tool_name") == self.fetch_tool_name + ] + if matching_tools: + fetch_tool = matching_tools[0] + litellm_params = fetch_tool.get("litellm_params", {}) + fetch_provider = litellm_params.get("fetch_provider") + api_key = litellm_params.get("api_key") + api_base = litellm_params.get("api_base") + verbose_logger.debug( + f"WebFetchInterception: Found fetch tool '{self.fetch_tool_name}' " + f"with provider '{fetch_provider}'" + ) + else: + verbose_logger.debug( + f"WebFetchInterception: Fetch tool '{self.fetch_tool_name}' not found in router" + ) + + # If no specific tool or not found, use first available + if not fetch_provider and llm_router.fetch_tools: + first_tool = llm_router.fetch_tools[0] + litellm_params = first_tool.get("litellm_params", {}) + fetch_provider = litellm_params.get("fetch_provider") + api_key = litellm_params.get("api_key") + api_base = litellm_params.get("api_base") + verbose_logger.debug( + f"WebFetchInterception: Using first available fetch tool with provider '{fetch_provider}'" + ) + + # Fallback to firecrawl if no router or no fetch tools configured + if not fetch_provider: + fetch_provider = "firecrawl" + verbose_logger.debug( + f"WebFetchInterception: No fetch tools configured in router, " + f"using default provider '{fetch_provider}'" + ) + + # Get the fetch config + fetch_config = ProviderConfigManager.get_provider_fetch_config(fetch_provider) + if fetch_config is None: + raise ValueError(f"Unknown fetch provider: {fetch_provider}") + + # Set up headers with API key + headers = {} + if api_key: + headers["api_key"] = api_key + if api_base: + headers["api_base"] = api_base + + fetch_config.validate_environment( + headers=headers, api_key=api_key, api_base=api_base + ) + + return fetch_config + + async def _execute_fetch(self, url: str) -> str: + """Execute a single web fetch using router's fetch tools""" + try: + # Get the fetch config + fetch_config = await self._get_fetch_config() + + verbose_logger.debug( + f"WebFetchInterception: Executing fetch for '{url}' using provider '{fetch_config.ui_friendly_name()}'" + ) + + # Execute fetch + fetch_response = await fetch_config.afetch_url( + url=url, + headers={}, # Headers already set up in _get_fetch_config + optional_params={}, + ) + + # Format using transformation function + fetch_result_text = WebFetchTransformation.format_fetch_response( + fetch_response + ) + + verbose_logger.debug( + f"WebFetchInterception: Fetch completed for '{url}', got {len(fetch_result_text)} chars" + ) + return fetch_result_text + except Exception as e: + verbose_logger.error( + f"WebFetchInterception: Fetch failed for '{url}': {str(e)}" + ) + raise + + @staticmethod + def _resolve_max_tokens( + optional_params: Dict, + kwargs: Dict, + ) -> int: + """Extract max_tokens and validate against thinking.budget_tokens.""" + max_tokens: int = optional_params.get( + "max_tokens", + kwargs.get("max_tokens", 1024), + ) + thinking_param = optional_params.get("thinking") + if thinking_param and isinstance(thinking_param, dict): + budget_tokens = thinking_param.get("budget_tokens") + if ( + budget_tokens is not None + and isinstance(budget_tokens, (int, float)) + and math.isfinite(budget_tokens) + and budget_tokens > 0 + ): + if max_tokens <= budget_tokens: + adjusted = math.ceil(budget_tokens) + 1024 + verbose_logger.debug( + "WebFetchInterception: max_tokens=%s <= thinking.budget_tokens=%s, " + "adjusting to %s to satisfy Anthropic API constraint", + max_tokens, + budget_tokens, + adjusted, + ) + max_tokens = adjusted + return max_tokens + + @staticmethod + def _prepare_followup_kwargs(kwargs: Dict) -> Dict: + """Build kwargs for the follow-up call, excluding internal keys.""" + _internal_keys = {"litellm_logging_obj"} + return { + k: v + for k, v in kwargs.items() + if not k.startswith("_webfetch_interception") and k not in _internal_keys + } + + async def _execute_agentic_loop( + self, + model: str, + messages: List[Dict], + tool_calls: List[Dict], + thinking_blocks: List[Dict], + anthropic_messages_optional_request_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + ) -> Any: + """Legacy path: execute fetch + build patch + run follow-up call.""" + request_patch = await self._build_anthropic_request_patch( + model=model, + messages=messages, + tool_calls=tool_calls, + thinking_blocks=thinking_blocks, + anthropic_messages_optional_request_params=anthropic_messages_optional_request_params, + logging_obj=logging_obj, + kwargs=kwargs, + ) + if request_patch.messages is None: + raise ValueError("WebFetchInterception: missing follow-up messages") + + optional_params = dict(anthropic_messages_optional_request_params) + optional_params.update(request_patch.optional_params) + max_tokens = request_patch.max_tokens + if max_tokens is None: + max_tokens = cast(Optional[int], optional_params.pop("max_tokens", None)) + else: + optional_params.pop("max_tokens", None) + if max_tokens is None: + max_tokens = cast(int, kwargs.get("max_tokens", 1024)) + + return await anthropic_messages.acreate( + max_tokens=max_tokens, + messages=request_patch.messages, + model=request_patch.model or model, + **optional_params, + **request_patch.kwargs, + ) + + async def _build_anthropic_request_patch( + self, + model: str, + messages: List[Dict], + tool_calls: List[Dict], + thinking_blocks: List[Dict], + anthropic_messages_optional_request_params: Dict, + logging_obj: Any, + kwargs: Dict, + ) -> AgenticLoopRequestPatch: + """Execute fetch and build follow-up request patch.""" + + # Extract URLs from tool_use blocks + fetch_tasks = [] + for tool_call in tool_calls: + url = tool_call["input"].get("url") + if url: + verbose_logger.debug( + f"WebFetchInterception: Queuing fetch for url='{url}'" + ) + fetch_tasks.append(self._execute_fetch(url)) + else: + verbose_logger.debug( + f"WebFetchInterception: Tool call {tool_call['id']} has no URL" + ) + # Add empty result for tools without URL + fetch_tasks.append(self._create_empty_fetch_result()) + + # Execute fetches in parallel + verbose_logger.debug( + f"WebFetchInterception: Executing {len(fetch_tasks)} fetch(es) in parallel" + ) + fetch_results = await asyncio.gather(*fetch_tasks, return_exceptions=True) + + # Handle any exceptions in fetch results + final_fetch_results: List[str] = [] + for i, result in enumerate(fetch_results): + if isinstance(result, Exception): + verbose_logger.error( + f"WebFetchInterception: Fetch {i} failed with error: {str(result)}" + ) + final_fetch_results.append(f"Fetch failed: {str(result)}") + elif isinstance(result, str): + final_fetch_results.append(cast(str, result)) + else: + verbose_logger.debug( + f"WebFetchInterception: Unexpected result type {type(result)} at index {i}" + ) + final_fetch_results.append(str(result)) + + # Build assistant and user messages using transformation + assistant_message, user_message = WebFetchTransformation.transform_response( + tool_calls=tool_calls, + fetch_results=final_fetch_results, + thinking_blocks=thinking_blocks, + ) + + follow_up_messages = messages + [assistant_message, cast(Dict, user_message)] + + # Correlation context for structured logging + _call_id = getattr(logging_obj, "litellm_call_id", None) or kwargs.get( + "litellm_call_id", "unknown" + ) + + full_model_name = model # safe default before try block + + max_tokens = self._resolve_max_tokens( + anthropic_messages_optional_request_params, kwargs + ) + + verbose_logger.debug( + f"WebFetchInterception: Using max_tokens={max_tokens} for follow-up request" + ) + + optional_params_without_max_tokens = { + k: v + for k, v in anthropic_messages_optional_request_params.items() + if k != "max_tokens" + } + kwargs_for_followup = self._prepare_followup_kwargs(kwargs) + + if logging_obj is not None: + agentic_params = logging_obj.model_call_details.get( + "agentic_loop_params", {} + ) + full_model_name = agentic_params.get("model", model) + verbose_logger.debug( + "WebFetchInterception: Built anthropic request patch " + "[call_id=%s model=%s messages=%d fetches=%d]", + _call_id, + full_model_name, + len(follow_up_messages), + len(final_fetch_results), + ) + return AgenticLoopRequestPatch( + model=full_model_name, + messages=follow_up_messages, + max_tokens=max_tokens, + optional_params=optional_params_without_max_tokens, + kwargs=kwargs_for_followup, + ) + + async def _execute_chat_completion_agentic_loop( + self, + model: str, + messages: List[Dict], + tool_calls: List[Dict], + optional_params: Dict, + logging_obj: Any, + stream: bool, + kwargs: Dict, + response_format: str = "openai", + ) -> Any: + """Legacy path: execute fetch + build patch + run follow-up call.""" + request_patch = await self._build_chat_completion_request_patch( + model=model, + messages=messages, + tool_calls=tool_calls, + optional_params=optional_params, + kwargs=kwargs, + response_format=response_format, + ) + if request_patch.messages is None: + raise ValueError("WebFetchInterception: missing follow-up messages") + params = dict(optional_params) + params.update(request_patch.optional_params) + return await litellm.acompletion( + model=request_patch.model or model, + messages=request_patch.messages, + **params, + **request_patch.kwargs, + ) + + async def _build_chat_completion_request_patch( + self, + model: str, + messages: List[Dict], + tool_calls: List[Dict], + optional_params: Dict, + kwargs: Dict, + response_format: str = "openai", + ) -> AgenticLoopRequestPatch: + """Execute fetch and build chat-completion rerun patch.""" + + # Extract URLs from tool_calls + fetch_tasks = [] + for tool_call in tool_calls: + # Handle both Anthropic-style input and OpenAI-style function.arguments + url = None + if "input" in tool_call and isinstance(tool_call["input"], dict): + url = tool_call["input"].get("url") + elif "function" in tool_call: + func = tool_call["function"] + if isinstance(func, dict): + args = func.get("arguments", {}) + if isinstance(args, dict): + url = args.get("url") + elif isinstance(args, str): + import json + + try: + args_dict = json.loads(args) + url = args_dict.get("url") + except (json.JSONDecodeError, TypeError): + pass + + if url: + verbose_logger.debug( + f"WebFetchInterception: Queuing fetch for url='{url}'" + ) + fetch_tasks.append(self._execute_fetch(url)) + else: + verbose_logger.debug( + f"WebFetchInterception: Tool call {tool_call.get('id')} has no URL" + ) + # Add empty result for tools without URL + fetch_tasks.append(self._create_empty_fetch_result()) + + # Execute fetches in parallel + verbose_logger.debug( + f"WebFetchInterception: Executing {len(fetch_tasks)} fetch(es) in parallel" + ) + fetch_results = await asyncio.gather(*fetch_tasks, return_exceptions=True) + + # Handle any exceptions in fetch results + final_fetch_results: List[str] = [] + for i, result in enumerate(fetch_results): + if isinstance(result, Exception): + verbose_logger.error( + f"WebFetchInterception: Fetch {i} failed with error: {str(result)}" + ) + final_fetch_results.append(f"Fetch failed: {str(result)}") + elif isinstance(result, str): + final_fetch_results.append(cast(str, result)) + else: + verbose_logger.debug( + f"WebFetchInterception: Unexpected result type {type(result)} at index {i}" + ) + final_fetch_results.append(str(result)) + + # Build assistant and tool messages using transformation + ( + assistant_message, + tool_messages_or_user, + ) = WebFetchTransformation.transform_response( + tool_calls=tool_calls, + fetch_results=final_fetch_results, + response_format=response_format, + ) + + # Make follow-up request with fetch results + if response_format == "openai": + follow_up_messages = ( + messages + [assistant_message] + cast(List[Dict], tool_messages_or_user) + ) + else: + follow_up_messages = messages + [ + assistant_message, + cast(Dict, tool_messages_or_user), + ] + + verbose_logger.debug( + "WebFetchInterception: Making follow-up chat completion request with fetch results" + ) + verbose_logger.debug( + f"WebFetchInterception: Follow-up messages count: {len(follow_up_messages)}" + ) + + # Remove internal parameters that shouldn't be passed to follow-up request + internal_params = { + "_webfetch_interception", + "acompletion", + "litellm_logging_obj", + "custom_llm_provider", + "model_alias_map", + "stream_response", + "custom_prompt_dict", + } + kwargs_for_followup = { + k: v + for k, v in kwargs.items() + if not k.startswith("_webfetch_interception") and k not in internal_params + } + + full_model_name = model + if "custom_llm_provider" in kwargs: + custom_llm_provider = kwargs["custom_llm_provider"] + if not model.startswith(custom_llm_provider) and "/" not in model: + full_model_name = f"{custom_llm_provider}/{model}" + + verbose_logger.debug( + "WebFetchInterception: Built chat completion request patch model=%s messages=%d", + full_model_name, + len(follow_up_messages), + ) + + tools_param = optional_params.get("tools") + optional_params_clean = { + k: v + for k, v in optional_params.items() + if k + not in { + "tools", + "extra_body", + "model_alias_map", + "stream_response", + "custom_prompt_dict", + } + } + if tools_param is not None: + optional_params_clean["tools"] = tools_param + + return AgenticLoopRequestPatch( + model=full_model_name, + messages=follow_up_messages, + optional_params=optional_params_clean, + kwargs=kwargs_for_followup, + ) + + async def _create_empty_fetch_result(self) -> str: + """Create an empty fetch result for tool calls without URLs""" + return "No URL provided for fetch" + + @staticmethod + def initialize_from_proxy_config( + litellm_settings: Dict[str, Any], + callback_specific_params: Dict[str, Any], + ) -> "WebFetchInterceptionLogger": + """ + Static method to initialize WebFetchInterceptionLogger from proxy config. + + Used in callback_utils.py to simplify initialization logic. + """ + # Get webfetch_interception_params from litellm_settings or callback_specific_params + webfetch_params: Dict[str, Any] = {} + if "webfetch_interception_params" in litellm_settings: + webfetch_params = litellm_settings["webfetch_interception_params"] + elif "webfetch_interception" in callback_specific_params: + webfetch_params = callback_specific_params["webfetch_interception"] + + # Use classmethod to initialize from config + return WebFetchInterceptionLogger.from_config_yaml(webfetch_params) diff --git a/litellm/integrations/webfetch_interception/tools.py b/litellm/integrations/webfetch_interception/tools.py new file mode 100644 index 00000000000..cee5cd0d5a4 --- /dev/null +++ b/litellm/integrations/webfetch_interception/tools.py @@ -0,0 +1,181 @@ +""" +LiteLLM WebFetch Tool Definition + +This module defines the standard web fetch tool used across LiteLLM. +Native provider tools (like Anthropic's web_fetch_20250305) are converted +to this format for consistent interception and execution. +""" + +from typing import Any, Dict, List, Optional + +from litellm.constants import LITELLM_WEB_FETCH_TOOL_NAME + + +# Native fetch tool formats that should be converted to LiteLLM standard +LITELLM_NATIVE_FETCH_TOOLS = [ + "web_fetch_20250305", # Anthropic native format + "web_fetch", # Claude Code CLI + "WebFetch", # Legacy format +] + + +def get_litellm_web_fetch_tool() -> Dict[str, Any]: + """ + Get the standard LiteLLM web fetch tool definition (Anthropic format). + + This is the canonical tool definition that all native web fetch tools + (like Anthropic's web_fetch_20250305, Claude Code's web_fetch, etc.) + are converted to for interception. + + Returns: + Dict containing the Anthropic-style tool definition with: + - name: Tool name + - description: What the tool does + - input_schema: JSON schema for tool parameters + """ + return { + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "description": ( + "Fetch and read the content of a specific URL. " + "Use this when you need to read or analyze the content of a web page." + ), + "input_schema": { + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "The URL to fetch and read content from", + } + }, + "required": ["url"], + }, + } + + +def get_litellm_web_fetch_tool_openai() -> Dict[str, Any]: + """ + Get the standard LiteLLM web fetch tool definition in OpenAI format. + + Used by async_pre_call_deployment_hook which runs in the chat completions + path where tools must be in OpenAI format (type: "function" with + function.parameters). + + Returns: + Dict containing the OpenAI-style tool definition. + """ + return { + "type": "function", + "function": { + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "description": ( + "Fetch and read the content of a specific URL. " + "Use this when you need to read or analyze the content of a web page." + ), + "parameters": { + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "The URL to fetch and read content from", + } + }, + "required": ["url"], + }, + }, + } + + +def is_web_fetch_tool_chat_completion(tool: Dict[str, Any]) -> bool: + """ + Check if a tool is a web fetch tool for Chat Completions API (strict check). + + This is a stricter version that ONLY checks for the exact LiteLLM web fetch tool name. + Use this for Chat Completions API to avoid false positives with user-defined tools. + + Detects ONLY: + - LiteLLM standard: name == "litellm_web_fetch" (Anthropic format) + - OpenAI format: type == "function" with function.name == "litellm_web_fetch" + + Args: + tool: Tool dictionary to check + + Returns: + True if tool is exactly the LiteLLM web fetch tool + """ + tool_name = tool.get("name", "") + tool_type = tool.get("type", "") + + # Check for OpenAI format: {"type": "function", "function": {"name": "litellm_web_fetch"}} + if tool_type == "function" and "function" in tool: + function_def = tool.get("function", {}) + function_name = function_def.get("name", "") + if function_name == LITELLM_WEB_FETCH_TOOL_NAME: + return True + + # Check for LiteLLM standard tool (Anthropic format) + if tool_name == LITELLM_WEB_FETCH_TOOL_NAME: + return True + + return False + + +def is_web_fetch_tool(tool: Dict[str, Any]) -> bool: + """ + Check if a tool is a web fetch tool (native or LiteLLM standard). + + Detects: + - LiteLLM standard: name == "litellm_web_fetch" + - OpenAI format: type == "function" with function.name == "litellm_web_fetch" + - Anthropic native: type starts with "web_fetch_" (e.g., "web_fetch_20250305") + - Claude Code: name == "web_fetch" with a type field + - Custom: name == "WebFetch" (legacy format) + + Args: + tool: Tool dictionary to check + + Returns: + True if tool is a web fetch tool + """ + tool_name = tool.get("name", "") + tool_type = tool.get("type", "") + + # Check for OpenAI format: {"type": "function", "function": {"name": "..."}} + if tool_type == "function" and "function" in tool: + function_def = tool.get("function", {}) + function_name = function_def.get("name", "") + if function_name == LITELLM_WEB_FETCH_TOOL_NAME: + return True + + # Check for LiteLLM standard tool (Anthropic format) + if tool_name == LITELLM_WEB_FETCH_TOOL_NAME: + return True + + # Check for native Anthropic web_fetch_* types + if tool_type and str(tool_type).startswith("web_fetch_"): + return True + + # Check for Claude Code's web_fetch with a type field + if tool_name == "web_fetch" and tool_type: + return True + + # Check for legacy WebFetch format + if tool_name == "WebFetch": + return True + + return False + + +def is_native_fetch_tool(name: str, tool_type: Optional[str] = None) -> bool: + """Check if a tool name/type is a native fetch tool (not LiteLLM standard).""" + if name == LITELLM_WEB_FETCH_TOOL_NAME: + return False # Already standard + if name in LITELLM_NATIVE_FETCH_TOOLS: + return True + if tool_type and str(tool_type).startswith("web_fetch_"): + return True + return False + + +def convert_native_fetch_to_litellm(native_tool: Dict[str, Any]) -> Dict[str, Any]: + """Convert a native fetch tool to LiteLLM standard format.""" + return get_litellm_web_fetch_tool() diff --git a/litellm/integrations/webfetch_interception/transformation.py b/litellm/integrations/webfetch_interception/transformation.py new file mode 100644 index 00000000000..cad4a9447d4 --- /dev/null +++ b/litellm/integrations/webfetch_interception/transformation.py @@ -0,0 +1,343 @@ +""" +WebFetch Tool Transformation + +Transforms between Anthropic/OpenAI tool_use format and LiteLLM fetch format. +""" + +from typing import Any, Dict, List, Optional, Tuple, Union + +from litellm._logging import verbose_logger +from litellm.constants import LITELLM_WEB_FETCH_TOOL_NAME +from litellm.llms.base_llm.fetch.transformation import WebFetchResponse + + +class WebFetchTransformation: + """ + Transformation class for WebFetch tool interception. + + Handles transformation between: + - Anthropic tool_use format → LiteLLM fetch requests + - OpenAI tool_calls format → LiteLLM fetch requests + - LiteLLM WebFetchResponse → Anthropic/OpenAI tool_result format + """ + + @staticmethod + def transform_request( + response: Any, + stream: bool, + response_format: str = "anthropic", + ) -> Tuple[bool, List[Dict]]: + """ + Transform model response to extract WebFetch tool calls. + + Detects if response contains WebFetch tool_use/tool_calls blocks and extracts + the URLs for execution. + + Args: + response: Model response (dict, AnthropicMessagesResponse, or ModelResponse) + stream: Whether response is streaming + response_format: Response format - "anthropic" or "openai" (default: "anthropic") + + Returns: + (has_webfetch, tool_calls): + has_webfetch: True if WebFetch tool_use found + tool_calls: List of tool_use/tool_calls dicts with id, name, input/function + """ + if stream: + verbose_logger.warning( + "WebFetchInterception: Unexpected streaming response, skipping interception" + ) + return False, [] + + # Parse non-streaming response based on format + if response_format == "openai": + return WebFetchTransformation._detect_from_openai_response(response) + else: + return WebFetchTransformation._detect_from_non_streaming_response(response) + + @staticmethod + def _detect_from_non_streaming_response( + response: Any, + ) -> Tuple[bool, List[Dict]]: + """Parse non-streaming response for WebFetch tool_use""" + + # Handle both dict and object responses + if isinstance(response, dict): + content = response.get("content", []) + else: + if not hasattr(response, "content"): + verbose_logger.debug( + "WebFetchInterception: Response has no content attribute" + ) + return False, [] + content = response.content or [] + + if not content: + verbose_logger.debug("WebFetchInterception: Response has empty content") + return False, [] + + # Find all WebFetch tool_use blocks + tool_calls = [] + for block in content: + # Handle both dict and object blocks + if isinstance(block, dict): + block_type = block.get("type") + block_name = block.get("name") + block_id = block.get("id") + block_input = block.get("input", {}) + else: + block_type = getattr(block, "type", None) + block_name = getattr(block, "name", None) + block_id = getattr(block, "id", None) + block_input = getattr(block, "input", {}) or {} + + # Check for web fetch tool_use blocks + if block_type == "tool_use" and block_name == LITELLM_WEB_FETCH_TOOL_NAME: + tool_calls.append( + { + "id": block_id, + "name": block_name, + "input": block_input, + } + ) + verbose_logger.debug( + f"WebFetchInterception: Detected tool_use for fetch: " + f"id={block_id}, url={block_input.get('url')}" + ) + + if tool_calls: + verbose_logger.debug( + f"WebFetchInterception: Found {len(tool_calls)} fetch tool call(s)" + ) + return True, tool_calls + + return False, [] + + @staticmethod + def _detect_from_openai_response( + response: Any, + ) -> Tuple[bool, List[Dict]]: + """Parse OpenAI chat completion response for WebFetch tool_calls""" + + # Handle ModelResponse (litellm type) + if hasattr(response, "choices") and response.choices: + choice = response.choices[0] + message = getattr(choice, "message", None) + if message and hasattr(message, "tool_calls"): + tool_calls_raw = message.tool_calls + else: + tool_calls_raw = None + elif isinstance(response, dict): + choices = response.get("choices", []) + if choices: + message = choices[0].get("message", {}) + tool_calls_raw = message.get("tool_calls") + else: + tool_calls_raw = None + else: + tool_calls_raw = None + + if not tool_calls_raw: + return False, [] + + # Filter for WebFetch tool_calls + tool_calls = [] + for tool_call in tool_calls_raw: + if isinstance(tool_call, dict): + function = tool_call.get("function", {}) + tool_name = function.get("name", "") + tool_id = tool_call.get("id", "") + tool_input = function.get("arguments", "") + else: + function = getattr(tool_call, "function", None) + if function: + tool_name = getattr(function, "name", "") + tool_id = getattr(tool_call, "id", "") + tool_input = getattr(function, "arguments", "") + else: + continue + + if tool_name == LITELLM_WEB_FETCH_TOOL_NAME: + # Parse arguments from JSON string if needed + if isinstance(tool_input, str): + try: + import json + + tool_input = json.loads(tool_input) + except (json.JSONDecodeError, TypeError): + tool_input = {} + + tool_calls.append( + { + "id": tool_id, + "function": { + "name": tool_name, + "arguments": tool_input, + }, + } + ) + verbose_logger.debug( + f"WebFetchInterception: Detected tool_call for fetch: " + f"id={tool_id}, url={tool_input.get('url') if isinstance(tool_input, dict) else 'unknown'}" + ) + + if tool_calls: + verbose_logger.debug( + f"WebFetchInterception: Found {len(tool_calls)} fetch tool call(s)" + ) + return True, tool_calls + + return False, [] + + @staticmethod + def transform_response( + tool_calls: List[Dict], + fetch_results: List[str], + response_format: str = "anthropic", + thinking_blocks: Optional[List[Dict]] = None, + ) -> Tuple[Dict, Union[Dict, List[Dict]]]: + """ + Transform fetch results into tool_result format for follow-up request. + + Args: + tool_calls: List of tool_use/tool_calls dicts + fetch_results: List of fetch result strings (markdown content) + response_format: "anthropic" or "openai" + thinking_blocks: Optional thinking blocks to preserve + + Returns: + (assistant_message, tool_result_message(s)): + assistant_message: Message with tool_use/tool_calls + tool_result_message: tool_result message(s) with fetch results + """ + if response_format == "openai": + return WebFetchTransformation._build_openai_messages( + tool_calls, fetch_results + ) + else: + return WebFetchTransformation._build_anthropic_messages( + tool_calls, fetch_results, thinking_blocks + ) + + @staticmethod + def _build_anthropic_messages( + tool_calls: List[Dict], + fetch_results: List[str], + thinking_blocks: Optional[List[Dict]] = None, + ) -> Tuple[Dict, Dict]: + """Build Anthropic-format assistant and user messages with tool results""" + + # Build assistant message with tool_use blocks + assistant_content = [] + + # Preserve thinking blocks first + if thinking_blocks: + assistant_content.extend(thinking_blocks) + + for i, tool_call in enumerate(tool_calls): + assistant_content.append( + { + "type": "tool_use", + "id": tool_call["id"], + "name": tool_call["name"], + "input": tool_call.get("input", {}), + } + ) + + assistant_message = { + "role": "assistant", + "content": assistant_content, + } + + # Build user message with tool_result blocks + tool_result_content = [] + for i, result in enumerate(fetch_results): + tool_call = tool_calls[i] if i < len(tool_calls) else {} + tool_result_content.append( + { + "type": "tool_result", + "tool_use_id": tool_call.get("id", "unknown"), + "content": result, + } + ) + + user_message = { + "role": "user", + "content": tool_result_content, + } + + return assistant_message, user_message + + @staticmethod + def _build_openai_messages( + tool_calls: List[Dict], + fetch_results: List[str], + ) -> Tuple[Dict, List[Dict]]: + """Build OpenAI-format assistant and tool messages""" + + # Build assistant message with tool_calls + openai_tool_calls = [] + for i, tool_call in enumerate(tool_calls): + function_data = tool_call.get("function", {}) + input_data = tool_call.get("input", {}) + arguments = function_data.get("arguments", "") + + # If arguments is empty or not a string, use input + if not arguments and input_data: + import json + + arguments = json.dumps(input_data) + elif not isinstance(arguments, str): + import json + + arguments = json.dumps(arguments) + + openai_tool_calls.append( + { + "id": tool_call.get("id", f"call_{i}"), + "type": "function", + "function": { + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "arguments": arguments, + }, + } + ) + + assistant_message = { + "role": "assistant", + "content": None, + "tool_calls": openai_tool_calls, + } + + # Build tool result messages + tool_messages = [] + for i, result in enumerate(fetch_results): + tool_call = tool_calls[i] if i < len(tool_calls) else {} + tool_messages.append( + { + "role": "tool", + "tool_call_id": tool_call.get("id", f"call_{i}"), + "content": result, + } + ) + + return assistant_message, tool_messages + + @staticmethod + def format_fetch_response(fetch_response: WebFetchResponse) -> str: + """ + Format a WebFetchResponse into a text string for LLM consumption. + + Args: + fetch_response: The fetch response object + + Returns: + Formatted string with title and content + """ + parts = [] + if fetch_response.title: + parts.append(f"Title: {fetch_response.title}") + parts.append(f"URL: {fetch_response.url}") + parts.append("") + parts.append(fetch_response.content) + return "\n".join(parts) diff --git a/litellm/llms/base_llm/fetch/__init__.py b/litellm/llms/base_llm/fetch/__init__.py new file mode 100644 index 00000000000..35facc539ea --- /dev/null +++ b/litellm/llms/base_llm/fetch/__init__.py @@ -0,0 +1,13 @@ +""" +Base Fetch transformation configuration. +""" + +from litellm.llms.base_llm.fetch.transformation import ( + BaseFetchConfig, + WebFetchResponse, +) + +__all__ = [ + "BaseFetchConfig", + "WebFetchResponse", +] diff --git a/litellm/llms/base_llm/fetch/transformation.py b/litellm/llms/base_llm/fetch/transformation.py new file mode 100644 index 00000000000..2a37aae39cf --- /dev/null +++ b/litellm/llms/base_llm/fetch/transformation.py @@ -0,0 +1,85 @@ +""" +Base Fetch transformation configuration. +""" + +from abc import ABC, abstractmethod +from typing import Any, Dict, Optional + +from pydantic import BaseModel, ConfigDict + + +class WebFetchResponse(BaseModel): + """Standard WebFetch response format.""" + + url: str + title: Optional[str] = None + content: str + metadata: Optional[Dict[str, Any]] = None + + model_config = ConfigDict(extra="allow") + + +class BaseFetchConfig(ABC): + """ + Base configuration for Fetch transformations. + Handles provider-agnostic Fetch operations. + """ + + def __init__(self) -> None: + pass + + @staticmethod + def ui_friendly_name() -> str: + """ + UI-friendly name for the fetch provider. + Override in provider-specific implementations. + """ + return "Unknown Fetch Provider" + + def validate_environment( + self, + headers: Dict, + api_key: Optional[str] = None, + api_base: Optional[str] = None, + **kwargs, + ) -> Dict: + """ + Validate environment and return headers. + Override in provider-specific implementations. + """ + return headers + + @abstractmethod + async def afetch_url( + self, + url: str, + headers: Dict[str, str], + optional_params: Dict[str, Any], + ) -> WebFetchResponse: + """ + Fetch content from a URL. + + Args: + url: URL to fetch + headers: HTTP headers (including auth) + optional_params: Optional parameters for the request + + Returns: + WebFetchResponse with content + """ + pass + + def get_error_class( + self, + error_message: str, + status_code: int, + headers: dict, + ) -> Exception: + """Get appropriate error class for the provider.""" + from litellm.llms.base_llm.chat.transformation import BaseLLMException + + return BaseLLMException( + status_code=status_code, + message=error_message, + headers=headers, + ) diff --git a/litellm/llms/firecrawl/fetch/__init__.py b/litellm/llms/firecrawl/fetch/__init__.py new file mode 100644 index 00000000000..03881e04dec --- /dev/null +++ b/litellm/llms/firecrawl/fetch/__init__.py @@ -0,0 +1,7 @@ +""" +Firecrawl Fetch API module. +""" + +from litellm.llms.firecrawl.fetch.transformation import FirecrawlFetchConfig + +__all__ = ["FirecrawlFetchConfig"] diff --git a/litellm/llms/firecrawl/fetch/transformation.py b/litellm/llms/firecrawl/fetch/transformation.py new file mode 100644 index 00000000000..3c82ffbde3e --- /dev/null +++ b/litellm/llms/firecrawl/fetch/transformation.py @@ -0,0 +1,121 @@ +""" +Firecrawl Fetch API module. + +Calls Firecrawl's /scrape endpoint to fetch and scrape web content. +""" + +import json +from typing import Any, Dict, Optional + +import httpx + +from litellm.llms.base_llm.fetch.transformation import BaseFetchConfig, WebFetchResponse +from litellm.secret_managers.main import get_secret_str + + +class FirecrawlFetchConfig(BaseFetchConfig): + """ + Firecrawl fetch configuration. + Uses the /scrape endpoint to fetch and convert web content to markdown. + """ + + FIRECRAWL_API_BASE = "https://api.firecrawl.dev/v1" + + @staticmethod + def ui_friendly_name() -> str: + return "Firecrawl" + + def validate_environment( + self, + headers: Dict, + api_key: Optional[str] = None, + api_base: Optional[str] = None, + **kwargs, + ) -> Dict: + """ + Validate environment and return headers. + """ + api_key = api_key or get_secret_str("FIRECRAWL_API_KEY") + if not api_key: + raise ValueError( + "FIRECRAWL_API_KEY is not set. Set `FIRECRAWL_API_KEY` environment variable." + ) + headers["Authorization"] = f"Bearer {api_key}" + headers["Content-Type"] = "application/json" + return headers + + def get_complete_url(self, api_base: Optional[str] = None) -> str: + """Get complete URL for Fetch endpoint.""" + api_base = ( + api_base or get_secret_str("FIRECRAWL_API_BASE") or self.FIRECRAWL_API_BASE + ) + + # Append "/scrape" to the api base if it's not already there + if not api_base.endswith("/scrape"): + api_base = f"{api_base}/scrape" + + return api_base + + async def afetch_url( + self, + url: str, + headers: Dict[str, str], + optional_params: Dict[str, Any], + ) -> WebFetchResponse: + """ + Fetch content from a URL using Firecrawl's /scrape endpoint. + + Args: + url: URL to fetch + headers: HTTP headers (including auth) + optional_params: Optional parameters for the request + - formats: List of formats to return (default: ["markdown"]) + - onlyMainContent: bool (default: True) + - includeTags: List[str] - tags to include + - excludeTags: List[str] - tags to exclude + - headers: Dict[str, str] - custom headers + - waitFor: int - time to wait in ms + - timeout: int - timeout in ms + + Returns: + WebFetchResponse with content + """ + request_data = { + "url": url, + "formats": optional_params.get("formats", ["markdown"]), + "onlyMainContent": optional_params.get("onlyMainContent", True), + } + + # Add optional parameters + for key in ["includeTags", "excludeTags", "headers", "waitFor", "timeout"]: + if key in optional_params: + request_data[key] = optional_params[key] + + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + self.get_complete_url(), + headers=headers, + json=request_data, + ) + + if response.status_code != 200: + raise self.get_error_class( + error_message=f"Firecrawl fetch failed: {response.status_code} - {response.text}", + status_code=response.status_code, + headers=dict(response.headers), + ) + + data = response.json() + + # Firecrawl response format: {"data": {"markdown": "...", "metadata": {...}}} + response_data = data.get("data", {}) + content = response_data.get("markdown") or response_data.get("content", "") + metadata = response_data.get("metadata", {}) + title = metadata.get("title") if isinstance(metadata, dict) else None + + return WebFetchResponse( + url=url, + title=title, + content=content, + metadata=metadata if isinstance(metadata, dict) else None, + ) diff --git a/litellm/proxy/common_utils/callback_utils.py b/litellm/proxy/common_utils/callback_utils.py index 7ddd722a80e..091974e4a65 100644 --- a/litellm/proxy/common_utils/callback_utils.py +++ b/litellm/proxy/common_utils/callback_utils.py @@ -295,6 +295,18 @@ def initialize_callbacks_on_proxy( # noqa: PLR0915 ) ) imported_list.append(websearch_interception_obj) + elif isinstance(callback, str) and callback == "webfetch_interception": + from litellm.integrations.webfetch_interception.handler import ( + WebFetchInterceptionLogger, + ) + + webfetch_interception_obj = ( + WebFetchInterceptionLogger.initialize_from_proxy_config( + litellm_settings=litellm_settings, + callback_specific_params=callback_specific_params, + ) + ) + imported_list.append(webfetch_interception_obj) elif isinstance(callback, str) and callback == "datadog_cost_management": from litellm.integrations.datadog.datadog_cost_management import ( DatadogCostManagementLogger, diff --git a/litellm/router.py b/litellm/router.py index 7448cdd1b47..17787e841ea 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -148,6 +148,7 @@ from litellm.types.router import ( CustomRoutingStrategyBase, Deployment, DeploymentTypedDict, + FetchToolTypedDict, GuardrailTypedDict, LiteLLM_Params, MockRouterTestingParams, @@ -239,6 +240,8 @@ class Router: assistants_config: Optional[AssistantsTypedDict] = None, ## SEARCH API ## search_tools: Optional[List[SearchToolTypedDict]] = None, + ## FETCH API ## + fetch_tools: Optional[List[FetchToolTypedDict]] = None, ## GUARDRAIL API ## guardrail_list: Optional[List[GuardrailTypedDict]] = None, ## CACHING ## @@ -411,6 +414,7 @@ class Router: self.assistants_config = assistants_config self.search_tools = search_tools or [] + self.fetch_tools = fetch_tools or [] self.guardrail_list = guardrail_list or [] self.deployment_names: List = ( [] diff --git a/litellm/types/integrations/webfetch_interception.py b/litellm/types/integrations/webfetch_interception.py new file mode 100644 index 00000000000..6bbd587e66f --- /dev/null +++ b/litellm/types/integrations/webfetch_interception.py @@ -0,0 +1,23 @@ +""" +Type definitions for WebFetch Interception integration. +""" + +from typing import List, Optional, TypedDict + + +class WebFetchInterceptionConfig(TypedDict, total=False): + """ + Configuration parameters for WebFetchInterceptionLogger. + + Used in proxy_config.yaml under litellm_settings: + litellm_settings: + webfetch_interception_params: + enabled_providers: ["bedrock"] + fetch_tool_name: "my-firecrawl-fetch" + """ + + enabled_providers: List[str] + """List of LLM provider names to enable interception for (e.g., ['bedrock', 'vertex_ai'])""" + + fetch_tool_name: Optional[str] + """Name of fetch tool configured in router's fetch_tools. If None, uses first available.""" diff --git a/litellm/types/router.py b/litellm/types/router.py index 33102f9ec48..b585d5c221a 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -586,6 +586,36 @@ class SearchToolTypedDict(TypedDict, total=False): search_tool_info: SearchToolInfoTypedDict +class FetchToolLiteLLMParams(TypedDict, total=False): + """ + LiteLLM params for fetch tools. + """ + + fetch_provider: Required[str] + api_key: Optional[str] + api_base: Optional[str] + timeout: Optional[Union[float, str, httpx.Timeout]] + max_retries: Optional[int] + + +class FetchToolTypedDict(TypedDict, total=False): + """ + Configuration for a fetch tool in the router. + + Example: + { + "fetch_tool_name": "litellm-fetch", + "litellm_params": { + "fetch_provider": "firecrawl", + "api_key": "os.environ/FIRECRAWL_API_KEY" + } + } + """ + + fetch_tool_name: Required[str] + litellm_params: Required[FetchToolLiteLLMParams] + + class GuardrailLiteLLMParams(TypedDict, total=False): """ LiteLLM params for guardrails. diff --git a/litellm/utils.py b/litellm/utils.py index 027c9fedced..0092435b57c 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -9192,6 +9192,23 @@ class ProviderConfigManager: return None return config_class() + @staticmethod + def get_provider_fetch_config( + provider: str, + ) -> Optional["BaseFetchConfig"]: + """ + Get Fetch configuration for a given provider. + """ + from litellm.llms.firecrawl.fetch.transformation import FirecrawlFetchConfig + + PROVIDER_TO_CONFIG_MAP = { + "firecrawl": FirecrawlFetchConfig, + } + config_class = PROVIDER_TO_CONFIG_MAP.get(provider, None) + if config_class is None: + return None + return config_class() + @staticmethod def get_provider_text_to_speech_config( model: str, diff --git a/tests/webfetch_tests/test_firecrawl_fetch.py b/tests/webfetch_tests/test_firecrawl_fetch.py new file mode 100644 index 00000000000..37f0849e003 --- /dev/null +++ b/tests/webfetch_tests/test_firecrawl_fetch.py @@ -0,0 +1,123 @@ +""" +WebFetch tests for Firecrawl fetch provider. +""" + +import pytest +from unittest.mock import AsyncMock, Mock, patch + +from litellm.llms.firecrawl.fetch.transformation import FirecrawlFetchConfig +from litellm.llms.base_llm.fetch.transformation import WebFetchResponse + + +class TestFirecrawlFetchConfig: + """Test Firecrawl fetch configuration.""" + + @pytest.fixture + def fetch_config(self): + return FirecrawlFetchConfig() + + @pytest.mark.asyncio + async def test_validate_environment_with_api_key(self, fetch_config): + """Test that API key validation works.""" + headers = {} + result = fetch_config.validate_environment(headers, api_key="test-key") + assert result["Authorization"] == "Bearer test-key" + assert result["Content-Type"] == "application/json" + + @pytest.mark.asyncio + async def test_validate_environment_without_api_key(self, fetch_config): + """Test that missing API key raises ValueError.""" + headers = {} + with patch( + "litellm.llms.firecrawl.fetch.transformation.get_secret_str", + return_value=None, + ): + with pytest.raises(ValueError, match="FIRECRAWL_API_KEY is not set"): + fetch_config.validate_environment(headers) + + def test_get_complete_url_default(self, fetch_config): + """Test default API base URL.""" + url = fetch_config.get_complete_url() + assert url == "https://api.firecrawl.dev/v1/scrape" + + def test_get_complete_url_custom(self, fetch_config): + """Test custom API base URL.""" + url = fetch_config.get_complete_url(api_base="https://custom.firecrawl.dev") + assert url == "https://custom.firecrawl.dev/scrape" + + @pytest.mark.asyncio + async def test_afetch_url_success(self, fetch_config): + """Test successful fetch.""" + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": { + "markdown": "# Test Content", + "metadata": {"title": "Test Title"}, + } + } + mock_response.headers = {} + + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.post = AsyncMock(return_value=mock_response) + + with patch("httpx.AsyncClient", return_value=mock_client): + result = await fetch_config.afetch_url( + url="https://example.com", + headers={"Authorization": "Bearer test-key"}, + optional_params={}, + ) + + assert isinstance(result, WebFetchResponse) + assert result.url == "https://example.com" + assert result.title == "Test Title" + assert result.content == "# Test Content" + + @pytest.mark.asyncio + async def test_afetch_url_error(self, fetch_config): + """Test fetch with error response.""" + mock_response = Mock() + mock_response.status_code = 500 + mock_response.text = "Internal Server Error" + mock_response.headers = {} + + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.post = AsyncMock(return_value=mock_response) + + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(Exception, match="Firecrawl fetch failed"): + await fetch_config.afetch_url( + url="https://example.com", + headers={"Authorization": "Bearer test-key"}, + optional_params={}, + ) + + +class TestWebFetchResponse: + """Test WebFetchResponse data model.""" + + def test_basic_response(self): + """Test creating a basic response.""" + response = WebFetchResponse( + url="https://example.com", + content="Test content", + ) + assert response.url == "https://example.com" + assert response.content == "Test content" + assert response.title is None + assert response.metadata is None + + def test_full_response(self): + """Test creating a response with all fields.""" + response = WebFetchResponse( + url="https://example.com", + title="Test Title", + content="Test content", + metadata={"author": "Test"}, + ) + assert response.title == "Test Title" + assert response.metadata == {"author": "Test"} diff --git a/tests/webfetch_tests/test_webfetch_interception.py b/tests/webfetch_tests/test_webfetch_interception.py new file mode 100644 index 00000000000..96fad36a530 --- /dev/null +++ b/tests/webfetch_tests/test_webfetch_interception.py @@ -0,0 +1,388 @@ +""" +WebFetch interception tests. +""" + +import pytest +from unittest.mock import AsyncMock, Mock, patch + +from litellm.constants import LITELLM_WEB_FETCH_TOOL_NAME +from litellm.integrations.webfetch_interception.handler import ( + WebFetchInterceptionLogger, +) +from litellm.integrations.webfetch_interception.tools import ( + get_litellm_web_fetch_tool, + is_web_fetch_tool, +) +from litellm.integrations.webfetch_interception.transformation import ( + WebFetchTransformation, +) +from litellm.types.utils import LlmProviders + + +class TestWebFetchInterceptionLogger: + """Test WebFetchInterceptionLogger.""" + + @pytest.fixture + def logger(self): + return WebFetchInterceptionLogger( + enabled_providers=[LlmProviders.BEDROCK], + fetch_tool_name="my-firecrawl-fetch", + ) + + @pytest.fixture + def mock_fetch_response(self): + from litellm.llms.base_llm.fetch.transformation import WebFetchResponse + + return WebFetchResponse( + url="https://example.com", + title="Test Title", + content="# Test Content\n\nThis is test content.", + ) + + def test_init(self, logger): + """Test logger initialization.""" + assert logger.enabled_providers == ["bedrock"] + assert logger.fetch_tool_name == "my-firecrawl-fetch" + + def test_init_all_providers(self): + """Test logger initialization with all providers.""" + logger = WebFetchInterceptionLogger(enabled_providers=None) + assert logger.enabled_providers == ["bedrock"] # Default + + @pytest.mark.asyncio + async def test_async_pre_call_deployment_hook_no_tools(self, logger): + """Test deployment hook with no tools.""" + kwargs = {"model": "bedrock/claude-3", "custom_llm_provider": "bedrock"} + result = await logger.async_pre_call_deployment_hook(kwargs, None) + assert result is None + + @pytest.mark.asyncio + async def test_async_pre_call_deployment_hook_with_fetch_tool(self, logger): + """Test deployment hook converts native fetch tools.""" + native_tool = { + "type": "web_fetch_20250305", + "name": "web_fetch", + } + kwargs = { + "model": "bedrock/claude-3", + "custom_llm_provider": "bedrock", + "tools": [native_tool], + } + result = await logger.async_pre_call_deployment_hook(kwargs, None) + + assert result is not None + assert "tools" in result + assert len(result["tools"]) == 1 + assert result["tools"][0]["function"]["name"] == LITELLM_WEB_FETCH_TOOL_NAME + + @pytest.mark.asyncio + async def test_async_pre_call_deployment_hook_disallowed_provider(self, logger): + """Test deployment hook skips non-enabled providers.""" + kwargs = { + "model": "openai/gpt-4", + "custom_llm_provider": "openai", + "tools": [{"type": "web_fetch_20250305", "name": "web_fetch"}], + } + result = await logger.async_pre_call_deployment_hook(kwargs, None) + assert result is None + + @pytest.mark.asyncio + async def test_async_should_run_agentic_loop_no_fetch(self, logger): + """Test agentic loop detection with no fetch tool.""" + response = {"content": [{"type": "text", "text": "Hello"}]} + should_run, info = await logger.async_should_run_agentic_loop( + response=response, + model="bedrock/claude-3", + messages=[], + tools=[], + stream=False, + custom_llm_provider="bedrock", + kwargs={}, + ) + assert should_run is False + assert info == {} + + @pytest.mark.asyncio + async def test_async_should_run_agentic_loop_with_fetch(self, logger): + """Test agentic loop detection with fetch tool_use.""" + response = { + "content": [ + { + "type": "tool_use", + "id": "fetch_1", + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "input": {"url": "https://example.com"}, + } + ] + } + should_run, info = await logger.async_should_run_agentic_loop( + response=response, + model="bedrock/claude-3", + messages=[], + tools=[{"name": LITELLM_WEB_FETCH_TOOL_NAME}], + stream=False, + custom_llm_provider="bedrock", + kwargs={}, + ) + assert should_run is True + assert "tool_calls" in info + assert len(info["tool_calls"]) == 1 + assert info["tool_type"] == "webfetch" + + @pytest.mark.asyncio + async def test_async_should_run_chat_completion_agentic_loop_with_fetch( + self, logger + ): + """Test chat completion agentic loop detection.""" + # Create explicit mock structure for OpenAI-style response + response = Mock() + + # Mock choice with message containing tool_calls + choice_mock = Mock() + message_mock = Mock() + + # Mock function inside tool_call + function_mock = Mock() + function_mock.name = LITELLM_WEB_FETCH_TOOL_NAME + function_mock.arguments = '{"url": "https://example.com"}' + + # Mock tool_call + tool_call_mock = Mock() + tool_call_mock.id = "call_1" + tool_call_mock.function = function_mock + tool_call_mock.type = "function" + + message_mock.tool_calls = [tool_call_mock] + choice_mock.message = message_mock + response.choices = [choice_mock] + + should_run, info = await logger.async_should_run_chat_completion_agentic_loop( + response=response, + model="bedrock/claude-3", + messages=[], + tools=[ + {"type": "function", "function": {"name": LITELLM_WEB_FETCH_TOOL_NAME}} + ], + stream=False, + custom_llm_provider="bedrock", + kwargs={}, + ) + assert should_run is True + assert "tool_calls" in info + + @pytest.mark.asyncio + async def test_async_run_agentic_loop(self, logger, mock_fetch_response): + """Test full agentic loop execution.""" + tool_calls = [ + { + "id": "fetch_1", + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "input": {"url": "https://example.com"}, + } + ] + tools_dict = { + "tool_calls": tool_calls, + "tool_type": "webfetch", + "provider": "bedrock", + "response_format": "anthropic", + "thinking_blocks": [], + } + + # Mock the fetch execution + with patch.object( + logger, "_get_fetch_config", new_callable=AsyncMock + ) as mock_get_config: + mock_config = AsyncMock() + mock_config.afetch_url = AsyncMock(return_value=mock_fetch_response) + mock_config.ui_friendly_name.return_value = "Firecrawl" + mock_get_config.return_value = mock_config + + with patch( + "litellm.integrations.webfetch_interception.handler.anthropic_messages.acreate", + new_callable=AsyncMock, + ) as mock_acreate: + mock_acreate.return_value = { + "content": [{"type": "text", "text": "Result"}] + } + + result = await logger.async_run_agentic_loop( + tools=tools_dict, + model="bedrock/claude-3", + messages=[{"role": "user", "content": "Fetch this"}], + response={}, + anthropic_messages_provider_config=None, + anthropic_messages_optional_request_params={}, + logging_obj=None, + stream=False, + kwargs={}, + ) + + assert result is not None + + @pytest.mark.asyncio + async def test_async_run_chat_completion_agentic_loop( + self, logger, mock_fetch_response + ): + """Test chat completion agentic loop execution.""" + tool_calls = [ + { + "id": "call_1", + "function": { + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "arguments": '{"url": "https://example.com"}', + }, + } + ] + tools_dict = { + "tool_calls": tool_calls, + "tool_type": "webfetch", + "provider": "bedrock", + "response_format": "openai", + } + + # Mock the fetch execution + with patch.object( + logger, "_get_fetch_config", new_callable=AsyncMock + ) as mock_get_config: + mock_config = AsyncMock() + mock_config.afetch_url = AsyncMock(return_value=mock_fetch_response) + mock_config.ui_friendly_name.return_value = "Firecrawl" + mock_get_config.return_value = mock_config + + with patch( + "litellm.acompletion", + new_callable=AsyncMock, + ) as mock_acompletion: + mock_acompletion.return_value = { + "choices": [{"message": {"content": "Result"}}] + } + + result = await logger.async_run_chat_completion_agentic_loop( + tools=tools_dict, + model="bedrock/claude-3", + messages=[{"role": "user", "content": "Fetch this"}], + response={}, + optional_params={}, + logging_obj=None, + stream=False, + kwargs={}, + ) + + assert result is not None + + +class TestWebFetchTransformation: + """Test WebFetchTransformation.""" + + def test_transform_request_anthropic_no_fetch(self): + """Test no fetch detected in Anthropic response.""" + response = {"content": [{"type": "text", "text": "Hello"}]} + has_fetch, tool_calls = WebFetchTransformation.transform_request( + response, stream=False, response_format="anthropic" + ) + assert has_fetch is False + assert tool_calls == [] + + def test_transform_request_anthropic_with_fetch(self): + """Test fetch detected in Anthropic response.""" + response = { + "content": [ + { + "type": "tool_use", + "id": "fetch_1", + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "input": {"url": "https://example.com"}, + } + ] + } + has_fetch, tool_calls = WebFetchTransformation.transform_request( + response, stream=False, response_format="anthropic" + ) + assert has_fetch is True + assert len(tool_calls) == 1 + assert tool_calls[0]["input"]["url"] == "https://example.com" + + def test_transform_response_anthropic(self): + """Test Anthropic response transformation.""" + tool_calls = [ + { + "id": "fetch_1", + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "input": {"url": "https://example.com"}, + } + ] + fetch_results = ["Title: Test\n\nContent: Hello"] + + assistant_msg, user_msg = WebFetchTransformation.transform_response( + tool_calls, fetch_results, response_format="anthropic" + ) + + assert assistant_msg["role"] == "assistant" + assert assistant_msg["content"][0]["type"] == "tool_use" + + assert user_msg["role"] == "user" + assert user_msg["content"][0]["type"] == "tool_result" + + def test_transform_response_openai(self): + """Test OpenAI response transformation.""" + tool_calls = [ + { + "id": "call_1", + "function": { + "name": LITELLM_WEB_FETCH_TOOL_NAME, + "arguments": '{"url": "https://example.com"}', + }, + } + ] + fetch_results = ["Title: Test\n\nContent: Hello"] + + assistant_msg, tool_msgs = WebFetchTransformation.transform_response( + tool_calls, fetch_results, response_format="openai" + ) + + assert assistant_msg["role"] == "assistant" + assert "tool_calls" in assistant_msg + + assert isinstance(tool_msgs, list) + assert tool_msgs[0]["role"] == "tool" + + def test_format_fetch_response(self): + """Test fetch response formatting.""" + from litellm.llms.base_llm.fetch.transformation import WebFetchResponse + + response = WebFetchResponse( + url="https://example.com", + title="Test Title", + content="Test content", + ) + + formatted = WebFetchTransformation.format_fetch_response(response) + assert "Test Title" in formatted + assert "https://example.com" in formatted + assert "Test content" in formatted + + +class TestWebFetchTools: + """Test web fetch tools.""" + + def test_get_litellm_web_fetch_tool(self): + """Test standard fetch tool definition.""" + tool = get_litellm_web_fetch_tool() + assert tool["name"] == LITELLM_WEB_FETCH_TOOL_NAME + assert "url" in tool["input_schema"]["properties"] + assert "url" in tool["input_schema"]["required"] + + def test_is_web_fetch_tool_standard(self): + """Test detection of standard fetch tool.""" + assert is_web_fetch_tool({"name": LITELLM_WEB_FETCH_TOOL_NAME}) is True + + def test_is_web_fetch_tool_native_anthropic(self): + """Test detection of native Anthropic fetch tool.""" + assert ( + is_web_fetch_tool({"type": "web_fetch_20250305", "name": "web_fetch"}) + is True + ) + + def test_is_web_fetch_tool_not_fetch(self): + """Test that non-fetch tools are not detected.""" + assert is_web_fetch_tool({"name": "calculator"}) is False