feat: add WebFetch interception with Firecrawl support

Implements server-side WebFetch tool interception for LLMs that don't
natively support web fetch (e.g., Bedrock/Claude).

Features:
- Intercepts litellm_web_fetch tool calls via agentic loop
- Supports Anthropic native format (web_fetch_20250305) conversion
- Firecrawl integration via /scrape endpoint with markdown output
- Modular provider architecture (BaseFetchConfig pattern)
- Full coexistence with WebSearch interception (chaining supported)
- Router fetch_tools configuration (Pattern A)
- Proxy YAML config support (webfetch_interception params)
- No public API (transparent, internal only)

Files added:
- litellm/integrations/webfetch_interception/ (handler, tools, transformation)
- litellm/llms/base_llm/fetch/ (BaseFetchConfig, WebFetchResponse)
- litellm/llms/firecrawl/fetch/ (FirecrawlFetchConfig)
- litellm/types/integrations/webfetch_interception.py (types)
- tests/webfetch_tests/ (27 tests, all passing)

Files modified:
- litellm/constants.py (add LITELLM_WEB_FETCH_TOOL_NAME)
- litellm/router.py (add fetch_tools to Router)
- litellm/types/router.py (add FetchToolTypedDict)
- litellm/utils.py (add get_provider_fetch_config)
- litellm/proxy/common_utils/callback_utils.py (add webfetch_interception)

Tested: 27/27 tests passing

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"

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",
+]

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()

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)

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",
+]

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,
+        )

litellm/llms/firecrawl/fetch/__init__.py

@@ -0,0 +1,7 @@
+"""
+Firecrawl Fetch API module.
+"""
+
+from litellm.llms.firecrawl.fetch.transformation import FirecrawlFetchConfig
+
+__all__ = ["FirecrawlFetchConfig"]

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,
+            )

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,

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 = (
             []

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."""

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.

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,

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"}

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