diff --git a/litellm/integrations/websearch_interception/ARCHITECTURE.md b/litellm/integrations/websearch_interception/ARCHITECTURE.md
index ce7f01c5a2a..4ea7a7ae527 100644
--- a/litellm/integrations/websearch_interception/ARCHITECTURE.md
+++ b/litellm/integrations/websearch_interception/ARCHITECTURE.md
@@ -207,6 +207,59 @@ response = await litellm.messages.acreate(
 
 ---
 
+## Loop Ceiling
+
+One intercepted request can chain several follow-up model calls, since the model often searches again after
+reading the first set of results. `max_agentic_loops` caps how many of those follow-ups run, and it defaults
+to 3. LiteLLM also breaks the loop early when the model asks for the exact same tool call twice in a row.
+
+Set the ceiling on the feature, which the interceptor applies to `/v1/messages` requests:
+
+```yaml
+litellm_settings:
+  websearch_interception_params:
+    enabled_providers: ["bedrock"]
+    max_agentic_loops: 5
+```
+
+Or per deployment, which wins over the feature-level setting:
+
+```yaml
+model_list:
+  - model_name: claude-sonnet-4-5
+    litellm_params:
+      model: bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0
+      max_agentic_loops: 5
+```
+
+Clients cannot set it. `max_agentic_loops` is on the proxy's untrusted-field list, so a request body that
+carries it is ignored and one request can never drive an unbounded number of upstream model calls.
+
+Both places are validated at config load, and a value that is not an integer of at least 1 stops the proxy
+from starting rather than surfacing later. The per-deployment one is checked while the model list is read,
+not on `LiteLLM_Params`, because the proxy builds its router with `ignore_invalid_deployments=True` and a
+validator down there would drop the deployment silently instead of refusing to start.
+
+When the ceiling is reached on a non-streaming `/v1/messages` request, the turn ends there and the client gets
+the last response back with the internal `litellm_web_search` tool call removed and `stop_reason: end_turn`.
+The client never declared that tool, so leaving the block in would hand it a tool call it has no way to answer.
+The answer can be less complete than it would have been with more loops, which is the tradeoff the ceiling
+buys. Where the refused call was the only block left, the turn comes back with no text in it at all.
+
+Non-streaming is not a limitation on the client here, because a client that asked for a stream gets the same
+treatment. Interception converts an intercepted `stream=True` request to non-streaming before the loop runs and
+rebuilds the SSE stream from the finalized turn afterwards, so the ceiling is always reached on a response the
+client has not seen yet. `AgenticStreamingIterator` is the one caller that reaches the loop with its events
+already on the wire, and it keeps raising, because a finalized turn would arrive there as a second message
+rather than as a replacement.
+
+Two other surfaces do not get that treatment yet. `/v1/responses` returns its own shape that the finalizer does
+not rewrite, so it still hands back the internal call. And `/v1/chat/completions` runs its own copy of these
+rails in `litellm_core_utils/chat_completion_agentic_loop.py`, which still raises rather than ending the turn.
+Both are tracked separately
+
+---
+
 ## Streaming Support
 
 WebSearch interception works transparently with both streaming and non-streaming requests.
diff --git a/litellm/integrations/websearch_interception/handler.py b/litellm/integrations/websearch_interception/handler.py
index e59ef0449d0..13a16947fb4 100644
--- a/litellm/integrations/websearch_interception/handler.py
+++ b/litellm/integrations/websearch_interception/handler.py
@@ -31,6 +31,9 @@ from litellm.integrations.websearch_interception.tools import (
 from litellm.integrations.websearch_interception.transformation import (
     WebSearchTransformation,
 )
+from litellm.litellm_core_utils.agentic_loop_settings import (
+    validated_max_agentic_loops,
+)
 from litellm.llms.base_llm.search.transformation import SearchResponse
 from litellm.types.integrations.custom_logger import (
     CHAT_COMPLETION_AGENTIC_SURFACE,
@@ -122,6 +125,7 @@ class WebSearchInterceptionLogger(CustomLogger):
         self,
         enabled_providers: list[LlmProviders | str] | None = None,
         search_tool_name: str | None = None,
+        max_agentic_loops: int | None = None,
     ):
         """
         Args:
@@ -131,6 +135,9 @@ class WebSearchInterceptionLogger(CustomLogger):
                               Default: None (all providers enabled)
             search_tool_name: Name of search tool configured in router's search_tools.
                              If None, will attempt to use first available search tool.
+            max_agentic_loops: How many follow-up model calls one intercepted request
+                              may chain before the loop is refused and the turn ends.
+                              If None, LiteLLM's default of 3 applies.
         """
         super().__init__()
         # Convert enum values to strings for comparison
@@ -139,8 +146,16 @@ class WebSearchInterceptionLogger(CustomLogger):
         else:
             self.enabled_providers = [p.value if isinstance(p, LlmProviders) else p for p in enabled_providers]
         self.search_tool_name = search_tool_name
+        self.max_agentic_loops = self._validated_max_agentic_loops(max_agentic_loops)
         self._request_has_websearch = False  # Track if current request has web search
 
+    @staticmethod
+    def _validated_max_agentic_loops(max_agentic_loops: object) -> int | None:
+        """
+        Reject loop ceilings the agentic loop cannot honor, at config load time.
+        """
+        return validated_max_agentic_loops(max_agentic_loops, field="websearch_interception_params.max_agentic_loops")
+
     async def try_short_circuit_search(
         self,
         model: str,
@@ -398,6 +413,7 @@ class WebSearchInterceptionLogger(CustomLogger):
                   websearch_interception_params:
                     enabled_providers: ["bedrock"]
                     search_tool_name: "my-perplexity-search"
+                    max_agentic_loops: 5
 
             Usage:
                 config = litellm_settings.get("websearch_interception_params", {})
@@ -406,6 +422,7 @@ class WebSearchInterceptionLogger(CustomLogger):
         # Extract parameters from config
         enabled_providers_str: Final = config.get("enabled_providers", None)
         search_tool_name: Final = config.get("search_tool_name", None)
+        max_agentic_loops: Final = config.get("max_agentic_loops", None)
 
         # Convert string provider names to LlmProviders enum values
         enabled_providers: list[LlmProviders | str] | None = None
@@ -423,6 +440,7 @@ class WebSearchInterceptionLogger(CustomLogger):
         return cls(
             enabled_providers=enabled_providers,
             search_tool_name=search_tool_name,
+            max_agentic_loops=max_agentic_loops,
         )
 
     @staticmethod
@@ -493,6 +511,10 @@ class WebSearchInterceptionLogger(CustomLogger):
 
         verbose_logger.debug("WebSearchInterception: Pre-request hook triggered for provider=%s", custom_llm_provider)
 
+        deployment_max_agentic_loops: Final = kwargs.get("max_agentic_loops")
+        if self.max_agentic_loops is not None and deployment_max_agentic_loops is None:
+            kwargs["max_agentic_loops"] = self.max_agentic_loops  # rebind-ok: this hook returns the kwargs it edits
+
         # If the client sent an Anthropic-native web_search_* tool, mark the
         # request so the agentic loop emits native web_search_tool_result
         # blocks in the final response (for citations panels, etc.). The flag
diff --git a/litellm/litellm_core_utils/agentic_loop_settings.py b/litellm/litellm_core_utils/agentic_loop_settings.py
new file mode 100644
index 00000000000..3dd8d437aef
--- /dev/null
+++ b/litellm/litellm_core_utils/agentic_loop_settings.py
@@ -0,0 +1,59 @@
+"""
+Shared validation for the agentic loop ceiling.
+
+``max_agentic_loops`` can be set in two places, and the two disagreed about
+what a bad value means. The feature-level
+``litellm_settings.websearch_interception_params.max_agentic_loops`` was
+checked at config load, while a per-deployment
+``model_list[].litellm_params.max_agentic_loops`` was passed straight through
+to ``int(... or 3)``. That let a per-deployment ``0`` read as the default 3,
+turning the tightest ceiling into the loosest one, and let a per-deployment
+``"three"`` boot the proxy and then fail every request to that model.
+
+Both settings now go through :func:`validated_max_agentic_loops`, which names
+the field it rejected so the error says which line of the config to fix.
+
+Anything that spells a whole number is still accepted, because the old
+``int(... or 3)`` accepted those and a ceiling is routinely parameterized as
+``max_agentic_loops: os.environ/MAX_AGENTIC_LOOPS``, which resolves to a
+string. Rejecting ``"5"`` would stop such a proxy from booting on upgrade.
+"""
+
+from typing import Final
+
+DEFAULT_MAX_AGENTIC_LOOPS: Final = 3
+
+
+def _as_whole_number(value: object) -> int | None:
+    """
+    Return ``value`` as an int when it spells a whole number, else ``None``.
+
+    ``bool`` is excluded explicitly because it is an ``int`` subclass, so
+    ``max_agentic_loops: true`` would otherwise be read as a ceiling of 1.
+    """
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value) if value.is_integer() else None
+    if isinstance(value, str):
+        try:
+            return int(value.strip())
+        except ValueError:
+            return None
+    return None
+
+
+def validated_max_agentic_loops(max_agentic_loops: object, field: str) -> int | None:
+    """
+    Return ``max_agentic_loops`` as an int, or raise naming ``field``.
+    """
+    if max_agentic_loops is None:
+        return None
+    ceiling: Final = _as_whole_number(max_agentic_loops)
+    if ceiling is None:
+        raise TypeError(f"{field} must be an integer, got {max_agentic_loops!r}")
+    if ceiling < 1:
+        raise ValueError(f"{field} must be at least 1, got {ceiling}")
+    return ceiling
diff --git a/litellm/litellm_core_utils/chat_completion_agentic_loop.py b/litellm/litellm_core_utils/chat_completion_agentic_loop.py
index b91c1785a54..07bed1f88ad 100644
--- a/litellm/litellm_core_utils/chat_completion_agentic_loop.py
+++ b/litellm/litellm_core_utils/chat_completion_agentic_loop.py
@@ -5,6 +5,10 @@ from typing import Final, cast
 
 from litellm._logging import verbose_logger
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.agentic_loop_settings import (
+    DEFAULT_MAX_AGENTIC_LOOPS,
+    validated_max_agentic_loops,
+)
 from litellm.types.integrations.custom_logger import (
     CHAT_COMPLETION_AGENTIC_SURFACE,
     NON_CODE_INTERPRETER_INTERCEPTION_INTERNAL_PREFIXES,
@@ -52,7 +56,10 @@ def _coerce_int(value: object, default: int) -> int:
 
 def _agentic_loop_settings(kwargs: dict[str, object]) -> tuple[int, int, list[str]]:
     depth: Final = _coerce_int(kwargs.get("_agentic_loop_depth"), 0)
-    max_loops: Final = max(_coerce_int(kwargs.get("max_agentic_loops"), 3), 1)
+    configured: Final = validated_max_agentic_loops(
+        kwargs.get("max_agentic_loops"), field="litellm_params.max_agentic_loops"
+    )
+    max_loops: Final = DEFAULT_MAX_AGENTIC_LOOPS if configured is None else configured
     raw_fingerprints: Final = kwargs.get("_agentic_loop_fingerprints")
     fingerprints: Final = [str(fp) for fp in raw_fingerprints] if isinstance(raw_fingerprints, list) else []
     return depth, max_loops, fingerprints
diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/fake_stream_iterator.py b/litellm/llms/anthropic/experimental_pass_through/messages/fake_stream_iterator.py
index 215d4a5b42b..14f1b7697cf 100644
--- a/litellm/llms/anthropic/experimental_pass_through/messages/fake_stream_iterator.py
+++ b/litellm/llms/anthropic/experimental_pass_through/messages/fake_stream_iterator.py
@@ -113,6 +113,14 @@ class FakeAnthropicMessagesStreamIterator:
             }
             chunks.append(f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode())
 
+        else:
+            passthrough_start: Final = {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": block_dict,
+            }
+            chunks.append(f"event: content_block_start\ndata: {json.dumps(passthrough_start)}\n\n".encode())
+
         content_block_stop: Final = {"type": "content_block_stop", "index": index}
         chunks.append(f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode())
         return chunks
diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py
index 369e150f6bd..ed079197513 100644
--- a/litellm/llms/custom_httpx/llm_http_handler.py
+++ b/litellm/llms/custom_httpx/llm_http_handler.py
@@ -19,6 +19,10 @@ import litellm.types.utils
 from litellm._logging import _redact_string, verbose_logger
 from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta
 from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
+from litellm.litellm_core_utils.agentic_loop_settings import (
+    DEFAULT_MAX_AGENTIC_LOOPS,
+    validated_max_agentic_loops,
+)
 from litellm.litellm_core_utils.asyncify import run_async_function
 from litellm.litellm_core_utils.realtime_errors import realtime_error_event, websocket_close_reason
 from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming
@@ -89,6 +93,7 @@ from litellm.types.files import StreamingMediaUploadConfig, TwoStepFileUploadCon
 from litellm.types.integrations.custom_logger import (
     AgenticLoopPlan,
     AgenticLoopRequestPatch,
+    AgenticLoopSafetyError,
 )
 from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
@@ -5077,9 +5082,12 @@ class BaseLLMHTTPHandler:
     @staticmethod
     def _get_agentic_loop_settings(kwargs: dict) -> tuple[int, int, list[str]]:
         depth: Final = int(kwargs.get("_agentic_loop_depth", 0) or 0)
-        max_loops: Final = int(kwargs.get("max_agentic_loops", 3) or 3)
+        configured: Final = validated_max_agentic_loops(
+            kwargs.get("max_agentic_loops"), field="litellm_params.max_agentic_loops"
+        )
+        max_loops: Final = DEFAULT_MAX_AGENTIC_LOOPS if configured is None else configured
         fingerprints: Final = list(kwargs.get("_agentic_loop_fingerprints", []) or [])
-        return depth, max(max_loops, 1), fingerprints
+        return depth, max_loops, fingerprints
 
     @staticmethod
     def _has_agentic_completion_hook(logging_obj: LiteLLMLoggingObj) -> bool:
@@ -5122,7 +5130,8 @@ class BaseLLMHTTPHandler:
         """
         Evaluate agentic-loop safety guards (fingerprint cycle / max depth).
 
-        Raises ValueError on abort.  Returns the current fingerprint on success.
+        Raises AgenticLoopSafetyError on abort.  Returns the current fingerprint
+        on success.
 
         These checks must not be swallowed by the per-callback ``except Exception``
         block that wraps callback dispatch — they are bounded-loop / cycle-break
@@ -5130,9 +5139,9 @@ class BaseLLMHTTPHandler:
         """
         fingerprint: Final = BaseLLMHTTPHandler._fingerprint_agentic_tools(tool_calls)
         if fingerprint in fingerprints:
-            raise ValueError("Agentic loop detected repeated tool-call fingerprint; aborting rerun")
+            raise AgenticLoopSafetyError("Agentic loop detected repeated tool-call fingerprint; aborting rerun")
         if depth >= max_loops:
-            raise ValueError(f"Exceeded max_agentic_loops={max_loops} for model={model}")
+            raise AgenticLoopSafetyError(f"Exceeded max_agentic_loops={max_loops} for model={model}")
         return fingerprint
 
     @staticmethod
@@ -5142,6 +5151,97 @@ class BaseLLMHTTPHandler:
         except Exception:
             return str(tools)
 
+    @staticmethod
+    def _refused_agentic_tool_identifiers(tool_calls: object) -> tuple[frozenset[str], frozenset[str]]:
+        """
+        Collect the ids and names of the tool calls a safety rail just refused.
+
+        Callbacks hand back either a bare list of tool calls or a dict wrapping
+        that list under ``tool_calls``, and both the anthropic and responses
+        shapes carry an ``id`` (or ``call_id``) plus a ``name``.
+        """
+        calls: Final = tool_calls.get("tool_calls") if isinstance(tool_calls, dict) else tool_calls
+        if not isinstance(calls, list):
+            return frozenset(), frozenset()
+        dict_calls: Final = (call for call in calls if isinstance(call, dict))
+        fields: Final = tuple((call.get("id"), call.get("call_id"), call.get("name")) for call in dict_calls)
+        ids: Final = frozenset(
+            value for call_id, caller_id, _ in fields for value in (call_id, caller_id) if isinstance(value, str)
+        )
+        names: Final = frozenset(name for _, _, name in fields if isinstance(name, str))
+        return ids, names
+
+    @staticmethod
+    def _is_refused_tool_use_block(block: object, refused_ids: frozenset[str], refused_names: frozenset[str]) -> bool:
+        """
+        Whether this response block belongs to a tool call the rail refused.
+
+        An id settles it on its own, so a block carrying one is matched on the id
+        alone and a client's own tool call survives even where it happens to
+        share a name with a refused one. The name is only consulted for tool call
+        shapes that arrive without an id.
+        """
+        if not isinstance(block, dict) or block.get("type") != "tool_use":
+            return False
+        block_id: Final = block.get("id")
+        if isinstance(block_id, str) and refused_ids:
+            return block_id in refused_ids
+        return block.get("name") in refused_names
+
+    @staticmethod
+    def _can_replace_turn_with_terminal_response(stream: bool, api_surface: str) -> bool:
+        """
+        Whether a refused rerun can still be answered with a finalized turn.
+
+        Only the anthropic messages surface can. The responses surface carries a
+        pydantic model the finalizer does not rewrite, so it keeps raising, which
+        is what every surface did before this path learned to end the turn.
+
+        The messages and responses call sites pass ``stream=False``, because
+        interception converts an intercepted stream to non-streaming before the
+        loop runs and rebuilds the SSE stream from the finalized turn
+        afterwards. ``AgenticStreamingIterator`` passes ``stream=True``, and
+        that path keeps raising: its events are already on the wire, so a
+        finalized turn would reach the client as a second message rather than
+        as a replacement.
+        """
+        return not stream and api_surface == "anthropic_messages"
+
+    @staticmethod
+    def _finalize_refused_agentic_response(response: object, tool_calls: object) -> object:
+        """
+        Turn the response into a terminal turn after a safety rail refused the rerun.
+
+        The refused tool calls target tools LiteLLM injected on the client's
+        behalf, so a client that never declared them cannot send back a matching
+        ``tool_result``. Their blocks are dropped and a ``tool_use`` stop reason
+        is closed out as ``end_turn``, which is what a provider-native web search
+        turn returns once it stops calling tools.
+
+        A ``tool_use`` block the client itself declared is left alone, and while
+        one is still in the response the stop reason stays ``tool_use`` so the
+        client knows to answer it.
+        """
+        if not isinstance(response, dict):
+            return response
+
+        refused_ids, refused_names = BaseLLMHTTPHandler._refused_agentic_tool_identifiers(tool_calls)
+        finalized: Final = dict(response)
+        content: Final = finalized.get("content")
+        if isinstance(content, list):
+            kept_blocks: Final = [
+                block
+                for block in content
+                if not BaseLLMHTTPHandler._is_refused_tool_use_block(block, refused_ids, refused_names)
+            ]
+            finalized["content"] = kept_blocks
+            client_tool_use_remains: Final = any(
+                isinstance(block, dict) and block.get("type") == "tool_use" for block in kept_blocks
+            )
+            if not client_tool_use_remains and finalized.get("stop_reason") == "tool_use":
+                finalized["stop_reason"] = "end_turn"
+        return finalized
+
     async def _execute_anthropic_agentic_plan(
         self,
         plan: AgenticLoopPlan,
@@ -5507,14 +5607,30 @@ class BaseLLMHTTPHandler:
                 continue
 
             # Safety guards must run OUTSIDE the callback try/except — they are
-            # bounded-loop / cycle-break rails that must propagate to the caller.
-            fingerprint = self._check_agentic_loop_safety(
-                tool_calls=tool_calls,
-                fingerprints=fingerprints,
-                depth=depth,
-                max_loops=max_loops,
-                model=model,
-            )
+            # bounded-loop / cycle-break rails, not callback bugs.
+            try:
+                fingerprint = self._check_agentic_loop_safety(
+                    tool_calls=tool_calls,
+                    fingerprints=fingerprints,
+                    depth=depth,
+                    max_loops=max_loops,
+                    model=model,
+                )
+            except AgenticLoopSafetyError as e:
+                if not self._can_replace_turn_with_terminal_response(stream, api_surface):
+                    raise
+                _call_id = getattr(logging_obj, "litellm_call_id", "unknown")
+                verbose_logger.warning(
+                    "LiteLLM.AgenticLoopRefused: ending turn [call_id=%s model=%s]: %s",
+                    _call_id,
+                    model,
+                    str(e),
+                )
+                return self._maybe_wrap_in_fake_stream(
+                    self._finalize_refused_agentic_response(response=response, tool_calls=tool_calls),
+                    logging_obj,
+                    api_surface,
+                )
 
             try:
                 kwargs_with_provider = hook_kwargs.copy()
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 9f16584340d..7dced4e26b6 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -254,6 +254,9 @@ from litellm.exceptions import RejectedRequestError
 from litellm.integrations.custom_guardrail import ModifyResponseException
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
+from litellm.litellm_core_utils.agentic_loop_settings import (
+    validated_max_agentic_loops,
+)
 from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.core_helpers import (
     _get_parent_otel_span_from_kwargs,
@@ -4081,6 +4084,27 @@ def resolve_complexity_router_plugins(
         complexity_router_config["classifier_plugin"] = resolved_classifier  # rebind-ok: out-param, resolved in place
 
 
+def validate_deployment_max_agentic_loops(model: Mapping[str, Any]) -> None:
+    """
+    Reject a per-deployment `max_agentic_loops` the agentic loop cannot honor.
+
+    Checked here rather than on `LiteLLM_Params` because the proxy builds its
+    router with `ignore_invalid_deployments=True`, so a validator down there
+    turns a bad value into a silently missing model instead of a refusal to
+    start. Left unchecked entirely, a `0` used to read as the default ceiling
+    of 3 and a non-integer failed every request to that model instead.
+    """
+    litellm_params: Final = model.get("litellm_params") or {}
+    if "max_agentic_loops" not in litellm_params:
+        return
+
+    model_name: Final = model.get("model_name", "")
+    validated_max_agentic_loops(
+        litellm_params["max_agentic_loops"],
+        field=f"litellm_params.max_agentic_loops on model {model_name!r}",
+    )
+
+
 def pin_complexity_router_model_id(model: dict) -> None:  # mutable-ok: out-param, model_info is stamped in place
     """
     Stamps `model_info.id` from the raw litellm_params before plugin resolution swaps
@@ -5416,6 +5440,7 @@ class ProxyConfig:
                 for k, v in model["litellm_params"].items():
                     if isinstance(v, str) and v.startswith("os.environ/"):
                         model["litellm_params"][k] = get_secret(v)
+                validate_deployment_max_agentic_loops(model)
                 pin_complexity_router_model_id(model)
                 complexity_router_config = model["litellm_params"].get("complexity_router_config")
                 if isinstance(complexity_router_config, dict):
diff --git a/litellm/types/integrations/custom_logger.py b/litellm/types/integrations/custom_logger.py
index 89b85bc5114..2cca16351af 100644
--- a/litellm/types/integrations/custom_logger.py
+++ b/litellm/types/integrations/custom_logger.py
@@ -23,6 +23,21 @@ def is_interception_internal_key(
     return any(key.startswith(prefix) for prefix in prefixes)
 
 
+class AgenticLoopSafetyError(ValueError):
+    """
+    Raised when an agentic-loop safety rail refuses a rerun.
+
+    Covers both rails: the bounded-loop cap (``max_agentic_loops``) and the
+    repeated tool-call fingerprint cycle break. Subclasses ``ValueError`` so
+    callers that already catch the broader type keep working.
+
+    Only the anthropic messages loop raises this today. The chat completions
+    loop in ``litellm_core_utils/chat_completion_agentic_loop.py`` still raises
+    a plain ``ValueError`` from its own copy of the same rails, so catching
+    this type alone will not cover that surface until it is moved over.
+    """
+
+
 class StandardCustomLoggerInitParams(BaseModel):
     """
     Params for initializing a CustomLogger.
diff --git a/litellm/types/integrations/websearch_interception.py b/litellm/types/integrations/websearch_interception.py
index 90713b270be..7926b9eee0a 100644
--- a/litellm/types/integrations/websearch_interception.py
+++ b/litellm/types/integrations/websearch_interception.py
@@ -5,6 +5,7 @@ Type definitions for WebSearch Interception integration.
 from typing import Literal, TypedDict
 
 from pydantic import BaseModel
+from typing_extensions import ReadOnly
 
 
 class AnthropicSearchQuery(BaseModel):
@@ -35,6 +36,7 @@ class WebSearchInterceptionConfig(TypedDict, total=False):
           websearch_interception_params:
             enabled_providers: ["bedrock"]
             search_tool_name: "my-perplexity-search"
+            max_agentic_loops: 5
     """
 
     enabled_providers: list[str]
@@ -42,3 +44,6 @@ class WebSearchInterceptionConfig(TypedDict, total=False):
 
     search_tool_name: str | None
     """Name of search tool configured in router's search_tools. If None, uses first available."""
+
+    max_agentic_loops: ReadOnly[int | None]
+    """How many follow-up model calls one intercepted request may chain. If None, LiteLLM's default of 3 applies."""
diff --git a/tests/test_litellm/integrations/websearch_interception/test_websearch_agentic_loop_cap.py b/tests/test_litellm/integrations/websearch_interception/test_websearch_agentic_loop_cap.py
new file mode 100644
index 00000000000..40fd8c4e9e6
--- /dev/null
+++ b/tests/test_litellm/integrations/websearch_interception/test_websearch_agentic_loop_cap.py
@@ -0,0 +1,754 @@
+"""
+Unit tests for what an intercepted request returns once a safety rail refuses
+another agentic loop.
+
+The web search interception loop injects an internal tool (litellm_web_search)
+that the client never declared. When the loop cap or the repeated-fingerprint
+guard trips, the turn has to end with a terminal response: leaking that internal
+tool_use block leaves the client holding a tool call it cannot answer.
+
+Also covers the max_agentic_loops knob on websearch_interception_params, from
+config.yaml through to the settings the loop actually reads.
+"""
+
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.websearch_interception.handler import (
+    WebSearchInterceptionLogger,
+)
+from litellm.llms.anthropic.experimental_pass_through.messages.fake_stream_iterator import (
+    FakeAnthropicMessagesStreamIterator,
+)
+from litellm.litellm_core_utils.agentic_loop_settings import DEFAULT_MAX_AGENTIC_LOOPS
+from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+from litellm.secret_managers.main import get_secret
+from litellm.types.integrations.custom_logger import (
+    AgenticLoopPlan,
+    AgenticLoopRequestPatch,
+    AgenticLoopSafetyError,
+)
+
+INTERNAL_TOOL_NAME = "litellm_web_search"
+
+
+@pytest.fixture(autouse=True)
+def only_the_callbacks_these_tests_register(monkeypatch):
+    """
+    These tests drive the hooks with a callback of their own on the logging
+    object, so a logger another test left on litellm.callbacks would join the
+    run and change what the hooks do.
+    """
+    monkeypatch.setattr(litellm, "callbacks", [])
+
+
+def _internal_tool_use_block(block_id: str = "toolu_internal_1") -> dict:
+    return {
+        "id": block_id,
+        "type": "tool_use",
+        "name": INTERNAL_TOOL_NAME,
+        "input": {"query": "who won the world cup"},
+    }
+
+
+def _native_search_blocks(index: int = 1) -> list[dict]:
+    return [
+        {
+            "type": "server_tool_use",
+            "id": f"srvtoolu_{index}",
+            "name": "web_search",
+            "input": {"query": "who won the world cup"},
+        },
+        {
+            "type": "web_search_tool_result",
+            "tool_use_id": f"srvtoolu_{index}",
+            "content": [{"type": "web_search_result", "url": "https://example.com", "title": "Result"}],
+        },
+    ]
+
+
+def _response_asking_for_another_search(block_id: str = "toolu_internal_1") -> dict:
+    return {
+        "id": "msg_123",
+        "type": "message",
+        "role": "assistant",
+        "model": "claude-sonnet-4-5",
+        "content": [
+            *_native_search_blocks(index=1),
+            {"type": "text", "text": "Let me check one more source."},
+            _internal_tool_use_block(block_id),
+        ],
+        "stop_reason": "tool_use",
+        "usage": {"input_tokens": 10, "output_tokens": 5},
+    }
+
+
+def _block_types(response: dict) -> list[str]:
+    return [block["type"] for block in response["content"]]
+
+
+def _tool_use_names(response: dict) -> list[str]:
+    return [block.get("name") for block in response["content"] if block.get("type") == "tool_use"]
+
+
+class _InterceptingCallback(CustomLogger):
+    """
+    Stands in for the websearch interceptor: asks for another loop whenever the
+    response carries an internal web search tool_use block, and injects the
+    native block pair on the way back out.
+    """
+
+    def __init__(self):
+        self.plan_calls = 0
+        self.post_hook_calls = 0
+
+    async def async_should_run_agentic_loop(
+        self, response, model, messages, tools, stream, custom_llm_provider, kwargs
+    ):
+        if not isinstance(response, dict):
+            return True, {"tool_calls": [_internal_tool_use_block()]}
+        tool_calls = [
+            block
+            for block in response.get("content", [])
+            if block.get("type") == "tool_use" and block.get("name") == INTERNAL_TOOL_NAME
+        ]
+        if not tool_calls:
+            return False, {}
+        return True, {"tool_calls": tool_calls, "tool_type": "websearch"}
+
+    async def async_build_agentic_loop_plan(
+        self,
+        tools,
+        model,
+        messages,
+        response,
+        anthropic_messages_provider_config,
+        anthropic_messages_optional_request_params,
+        logging_obj,
+        stream,
+        kwargs,
+    ):
+        self.plan_calls += 1
+        return AgenticLoopPlan(
+            run_agentic_loop=True,
+            request_patch=AgenticLoopRequestPatch(
+                messages=[{"role": "user", "content": "here are the search results"}],
+                max_tokens=1024,
+            ),
+        )
+
+    async def async_post_agentic_loop_response_hook(self, response, plan, kwargs):
+        self.post_hook_calls += 1
+        if isinstance(response, dict):
+            response["content"] = [*_native_search_blocks(index=2), *response.get("content", [])]
+        return response
+
+
+def _logging_obj(callback: CustomLogger, converted_stream: bool = False) -> MagicMock:
+    logging_obj = MagicMock()
+    logging_obj.model_call_details = {"websearch_interception_converted_stream": converted_stream}
+    logging_obj.dynamic_success_callbacks = [callback]
+    logging_obj.litellm_call_id = "call-abc"
+    return logging_obj
+
+
+async def _run_hooks(
+    handler: BaseLLMHTTPHandler,
+    callback: CustomLogger,
+    kwargs: dict,
+    response: object = None,
+    stream: bool = False,
+    converted_stream: bool = False,
+    api_surface: str = "anthropic_messages",
+):
+    return await handler._call_agentic_completion_hooks(
+        response=_response_asking_for_another_search() if response is None else response,
+        model="claude-sonnet-4-5",
+        messages=[{"role": "user", "content": "who won the world cup"}],
+        anthropic_messages_provider_config=MagicMock(),
+        anthropic_messages_optional_request_params={},
+        logging_obj=_logging_obj(callback, converted_stream=converted_stream),
+        stream=stream,
+        custom_llm_provider="anthropic",
+        kwargs=kwargs,
+        api_surface=api_surface,
+    )
+
+
+class TestCappedLoopReturnsTerminalResponse:
+    def setup_method(self):
+        self.handler = BaseLLMHTTPHandler()
+        self.callback = _InterceptingCallback()
+
+    @pytest.mark.asyncio
+    async def test_internal_tool_use_block_is_dropped(self):
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+        )
+
+        assert isinstance(result, dict)
+        assert INTERNAL_TOOL_NAME not in _tool_use_names(result)
+
+    @pytest.mark.asyncio
+    async def test_stop_reason_is_closed_out(self):
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+        )
+
+        assert result["stop_reason"] == "end_turn"
+
+    @pytest.mark.asyncio
+    async def test_native_blocks_and_text_survive(self):
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+        )
+
+        assert _block_types(result) == ["server_tool_use", "web_search_tool_result", "text"]
+
+    @pytest.mark.asyncio
+    async def test_turn_carrying_only_the_refused_call_still_ends_cleanly(self):
+        """
+        The refused call can be every block the model produced, which leaves the
+        turn with no content once it is dropped. That still has to come back as a
+        finished turn rather than as the leaked call, so the client stops instead
+        of waiting on a tool it cannot run, and the rest of the message survives
+        so the request is still billed and traceable.
+
+        An empty turn renders as nothing, which is the ceiling being set too low
+        for the question rather than a malformed response.
+        """
+        nothing_but_the_refused_call = {
+            "id": "msg_123",
+            "type": "message",
+            "role": "assistant",
+            "model": "claude-sonnet-4-5",
+            "content": [_internal_tool_use_block()],
+            "stop_reason": "tool_use",
+            "usage": {"input_tokens": 10, "output_tokens": 5},
+        }
+
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+            response=nothing_but_the_refused_call,
+        )
+
+        assert result["content"] == []
+        assert result["stop_reason"] == "end_turn"
+        assert result["usage"] == {"input_tokens": 10, "output_tokens": 5}
+        assert result["id"] == "msg_123"
+
+    @pytest.mark.asyncio
+    async def test_no_follow_up_model_call_is_planned(self):
+        """
+        The rail has to end the turn without planning another model call, and it
+        has to end it by returning rather than by raising, which is the half that
+        the caller's response depends on.
+        """
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+        )
+
+        assert self.callback.plan_calls == 0
+        assert result["stop_reason"] == "end_turn"
+
+    @pytest.mark.asyncio
+    async def test_original_response_is_not_mutated(self):
+        response = _response_asking_for_another_search()
+
+        await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+            response=response,
+        )
+
+        assert response["stop_reason"] == "tool_use"
+        assert INTERNAL_TOOL_NAME in _tool_use_names(response)
+
+    @pytest.mark.asyncio
+    async def test_repeated_fingerprint_guard_is_terminal_too(self):
+        tool_calls = {"tool_calls": [_internal_tool_use_block()], "tool_type": "websearch"}
+        seen = json.dumps(tool_calls, sort_keys=True, default=str)
+
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 0, "max_agentic_loops": 3, "_agentic_loop_fingerprints": [seen]},
+        )
+
+        assert self.callback.plan_calls == 0
+        assert INTERNAL_TOOL_NAME not in _tool_use_names(result)
+        assert result["stop_reason"] == "end_turn"
+
+    @pytest.mark.asyncio
+    async def test_client_declared_tool_use_is_left_alone(self):
+        response = _response_asking_for_another_search()
+        client_tool_use = {"id": "toolu_client_1", "type": "tool_use", "name": "get_weather", "input": {}}
+        response["content"].append(client_tool_use)
+
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+            response=response,
+        )
+
+        assert _tool_use_names(result) == ["get_weather"]
+        assert result["stop_reason"] == "tool_use"
+
+    def test_only_the_refused_tool_calls_are_dropped(self):
+        """
+        A block is matched on the id the rail refused, not on the tool name, so a
+        second block sharing that name survives when the rail never listed it. A
+        callback that picks its tool calls out by name hands both over and both
+        go, which is its own call to make; this is about not widening it here.
+        """
+        response = _response_asking_for_another_search()
+        response["content"].append(
+            {"id": "toolu_client_1", "type": "tool_use", "name": INTERNAL_TOOL_NAME, "input": {}}
+        )
+
+        result = BaseLLMHTTPHandler._finalize_refused_agentic_response(
+            response=response,
+            tool_calls={"tool_calls": [_internal_tool_use_block()]},
+        )
+
+        assert [block["id"] for block in result["content"] if block.get("type") == "tool_use"] == ["toolu_client_1"]
+        assert result["stop_reason"] == "tool_use"
+
+    def test_tool_calls_without_ids_still_match_by_name(self):
+        """
+        Not every callback shape carries ids on its tool calls, so the name is
+        still what decides when the rail refused a call that has no id.
+        """
+        result = BaseLLMHTTPHandler._finalize_refused_agentic_response(
+            response=_response_asking_for_another_search(),
+            tool_calls={"tool_calls": [{"name": INTERNAL_TOOL_NAME, "input": {}}]},
+        )
+
+        assert _tool_use_names(result) == []
+        assert result["stop_reason"] == "end_turn"
+
+    @pytest.mark.asyncio
+    async def test_streaming_caller_is_left_to_its_existing_behavior(self):
+        """
+        A streaming caller has already sent the original message to the client, so
+        a finalized turn would land as a second message rather than replace the
+        first. The rail keeps raising there and the caller handles it as before.
+        """
+        with pytest.raises(AgenticLoopSafetyError):
+            await _run_hooks(
+                self.handler,
+                self.callback,
+                kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+                stream=True,
+            )
+
+        assert self.callback.plan_calls == 0
+
+    @pytest.mark.asyncio
+    async def test_responses_surface_is_left_to_its_existing_behavior(self):
+        """
+        The responses surface carries a pydantic model rather than the anthropic
+        dict this finalizer rewrites, so it keeps raising instead of being handed
+        a response that was never actually finalized.
+        """
+        with pytest.raises(AgenticLoopSafetyError):
+            await _run_hooks(
+                self.handler,
+                self.callback,
+                kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+                api_surface="responses",
+            )
+
+    @pytest.mark.asyncio
+    async def test_non_dict_response_is_returned_untouched(self):
+        response = MagicMock()
+
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+            response=response,
+        )
+
+        assert result is response
+
+    @pytest.mark.asyncio
+    async def test_converted_stream_gets_a_terminal_fake_stream(self):
+        """
+        A converted stream is wrapped back into an Anthropic SSE stream here, the
+        same as every other return in this function, so a streaming client gets a
+        terminal stream rather than a bare dict. The interceptor turns the client's
+        stream into a non-streaming upstream call, so stream is False on this path
+        and the converted flag on the logging object is what marks it.
+        """
+        result = await _run_hooks(
+            self.handler,
+            self.callback,
+            kwargs={"_agentic_loop_depth": 3, "max_agentic_loops": 3},
+            converted_stream=True,
+        )
+
+        assert isinstance(result, FakeAnthropicMessagesStreamIterator)
+        assert result.response["stop_reason"] == "end_turn"
+        assert INTERNAL_TOOL_NAME not in _tool_use_names(result.response)
+
+    def test_rails_cannot_trip_in_the_outermost_frame(self):
+        """
+        Backs the invariant the test above relies on: at depth 0 the fingerprint set
+        is empty and the ceiling is at least 1, so neither rail can refuse.
+        """
+        depth, max_loops, fingerprints = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={})
+
+        assert depth == 0
+        assert fingerprints == []
+        assert max_loops >= 1
+
+        depth, max_loops, fingerprints = BaseLLMHTTPHandler._get_agentic_loop_settings(
+            kwargs={"max_agentic_loops": 1}
+        )
+
+        assert max_loops == 1
+        assert BaseLLMHTTPHandler._check_agentic_loop_safety(
+            tool_calls={"tool_calls": [_internal_tool_use_block()]},
+            fingerprints=fingerprints,
+            depth=depth,
+            max_loops=max_loops,
+            model="claude-sonnet-4-5",
+        )
+
+    def test_safety_error_is_still_a_value_error(self):
+        assert issubclass(AgenticLoopSafetyError, ValueError)
+
+    def test_safety_error_type_names_the_rail(self):
+        with pytest.raises(AgenticLoopSafetyError, match="max_agentic_loops"):
+            BaseLLMHTTPHandler._check_agentic_loop_safety(
+                tool_calls={"tool_calls": [_internal_tool_use_block()]},
+                fingerprints=[],
+                depth=3,
+                max_loops=3,
+                model="claude-sonnet-4-5",
+            )
+
+
+class TestOuterFramePostHookStillRuns:
+    """
+    The cap used to raise through the parent frame's await, which skipped the
+    parent's post-loop hook. The parent now gets its terminal response back and
+    finishes normally, so the blocks it was going to inject still land.
+    """
+
+    @pytest.mark.asyncio
+    async def test_parent_frame_injects_its_blocks_after_the_cap_trips(self, monkeypatch):
+        handler = BaseLLMHTTPHandler()
+        callback = _InterceptingCallback()
+
+        async def fake_acreate(**call_kwargs):
+            return await handler._call_agentic_completion_hooks(
+                response=_response_asking_for_another_search(block_id="toolu_internal_2"),
+                model=call_kwargs["model"],
+                messages=call_kwargs["messages"],
+                anthropic_messages_provider_config=MagicMock(),
+                anthropic_messages_optional_request_params={},
+                logging_obj=_logging_obj(callback),
+                stream=False,
+                custom_llm_provider="anthropic",
+                kwargs={
+                    key: call_kwargs[key]
+                    for key in ("_agentic_loop_depth", "max_agentic_loops", "_agentic_loop_fingerprints")
+                    if key in call_kwargs
+                },
+            )
+
+        monkeypatch.setattr("litellm.anthropic_interface.messages.acreate", fake_acreate)
+
+        result = await _run_hooks(
+            handler,
+            callback,
+            kwargs={"_agentic_loop_depth": 0, "max_agentic_loops": 1},
+        )
+
+        assert callback.plan_calls == 1
+        assert callback.post_hook_calls == 1
+        assert _block_types(result)[:2] == ["server_tool_use", "web_search_tool_result"]
+        assert INTERNAL_TOOL_NAME not in _tool_use_names(result)
+        assert result["stop_reason"] == "end_turn"
+
+
+class TestMaxAgenticLoopsConfigKnob:
+    def test_from_config_yaml_reads_the_knob(self):
+        logger = WebSearchInterceptionLogger.from_config_yaml(
+            {"enabled_providers": ["bedrock"], "max_agentic_loops": 7}
+        )
+
+        assert logger.max_agentic_loops == 7
+
+    def test_from_config_yaml_leaves_it_unset_by_default(self):
+        logger = WebSearchInterceptionLogger.from_config_yaml({"enabled_providers": ["bedrock"]})
+
+        assert logger.max_agentic_loops is None
+
+    @pytest.mark.parametrize("bad_value", [0, -1])
+    def test_out_of_range_ceilings_are_rejected_at_config_load(self, bad_value):
+        with pytest.raises(ValueError, match="max_agentic_loops"):
+            WebSearchInterceptionLogger.from_config_yaml(
+                {"enabled_providers": ["bedrock"], "max_agentic_loops": bad_value}
+            )
+
+    @pytest.mark.parametrize("bad_value", ["three", True, 2.5])
+    def test_non_integer_ceilings_are_rejected_at_config_load(self, bad_value):
+        with pytest.raises(TypeError, match="max_agentic_loops"):
+            WebSearchInterceptionLogger.from_config_yaml(
+                {"enabled_providers": ["bedrock"], "max_agentic_loops": bad_value}
+            )
+
+    def test_a_ceiling_spelled_as_a_string_is_read_at_config_load(self):
+        """
+        `max_agentic_loops: os.environ/MAX_AGENTIC_LOOPS` resolves to a string
+        before it reaches the knob, so refusing "5" would break a config that
+        works today.
+        """
+        logger = WebSearchInterceptionLogger.from_config_yaml(
+            {"enabled_providers": ["bedrock"], "max_agentic_loops": "5"}
+        )
+
+        assert logger.max_agentic_loops == 5
+
+    @pytest.mark.asyncio
+    async def test_knob_reaches_the_loop_settings(self):
+        logger = WebSearchInterceptionLogger.from_config_yaml(
+            {"enabled_providers": ["bedrock"], "max_agentic_loops": 7}
+        )
+        kwargs = {
+            "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}],
+            "litellm_params": {"custom_llm_provider": "bedrock"},
+        }
+
+        updated = await logger.async_pre_request_hook(model="claude-sonnet-4-5", messages=[], kwargs=kwargs)
+
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs=updated)
+        assert max_loops == 7
+
+    @pytest.mark.asyncio
+    async def test_deployment_setting_wins_over_the_feature_setting(self):
+        logger = WebSearchInterceptionLogger.from_config_yaml(
+            {"enabled_providers": ["bedrock"], "max_agentic_loops": 7}
+        )
+        kwargs = {
+            "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}],
+            "litellm_params": {"custom_llm_provider": "bedrock"},
+            "max_agentic_loops": 2,
+        }
+
+        updated = await logger.async_pre_request_hook(model="claude-sonnet-4-5", messages=[], kwargs=kwargs)
+
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs=updated)
+        assert max_loops == 2
+
+    @pytest.mark.asyncio
+    async def test_default_ceiling_applies_when_the_knob_is_unset(self):
+        logger = WebSearchInterceptionLogger.from_config_yaml({"enabled_providers": ["bedrock"]})
+        kwargs = {
+            "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 5}],
+            "litellm_params": {"custom_llm_provider": "bedrock"},
+        }
+
+        updated = await logger.async_pre_request_hook(model="claude-sonnet-4-5", messages=[], kwargs=kwargs)
+
+        assert "max_agentic_loops" not in updated
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs=updated)
+        assert max_loops == 3
+
+
+def _stream_events(response: dict) -> list[dict]:
+    events: list[dict] = []
+    for chunk in FakeAnthropicMessagesStreamIterator(response=response):
+        for line in chunk.decode().splitlines():
+            if line.startswith("data: "):
+                events.append(json.loads(line[len("data: ") :]))
+    return events
+
+
+class TestBothCeilingKnobsAreValidated:
+    """
+    ``max_agentic_loops`` is settable per deployment and feature-wide, and the
+    per-deployment one wins. Only the feature-wide one used to be checked, so a
+    per-deployment ``0`` was swallowed by an ``or 3`` and read as the default 3,
+    handing the loosest ceiling to whoever asked for the tightest.
+    """
+
+    def test_a_per_deployment_zero_is_rejected_not_read_as_the_default(self):
+        with pytest.raises(ValueError, match="must be at least 1, got 0"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": 0})
+
+    def test_a_per_deployment_non_integer_names_the_field_it_came_from(self):
+        with pytest.raises(TypeError, match=r"litellm_params\.max_agentic_loops must be an integer"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": "three"})
+
+    def test_a_per_deployment_true_is_not_read_as_a_ceiling_of_one(self):
+        with pytest.raises(TypeError, match="must be an integer"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": True})
+
+    def test_an_absent_ceiling_falls_back_to_the_shared_default(self):
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={})
+
+        assert max_loops == DEFAULT_MAX_AGENTIC_LOOPS
+
+    def test_an_explicit_none_falls_back_to_the_shared_default(self):
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": None})
+
+        assert max_loops == DEFAULT_MAX_AGENTIC_LOOPS
+
+    def test_a_valid_per_deployment_ceiling_is_passed_through(self):
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": 6})
+
+        assert max_loops == 6
+
+    @pytest.mark.parametrize("rejected", [0, -1, "three", True])
+    def test_the_two_knobs_reject_the_same_values(self, rejected):
+        with pytest.raises((TypeError, ValueError)):
+            WebSearchInterceptionLogger(max_agentic_loops=rejected)
+        with pytest.raises((TypeError, ValueError)):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": rejected})
+
+    def test_each_knob_names_its_own_config_field(self):
+        with pytest.raises(ValueError, match=r"websearch_interception_params\.max_agentic_loops"):
+            WebSearchInterceptionLogger(max_agentic_loops=0)
+        with pytest.raises(ValueError, match=r"litellm_params\.max_agentic_loops"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": 0})
+
+
+class TestACeilingThatSpellsAWholeNumberStillWorks:
+    """
+    The ceiling used to go through ``int(... or 3)``, which accepted anything
+    ``int()`` accepted. A ceiling is routinely parameterized as
+    ``max_agentic_loops: os.environ/MAX_AGENTIC_LOOPS``, and ``get_secret``
+    hands that back as the string ``"5"``, so tightening the check to
+    ``isinstance(int)`` would stop such a proxy from booting on upgrade.
+    """
+
+    @pytest.mark.parametrize("spelled", ["5", " 5 ", 5.0])
+    def test_a_ceiling_that_spells_five_is_accepted_by_both_knobs(self, spelled):
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": spelled})
+
+        assert max_loops == 5
+        assert WebSearchInterceptionLogger(max_agentic_loops=spelled).max_agentic_loops == 5
+
+    def test_an_env_var_sourced_ceiling_survives_secret_resolution(self, monkeypatch):
+        monkeypatch.setenv("MAX_AGENTIC_LOOPS_UNDER_TEST", "7")
+        resolved = get_secret("os.environ/MAX_AGENTIC_LOOPS_UNDER_TEST")
+
+        assert isinstance(resolved, str)
+        _, max_loops, _ = BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": resolved})
+        assert max_loops == 7
+
+    def test_a_spelled_zero_is_still_refused_and_reports_the_number(self):
+        with pytest.raises(ValueError, match="must be at least 1, got 0"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": "0"})
+
+    def test_a_word_is_still_refused(self):
+        with pytest.raises(TypeError, match=r"litellm_params\.max_agentic_loops must be an integer"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": "three"})
+
+    def test_a_fractional_ceiling_is_refused_rather_than_truncated(self):
+        with pytest.raises(TypeError, match="must be an integer"):
+            BaseLLMHTTPHandler._get_agentic_loop_settings(kwargs={"max_agentic_loops": 5.5})
+
+
+class TestRebuiltStreamIsWellFormed:
+    """
+    A capped turn is rebuilt into SSE by FakeAnthropicMessagesStreamIterator.
+
+    Anthropic's SDK accumulator appends on content_block_start and then indexes
+    content[event.index] on content_block_delta, so a block that stops without
+    ever starting shifts every later index and the accumulator raises
+    IndexError. A web search turn carries server_tool_use and
+    web_search_tool_result blocks, which is exactly where that used to happen.
+    """
+
+    @staticmethod
+    def _capped_search_turn() -> dict:
+        return {
+            "id": "msg_01",
+            "type": "message",
+            "role": "assistant",
+            "model": "claude-sonnet-4-5",
+            "stop_reason": "end_turn",
+            "content": [
+                {
+                    "type": "server_tool_use",
+                    "id": "srvtoolu_01",
+                    "name": "web_search",
+                    "input": {"query": "on-demand H100 hourly price"},
+                },
+                {
+                    "type": "web_search_tool_result",
+                    "tool_use_id": "srvtoolu_01",
+                    "content": [
+                        {
+                            "type": "web_search_result",
+                            "url": "https://example.com/h100",
+                            "title": "H100 pricing",
+                        }
+                    ],
+                },
+                {"type": "text", "text": "AWS lists the H100 at $12.29 an hour."},
+            ],
+            "usage": {"input_tokens": 100, "output_tokens": 20},
+        }
+
+    def test_every_content_block_stop_has_a_matching_start(self):
+        events = _stream_events(self._capped_search_turn())
+
+        started = [event["index"] for event in events if event["type"] == "content_block_start"]
+        stopped = [event["index"] for event in events if event["type"] == "content_block_stop"]
+
+        assert started == [0, 1, 2]
+        assert stopped == [0, 1, 2]
+
+    def test_no_delta_indexes_past_the_blocks_started_before_it(self):
+        events = _stream_events(self._capped_search_turn())
+
+        blocks_started = 0
+        for event in events:
+            if event["type"] == "content_block_start":
+                blocks_started += 1
+            elif event["type"] == "content_block_delta":
+                assert event["index"] < blocks_started
+
+    def test_search_blocks_reach_the_client(self):
+        events = _stream_events(self._capped_search_turn())
+
+        started_types = [
+            event["content_block"]["type"] for event in events if event["type"] == "content_block_start"
+        ]
+
+        assert started_types == ["server_tool_use", "web_search_tool_result", "text"]
+
+    def test_the_search_result_survives_the_rebuild_intact(self):
+        events = _stream_events(self._capped_search_turn())
+
+        result_block = next(
+            event["content_block"]
+            for event in events
+            if event["type"] == "content_block_start"
+            and event["content_block"]["type"] == "web_search_tool_result"
+        )
+
+        assert result_block["tool_use_id"] == "srvtoolu_01"
+        assert result_block["content"][0]["url"] == "https://example.com/h100"
diff --git a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
index b0b2c68e30d..ee0de8840f6 100644
--- a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
@@ -26,6 +26,7 @@ from litellm.proxy.proxy_server import (
     _scrub_guardrail_inner,
     resolve_complexity_router_plugins,
     resolve_routing_plugins,
+    validate_deployment_max_agentic_loops,
 )
 
 from .conftest import normalize
@@ -153,6 +154,71 @@ def test_resolve_complexity_router_plugins_resolves_dotted_path_to_live_instance
     assert type(config["plugins"][0]).__name__ == "_Plugin"
 
 
+def test_validate_deployment_max_agentic_loops_allows_a_deployment_without_the_key():
+    model = {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o"}}
+
+    validate_deployment_max_agentic_loops(model)
+
+    assert "max_agentic_loops" not in model["litellm_params"]
+
+
+def test_validate_deployment_max_agentic_loops_leaves_a_valid_ceiling_alone():
+    model = {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o", "max_agentic_loops": 5}}
+
+    validate_deployment_max_agentic_loops(model)
+
+    assert model["litellm_params"]["max_agentic_loops"] == 5
+
+
+def test_validate_deployment_max_agentic_loops_rejects_zero():
+    """
+    A per-deployment 0 used to be swallowed by an `or 3` and read as the default
+    ceiling of 3, handing the loosest setting to whoever asked for the tightest.
+    """
+    with pytest.raises(ValueError, match="must be at least 1, got 0"):
+        validate_deployment_max_agentic_loops(
+            {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o", "max_agentic_loops": 0}}
+        )
+
+
+def test_validate_deployment_max_agentic_loops_rejects_a_non_integer():
+    """
+    A per-deployment non-integer used to let the proxy boot and then fail every
+    request to that model with `invalid literal for int() with base 10`.
+    """
+    with pytest.raises(TypeError, match="must be an integer"):
+        validate_deployment_max_agentic_loops(
+            {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o", "max_agentic_loops": "three"}}
+        )
+
+
+def test_validate_deployment_max_agentic_loops_rejects_a_bool():
+    with pytest.raises(TypeError, match="must be an integer"):
+        validate_deployment_max_agentic_loops(
+            {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o", "max_agentic_loops": True}}
+        )
+
+
+def test_validate_deployment_max_agentic_loops_accepts_a_ceiling_from_an_env_var():
+    """
+    `max_agentic_loops: os.environ/MAX_AGENTIC_LOOPS` is resolved to a string
+    before this check runs, and the old `int(... or 3)` accepted that, so
+    refusing it here would stop an already working proxy from booting.
+    """
+    model = {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o", "max_agentic_loops": "5"}}
+
+    validate_deployment_max_agentic_loops(model)
+
+    assert model["litellm_params"]["max_agentic_loops"] == "5"
+
+
+def test_validate_deployment_max_agentic_loops_names_the_offending_model():
+    with pytest.raises(ValueError, match="on model 'claude-sonnet-4-5'"):
+        validate_deployment_max_agentic_loops(
+            {"model_name": "claude-sonnet-4-5", "litellm_params": {"max_agentic_loops": -1}}
+        )
+
+
 def test_resolve_complexity_router_plugins_rejects_non_routing_plugin_object(tmp_path):
     plugin_file = tmp_path / "bad_plugin.py"
     plugin_file.write_text("not_a_plugin = object()\n")