Tool-path rate limits: retry at the bridge, then fail the run fast (#492)

* Tool-path rate limits: retry at the bridge, then fail the run fast

A PageIndex cloud 429 (or 5xx) on a tool call used to reach the model as
an INTERNAL_ERROR envelope saying "try again": the model re-called once
with no wait, then wrote the failure into its answer, and chat() returned
normally with no status anywhere. The same 429 before the loop (the
doc_id targeting lookup) already propagated raw.

- McpBridge mounts a urllib3 Retry: 429/502/503 and connection failures,
  three attempts, 0/2/4 s apart or as Retry-After says; read timeouts
  are never replayed (240 s each, and the server may have acted); a
  Retry-After past a minute is a quota, not a blip, so the backoff runs
  instead of sleeping it out. Exhausted, the last response falls through
  to the existing >= 400 branch, so the status_code survives.
- _bridge_invoker re-raises 429/5xx alongside 401/403. The frameworks
  turn a raised tool exception back into model-visible text, so each
  chat() door gets its own escape: the in-process MCPServer's
  failure_error_function lets a PageIndex-caused failure propagate and
  _translate_run_error unwraps it from the framework's wrapper (which
  also un-flattens the 401 case); the Messages lane runs each turn's
  tools through the runner's public generate_tool_call_response() and
  raises before the next model call.
- _model_backend_error keeps the provider's status_code.

Claude Agent SDK tools cannot fail fast: the SDK MCP server converts
handler exceptions into JSON-RPC errors for Claude Code by design.

Claude-Session: https://claude.ai/code/session_014S88dcSz7jykegAWyWZk8E

* Tool-path fail-fast: cover unreachable servers and all 5xx

The bridge retry is now a plain urllib3 Retry: 429 and every 5xx retried
three times at the fixed 0/2/4 s backoff, Retry-After ignored. That drops
the _Retry subclass, whose get_retry_after raised InvalidHeader on a
non-integer header (turning a 429 into "could not reach the server"),
honoured a 60 s Retry-After three times over, and let a 413 carrying
Retry-After replay. 500 and 504 join the forcelist so the invoker's "what
survived the bridge's retries" holds for every status it re-raises. Retry
is imported from requests.adapters, the declared dependency.

The invoker re-raises transport failures too: once the bridge's own
connection retries fail, the model cannot reach the server either, and
the envelope only sent it round the retry loop.

The handshake error blames the API key only on 401/403: a rate-limited
handshake is now a run-terminating error and was telling users to rotate
a working key.

Docstrings on agent_tools()/build_agent_tools and the Anthropic adapter
state the real raise set: 401/403, post-retry 429/5xx, unreachable server.

Claude-Session: https://claude.ai/code/session_013xk3xt9KgHNTjsYmFLKxbu

* fix: surface Messages tool failures before advancing runner

* fix: require urllib3 1.26 for MCP retries

* fix: keep the Messages fail-fast quiet and single-path

Raise ToolError from the tools chat(protocol="messages") runs instead of
the raw PageIndexAPIError: the Anthropic runner log.exception()s any
other exception, so every fail-fast printed a 20-line traceback from
anthropic's internals before the SDK raised its own error. The lane
still records the failure and raises it right after the runner's tool
batch, so the ToolError content never reaches the model.

Drop the two post-loop _messages_fail_fast calls: the runner executes
tools only through the public generate_tool_call_response (0.108.0
through 1.4.0), which checked_tool_response wraps, so they could never
fire. Annotate _pageindex_cause for the py.typed package.

Claude-Session: https://claude.ai/code/session_01CfYSeq8kM7HjfF79TGbsiT

* fix: fail fast on the cloud's account-limit tool errors; one 5xx list

RATE_LIMITED / USAGE_LIMIT_REACHED (pageindex-chat #472) arrive as a
normal tool error inside HTTP 200, already retried server-side; the
invoker re-raises them as 429 / 402, the way a post-retry status
escapes, so every lane fails fast without a per-lane change.

The bridge retries the whole 5xx range, the same range the invoker
re-raises; a non-JSON 200 body (a JSONDecodeError is a RequestException
too) stays a model-visible envelope, since the server was reached. The
MCP stub tests keep the machine's proxy out of 127.0.0.1.

Claude-Session: https://claude.ai/code/session_01F7vqmZdnWeKC9SUBytDrdf

* fix: raise the account-limit escape outside the invoker's own try

_raise_account_limit fired inside _invoke's try, so its escape depended
on 429 and 402 also appearing in the except's re-raise tuple: two lists
in one function that had to agree. The check now runs after the try,
where the except cannot swallow it, and 402 leaves the tuple (the MCP
route never answers HTTP 402; it was there only to let the raise through).

The mcp_stub fixture also sets the lowercase no_proxy: requests reads
that spelling first, so a machine with no_proxy set still routed the
stub requests through its proxy despite NO_PROXY.
This commit is contained in:
Ray
2026-09-10 19:34:05 +08:00
committed by GitHub
parent 3e0c2a3add
commit 5f7a39e175
11 changed files with 569 additions and 46 deletions
+43 -10
View File
@@ -14,8 +14,10 @@ same JSON envelope the cloud emits ({"success": true, ...} /
{"error": ...}) — arguments outside a pruned local signature come back as
that envelope too, on the direct and the call_tool path alike, except
browse_documents' ``recursive``: call_tool honors it, because the flat
no-folders shape it asks for is trivially true here. One exception to
never-raise: a cloud 401/403 re-raises PageIndexAPIError.
no-folders shape it asks for is trivially true here. The exceptions to
never-raise, all cloud: a 401/403, a 429/5xx that outlived the bridge's
retries, an unreachable server and a RATE_LIMITED / USAGE_LIMIT_REACHED
tool error re-raise PageIndexAPIError.
"""
from __future__ import annotations
@@ -29,6 +31,8 @@ import time
import weakref
from typing import Any, Callable, Optional
import requests
from .errors import PageIndexAPIError
from .mcp_bridge import render_text
@@ -1357,22 +1361,46 @@ def _annotation_for(spec: dict) -> Any:
return Optional[base] if nullable else base
_ACCOUNT_LIMITS = {"RATE_LIMITED": 429, "USAGE_LIMIT_REACHED": 402}
def _raise_account_limit(blocks: list) -> None:
"""The cloud's account-level tool errors (retried server-side already;
the model can act on neither) re-raise as the status they stand for."""
try:
payload = json.loads(blocks[0]["text"])
status = _ACCOUNT_LIMITS[payload["errorCode"]]
except (LookupError, TypeError, ValueError):
return
message = str(payload.get("error") or payload["errorCode"])
for key in ("retry_after_seconds", "open_url"):
if key in payload:
message += f" ({key}: {payload[key]})"
raise PageIndexAPIError(message, status_code=status)
def _bridge_invoker(bridge, name: str, schema: dict,
) -> "Callable[[dict], tuple[list, bool]]":
"""One cloud tool call proxied over MCP: string booleans are coerced
(same as call_tool), None-valued arguments are dropped (None ≡ omitted,
matching the contract's "omit if ..." semantics) and failures are
contained in the error envelope — except 401/403, which re-raise.
Returns (content blocks, is_error), like the bridge."""
contained in the error envelope — except auth failures, what
survived the bridge's own retries (401/403/429/5xx), an unreachable
server and the cloud's RATE_LIMITED / USAGE_LIMIT_REACHED tool errors
(429 / 402), which re-raise: the model can act on none of them. Returns
(content blocks, is_error), like the bridge."""
def _invoke(arguments: dict[str, Any]) -> tuple[list, bool]:
try:
arguments = {key: value for key, value in arguments.items()
if value is not None}
_coerce_bool_args(schema, arguments)
return bridge.call_tool(name, arguments)
blocks, is_error = bridge.call_tool(name, arguments)
except Exception as exc:
if (isinstance(exc, PageIndexAPIError)
and exc.status_code in (401, 403)):
if isinstance(exc, PageIndexAPIError) and (
exc.status_code in (401, 403, 429)
or (exc.status_code or 0) >= 500
or (exc.status_code is None and isinstance(
exc.__cause__, requests.RequestException))):
raise
payload, _ = _failure(
f"{name} failed: {exc}", None,
@@ -1383,6 +1411,9 @@ def _bridge_invoker(bridge, name: str, schema: dict,
"INTERNAL_ERROR",
)
return [{"type": "text", "text": _dumps(payload)}], True
if is_error:
_raise_account_limit(blocks)
return blocks, is_error
return _invoke
@@ -1432,8 +1463,8 @@ def _make_tool_function(name: str, description: str, schema: dict,
inner.__annotations__ = annotations
def proxy(*args: Any, **kwargs: Any) -> str:
# The invoker lets only 401/403 auth failures through, so a
# TypeError here is the binding rejecting the arguments.
# The invoker lets only PageIndexAPIError through, so a TypeError
# here is the binding rejecting the arguments.
try:
return inner(*args, **kwargs)
except TypeError as exc:
@@ -1547,7 +1578,9 @@ def build_agent_tools(client, include_management: bool = False,
the built-in contract tools over the local store. Every function returns
the JSON envelope as a string (binary content, such as a cloud page
image, as a size stub) and never raises for arguments its
signature accepts — except a cloud 401/403, which re-raises
signature accepts — except a cloud 401/403, a 429/5xx that outlived
the bridge's retries, an unreachable server or a RATE_LIMITED /
USAGE_LIMIT_REACHED tool error, which re-raise
PageIndexAPIError (cloud-only parameters are absent from the local
signatures; the call_tool path answers them with the guided envelope).
``doc_ids`` is the local allowlist, as in ``_tool_specs``.
+13 -3
View File
@@ -1549,7 +1549,9 @@ class PageIndexClient:
string (binary results such as ``get_document_image`` as a size
stub — a string cannot carry an image; the framework adapters
can), and reports failures inside that JSON instead of raising —
except a cloud 401/403, which raises PageIndexAPIError.
except a cloud 401/403, a 429/5xx that outlived the bridge's
retries, an unreachable server or a RATE_LIMITED /
USAGE_LIMIT_REACHED tool error, which raise PageIndexAPIError.
Args:
include_management (bool): Also expose tools that modify the
@@ -1570,7 +1572,11 @@ class PageIndexClient:
"""
Tools for the OpenAI Agents SDK — pass to ``Agent(tools=...)``
(or ``openai_agent_config()`` for all the Agent slots in one
call).
call). In-process cloud tools abort the run on a 401/403, a
429/5xx that outlived the bridge's retries, an unreachable server
or a RATE_LIMITED / USAGE_LIMIT_REACHED tool error: the
framework's AgentsException, the PageIndexAPIError as its
``__cause__``.
Cloud (default): the full live read tool set (search, folders,
images — as enabled for your key) as function tools, discovered
@@ -1712,7 +1718,11 @@ class PageIndexClient:
The default flavor is for the sync ``Anthropic`` client; pass
``asynchronous=True`` for ``AsyncAnthropic``. For a manual
``messages.create`` loop, serialize with
``[tool.to_dict() for tool in ...]``.
``[tool.to_dict() for tool in ...]``. A cloud 401/403, a 429/5xx
that outlived the bridge's retries, an unreachable server or a
RATE_LIMITED / USAGE_LIMIT_REACHED tool error raises
PageIndexAPIError, which the tool runner flattens into an
is_error result.
Cloud: the full live read tool set (search, folders, images — as
enabled for your key), discovered from the PageIndex MCP server
+9
View File
@@ -5,3 +5,12 @@ class PageIndexAPIError(Exception):
def __init__(self, *args: object, status_code: int | None = None) -> None:
super().__init__(*args)
self.status_code = status_code
def _pageindex_cause(exc: BaseException | None) -> PageIndexAPIError | None:
"""The PageIndexAPIError behind a framework's wrapper exception, if any."""
while exc is not None:
if isinstance(exc, PageIndexAPIError):
return exc
exc = exc.__cause__
return None
+19 -4
View File
@@ -6,7 +6,11 @@ input_schema are the same shape), calls proxied over MCP. Local clients get
the in-process tools — the same set chat(protocol="messages") runs
internally. Failed
calls raise ToolError so the runner emits the tool_result with
``is_error: true`` and the envelope as its content.
``is_error: true`` and the envelope as its content; the failures the
invoker re-raises (auth, limits, unreachable server) propagate as
PageIndexAPIError, which a caller-owned runner flattens into an is_error
result carrying the exception text, or land in ``failures`` when one is
supplied, for chat(protocol="messages") to fail fast on between turns.
Tool results are MCP content, rendered by the Anthropic SDK's own MCP
conversion (text as text, images as image blocks); the SDK carries the
@@ -15,13 +19,16 @@ MCP types and renders nothing.
from __future__ import annotations
import asyncio
from typing import Any
from typing import Any, Optional
from ..errors import PageIndexAPIError
def build_anthropic_tools(client, include_management: bool = False,
asynchronous: bool = False, doc_ids=None) -> list:
asynchronous: bool = False, doc_ids=None,
failures: Optional[list] = None) -> list:
"""``failures`` records the invoker's re-raised failures for
chat(protocol="messages") to fail fast on."""
try:
from anthropic import beta_async_tool, beta_tool
from anthropic.lib.tools import ToolError
@@ -41,7 +48,15 @@ def build_anthropic_tools(client, include_management: bool = False,
moves the blocking bridge/store call into a worker thread so it
never blocks the caller's event loop."""
def run(kwargs: dict) -> list:
blocks, is_error = invoke(kwargs)
try:
blocks, is_error = invoke(kwargs)
except PageIndexAPIError as exc:
if failures is None:
raise
failures.append(exc)
# the runner logs a traceback for anything but ToolError;
# the lane raises exc itself before the runner advances
raise ToolError(str(exc)) from exc
result = CallToolResult.model_validate(
{"content": blocks, "isError": is_error})
content = [mcp_content(block) for block in result.content]
+13 -2
View File
@@ -16,7 +16,18 @@ from __future__ import annotations
import asyncio
from ..errors import PageIndexAPIError
from ..errors import PageIndexAPIError, _pageindex_cause
def _tool_failure(ctx, error):
"""The framework's tool-failure formatter, narrowed: a PageIndex failure
the invoker re-raised (auth, limits, post-retry transport) escapes the run
instead of becoming model-visible text; anything else keeps the
framework default."""
from agents.tool import default_tool_error_function
if _pageindex_cause(error) is not None:
raise error
return default_tool_error_function(ctx, error)
def build_mcp_server(client, include_management: bool = False, doc_ids=None):
@@ -29,7 +40,7 @@ def build_mcp_server(client, include_management: bool = False, doc_ids=None):
class _ToolServer(MCPServer):
def __init__(self):
super().__init__()
super().__init__(failure_error_function=_tool_failure)
self.tools = [mcp_types.Tool(name=name, description=description,
inputSchema=schema)
for name, description, schema, _ in specs]
+21 -4
View File
@@ -15,7 +15,7 @@ from typing import Any, Iterator, Mapping, Optional, Union
from .agent_tools import _base_instructions, doc_targeting_block
from .chat_stream import ChatStream
from .errors import PageIndexAPIError
from .errors import PageIndexAPIError, _pageindex_cause
CHAT_HEADER = (
"You are PageIndex by Vectify AI, a document-focused assistant. "
@@ -451,7 +451,8 @@ def _model_backend_error(exc, lane: str, client=None) -> PageIndexAPIError:
", or drop the chat model configuration to use the managed "
"cloud chat." if lane == "chat" else "."
)
return PageIndexAPIError(message)
return PageIndexAPIError(message,
status_code=getattr(exc, "status_code", None))
def _translate_run_error(exc, max_turns, lane, client=None) -> PageIndexAPIError:
@@ -460,6 +461,10 @@ def _translate_run_error(exc, max_turns, lane, client=None) -> PageIndexAPIError
if isinstance(exc, MaxTurnsExceeded):
return _wrap_max_turns(max_turns)
if isinstance(exc, AgentsException):
cause = _pageindex_cause(exc)
if cause is not None:
# a tool failure the invoker re-raised, wrapped on its way out
return PageIndexAPIError(str(cause), status_code=cause.status_code)
return PageIndexAPIError(f"The agent backend failed: {exc}")
return _model_backend_error(exc, lane, client)
@@ -1378,7 +1383,8 @@ def run_messages(client, messages, model: str,
if _cache_marks(system_blocks, prepared) < 4 else {})
# Tools before the transport: on a bridge client building them is
# network I/O, and a failure there must not strand the client below.
tools = build_anthropic_tools(client, doc_ids=scope)
failures: list = []
tools = build_anthropic_tools(client, doc_ids=scope, failures=failures)
merged = _merged_backend(client, backend)
backend_client = _anthropic_client(merged)
# Close only a per-call construction: cached clients stay open for
@@ -1400,6 +1406,17 @@ def run_messages(client, messages, model: str,
**passthrough,
**cached,
)
# Older Anthropic versions also execute tools on max_tokens turns, newer
# ones skip them: check right after the runner's own tool step.
generate_tool_response = runner.generate_tool_call_response
def checked_tool_response():
response = generate_tool_response()
if failures:
raise failures[0]
return response
runner.generate_tool_call_response = checked_tool_response
if stream:
def events() -> Iterator[Any]:
@@ -1424,7 +1441,7 @@ def run_messages(client, messages, model: str,
return events()
try:
turns = [turn for turn in runner]
turns = list(runner)
except anthropic.AnthropicError as exc:
raise _model_backend_error(exc, "messages", client) from exc
except TypeError as exc:
+13 -2
View File
@@ -16,12 +16,20 @@ import threading
from typing import Any, Optional
import requests
from requests.adapters import HTTPAdapter, Retry
from ._version import sdk_version
from .errors import PageIndexAPIError
_PROTOCOL_VERSION = "2025-06-18"
_TIMEOUT = (10, 240) # tools may wait server-side (wait_for_completion: 3 min)
# Below the tool layer, so the model never plays retry loop. read=0: a read
# timeout is a full wait the server may have acted on, never replayed.
# Retry-After is ignored: a long one is a quota, not a blip.
_RETRY = Retry(total=3, read=0, backoff_factor=1,
status_forcelist=(429, *range(500, 600)),
allowed_methods=None, raise_on_status=False,
respect_retry_after_header=False)
def _parse_sse(text: str) -> list[dict]:
@@ -45,6 +53,8 @@ class McpBridge:
self._url = url
self._auth_headers = dict(headers)
self._session = requests.Session() # agent tool calls come in bursts
for scheme in ("https://", "http://"):
self._session.mount(scheme, HTTPAdapter(max_retries=_RETRY))
self._session_id: Optional[str] = None
self._protocol_version: Optional[str] = None
self._instructions: Optional[str] = None
@@ -154,10 +164,11 @@ class McpBridge:
},
})
if response.status_code >= 400:
hint = (" Check your API key."
if response.status_code in (401, 403) else "")
raise PageIndexAPIError(
f"Could not connect to the PageIndex MCP server: HTTP "
f"{response.status_code} ({response.text[:200]}). Check "
"your API key.",
f"{response.status_code} ({response.text[:200]}).{hint}",
status_code=response.status_code,
)
result = self._extract_result(response, request_id) or {}
+2
View File
@@ -27,6 +27,8 @@ exclude = ["pageindex/flash/assets"]
[tool.poetry.dependencies]
python = ">=3.10"
requests = ">=2.28.0"
# MCP retries use Retry(allowed_methods=...), added in urllib3 1.26.
urllib3 = ">=1.26"
openai = ">=1.70.0"
# Older releases crash on current openai before the request is sent.
openai-agents = ">=0.18.1"
+2
View File
@@ -1,6 +1,8 @@
litellm==1.97.0
openai>=1.70.0
requests>=2.28.0
# MCP retries use Retry(allowed_methods=...), added in urllib3 1.26.
urllib3>=1.26
openai-agents>=0.18.1
mcp>=1.19.0,<3
# pymupdf # optional
+216 -20
View File
@@ -1038,24 +1038,55 @@ def test_anthropic_runner_config_thinking_lifts_max_tokens(client):
model="claude-sonnet-4-5")
def test_bridge_invoker_reraises_auth_failures():
class Revoked:
def call_tool(self, name, arguments):
raise PageIndexAPIError("HTTP 401", status_code=401)
def test_bridge_invoker_reraises_auth_and_transport_failures():
"""Auth failures and what survives the bridge's own retries (429/5xx)
escape to the caller; a status-less JSON-RPC failure (the model's own
bad arguments) stays a model-visible envelope, and so does a non-JSON
200 body: its JSONDecodeError is a RequestException too, but the server
was reached."""
import requests
invoke = agent_tools_module._bridge_invoker(Revoked(), "get_document", {})
with pytest.raises(PageIndexAPIError, match="401"):
invoke({})
# Transport blips stay contained in the retryable envelope.
def failing(exc):
class Bridge:
def call_tool(self, name, arguments):
raise exc
return agent_tools_module._bridge_invoker(Bridge(), "get_document", {})
class Down:
def call_tool(self, name, arguments):
raise PageIndexAPIError("HTTP 503", status_code=503)
blocks, is_error = agent_tools_module._bridge_invoker(
Down(), "get_document", {})({})
for status in (401, 403, 429, 503):
with pytest.raises(PageIndexAPIError) as info:
failing(PageIndexAPIError(f"HTTP {status}", status_code=status))({})
assert info.value.status_code == status
blocks, is_error = failing(
PageIndexAPIError("MCP error -32602: bad params"))({})
assert is_error
assert json.loads(blocks[0]["text"])["errorCode"] == "INTERNAL_ERROR"
garbled = PageIndexAPIError("non-JSON response (HTTP 200).", status_code=200)
garbled.__cause__ = requests.exceptions.JSONDecodeError("bad", "<html>", 0)
blocks, is_error = failing(garbled)({})
assert is_error
assert json.loads(blocks[0]["text"])["errorCode"] == "INTERNAL_ERROR"
def test_bridge_invoker_reraises_account_limits():
"""The cloud answers upstream throttling and an exhausted quota as a
normal tool error (HTTP 200 + errorCode): the model can act on neither,
so they escape like a post-retry 429; every other code stays a
model-visible envelope."""
def answering(payload):
class Bridge:
def call_tool(self, name, arguments):
return [{"type": "text", "text": json.dumps(payload)}], True
return agent_tools_module._bridge_invoker(Bridge(), "get_document", {})
for code, status in (("RATE_LIMITED", 429), ("USAGE_LIMIT_REACHED", 402)):
with pytest.raises(PageIndexAPIError,
match=r"limit \(retry_after_seconds: 7\)") as info:
answering({"error": "limit", "errorCode": code,
"retry_after_seconds": 7})({})
assert info.value.status_code == status
blocks, is_error = answering({"error": "gone", "errorCode": "NOT_FOUND"})({})
assert is_error
assert json.loads(blocks[0]["text"])["errorCode"] == "NOT_FOUND"
def test_cloud_bridge_gates_the_endpoint(monkeypatch):
@@ -1507,7 +1538,7 @@ def test_mcp_bridge_protocol(monkeypatch):
# Replace the module's own `requests` binding — patching the shared
# requests module would leak the fake process-wide.
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=fake_post),
Session=lambda: types.SimpleNamespace(post=fake_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp",
{"Authorization": "Bearer k"})
@@ -1572,7 +1603,7 @@ def test_mcp_bridge_400_is_an_error_not_session_expiry(monkeypatch):
return _Resp(400, text="unknown tool")
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=fake_post),
Session=lambda: types.SimpleNamespace(post=fake_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp",
{"Authorization": "Bearer k"})
@@ -1634,7 +1665,7 @@ def test_mcp_bridge_init_notification_bars_concurrent_requests(monkeypatch):
return resp
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=fake_post),
Session=lambda: types.SimpleNamespace(post=fake_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp",
{"Authorization": "Bearer k"})
@@ -2023,7 +2054,7 @@ def test_bridge_call_tool_surfaces_iserror(monkeypatch):
"content": [{"type": "text", "text": '{"error": "denied"}'}]}})
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=fake_post),
Session=lambda: types.SimpleNamespace(post=fake_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp", {})
assert bridge.call_tool("t", {}) == (
@@ -2062,7 +2093,7 @@ def test_bridge_rejects_mismatched_reply_id(monkeypatch):
"text": "old"}]}})
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=fake_post),
Session=lambda: types.SimpleNamespace(post=fake_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp", {})
with pytest.raises(PageIndexAPIError, match="no reply matching"):
@@ -2087,13 +2118,26 @@ def test_bridge_transport_error_is_pageindex_error(monkeypatch):
raise requests_mod.ConnectionError("dns down")
monkeypatch.setattr(mcp_bridge, "requests", types.SimpleNamespace(
Session=lambda: types.SimpleNamespace(post=dead_post),
Session=lambda: types.SimpleNamespace(post=dead_post, mount=lambda *a: None),
RequestException=requests_mod.RequestException))
bridge = McpBridge("https://api.pageindex.ai/mcp", {})
with pytest.raises(PageIndexAPIError, match="Could not reach"):
bridge.list_tools()
def test_handshake_failure_blames_the_key_only_on_auth_statuses(monkeypatch):
"""A rate-limited or failing handshake is not a key problem."""
import types
from pageindex.mcp_bridge import McpBridge
bridge = McpBridge("https://api.pageindex.ai/mcp", {})
for status, blames_key in ((401, True), (429, False), (503, False)):
monkeypatch.setattr(bridge, "_post", lambda payload, *a, s=status: (
types.SimpleNamespace(status_code=s, text="no", headers={})))
with pytest.raises(PageIndexAPIError, match=f"HTTP {status}") as info:
bridge.list_tools()
assert ("Check your API key" in str(info.value)) is blames_key
def test_await_completion_preserves_metadata_over_null_refetch(monkeypatch):
"""A status refetch that nulls out metadata must not clobber the
listing's copy (setdefault is a no-op on an existing None value)."""
@@ -2784,3 +2828,155 @@ def test_cloud_tool_list_empty_raises(monkeypatch):
cloud = PageIndexCloudClient(api_key="pi-test-key")
with pytest.raises(PageIndexAPIError, match="no tools"):
cloud.as_openai_tools()
# ── tool-path rate limits: retried below the tool layer, then fail fast ──
class _McpStub:
"""A local MCP endpoint: initialize always succeeds; tools/call answers
follow the scripted statuses (a 200 carries a text result), the last
one repeating."""
def __init__(self, statuses, delay=0.0):
import http.server
import threading
stub = self
self.calls = 0
class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_POST(self):
body = json.loads(self.rfile.read(
int(self.headers["Content-Length"])))
if "id" not in body: # notifications/initialized: 202, uncounted
self.send_response(202)
self.send_header("Content-Length", "0")
self.end_headers()
return
if body["method"] == "initialize":
return self._reply(200, body["id"], {
"protocolVersion": "2025-06-18", "capabilities": {},
"serverInfo": {"name": "stub", "version": "0"}})
stub.calls += 1
if delay:
time.sleep(delay)
self._reply(statuses[min(stub.calls, len(statuses)) - 1],
body["id"],
{"content": [{"type": "text", "text": "ok"}],
"isError": False})
def _reply(self, status, request_id, result):
payload = json.dumps({"jsonrpc": "2.0", "id": request_id,
"result": result}).encode()
self.send_response(status)
if status == 429:
self.send_header("Retry-After", "1") # ignored
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
try:
self.wfile.write(payload)
except OSError: # the client gave up (timeout tests)
pass
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0),
Handler)
self.server.daemon_threads = True
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.url = f"http://127.0.0.1:{self.server.server_port}/mcp"
def close(self):
self.server.shutdown()
self.server.server_close()
@pytest.fixture
def mcp_stub(monkeypatch):
monkeypatch.setenv("NO_PROXY", "127.0.0.1") # keep the machine's proxy out
monkeypatch.setenv("no_proxy", "127.0.0.1") # requests reads this spelling first
stubs = []
def make(statuses, delay=0.0):
stubs.append(_McpStub(statuses, delay))
return stubs[-1]
yield make
for stub in stubs:
stub.close()
def test_bridge_retries_rate_limits_below_the_tool_layer(mcp_stub, monkeypatch):
"""Two 429s then a 200: the call succeeds without the model ever seeing
an error, and the waits are the fixed backoff, not the server's
Retry-After."""
import urllib3.util.retry as retry_module
from pageindex.mcp_bridge import McpBridge
slept = []
monkeypatch.setattr(retry_module.time, "sleep", slept.append)
stub = mcp_stub([429, 429, 200])
assert McpBridge(stub.url, {}).call_tool("get_document", {}) == (
[{"type": "text", "text": "ok"}], False)
assert stub.calls == 3
assert slept == [2]
@pytest.mark.parametrize("status", [429, 504, 529])
def test_bridge_rate_limit_exhausted_raises_with_status(mcp_stub, monkeypatch,
status):
"""Three retries and still failing: the caller gets the status."""
import urllib3.util.retry as retry_module
from pageindex.mcp_bridge import McpBridge
monkeypatch.setattr(retry_module.time, "sleep", lambda seconds: None)
stub = mcp_stub([status])
with pytest.raises(PageIndexAPIError, match=f"HTTP {status}") as info:
McpBridge(stub.url, {}).call_tool("get_document", {})
assert info.value.status_code == status
assert stub.calls == 4
def test_bridge_read_timeout_is_not_retried(mcp_stub, monkeypatch):
"""A read timeout is a full wait the server may have acted on:
surfaced once, never replayed."""
import pageindex.mcp_bridge as mcp_bridge
monkeypatch.setattr(mcp_bridge, "_TIMEOUT", (10, 0.2))
stub = mcp_stub([200], delay=0.6)
with pytest.raises(PageIndexAPIError, match="Could not reach"):
mcp_bridge.McpBridge(stub.url, {}).call_tool("get_document", {})
assert stub.calls == 1
def test_bridge_invoker_reraises_an_unreachable_server(mcp_stub, monkeypatch):
"""A server the bridge could not reach after its own connection retries
escapes to the caller like a 429: the model cannot reach it either."""
import urllib3.util.retry as retry_module
from pageindex.mcp_bridge import McpBridge
monkeypatch.setattr(retry_module.time, "sleep", lambda seconds: None)
stub = mcp_stub([200])
stub.close()
bridge = McpBridge(stub.url, {})
invoke = agent_tools_module._bridge_invoker(bridge, "get_document", {})
with pytest.raises(PageIndexAPIError, match="Could not reach"):
invoke({})
class _RateLimitedBridge(_ImageBridge):
def call_tool(self, name, arguments):
raise PageIndexAPIError("MCP request failed: HTTP 429",
status_code=429)
def test_as_openai_tools_transport_failure_escapes_the_run(monkeypatch):
"""The framework's default turns every tool exception into model-visible
text; the SDK's server narrows that so a failure the invoker re-raised
escapes the run (a model-side slip staying model-visible is covered end
to end in test_local_chat)."""
pytest.importorskip("agents")
import pageindex.mcp_bridge as mcp_bridge
from pageindex.errors import _pageindex_cause
monkeypatch.setattr(mcp_bridge, "McpBridge", _RateLimitedBridge)
tool = PageIndexCloudClient(api_key="pi-test-key").as_openai_tools()[0]
with pytest.raises(Exception) as info:
asyncio.run(tool.on_invoke_tool(None, '{"image_path": "x"}'))
assert _pageindex_cause(info.value).status_code == 429
+218 -1
View File
@@ -1157,6 +1157,34 @@ def _anthropic_message(content, stop_reason):
}
def _anthropic_sse(message):
"""Render text/tool turns for the real streaming tool runner."""
events = [{"type": "message_start", "message": {
**message, "content": [], "stop_reason": None}}]
for index, block in enumerate(message["content"]):
if block["type"] == "tool_use":
initial = {**block, "input": {}}
delta = {"type": "input_json_delta",
"partial_json": json.dumps(block["input"])}
else:
initial = {**block, "text": ""}
delta = {"type": "text_delta", "text": block["text"]}
events.extend([
{"type": "content_block_start", "index": index,
"content_block": initial},
{"type": "content_block_delta", "index": index, "delta": delta},
{"type": "content_block_stop", "index": index},
])
events.extend([
{"type": "message_delta", "delta": {
"stop_reason": message["stop_reason"], "stop_sequence": None},
"usage": {"output_tokens": message["usage"]["output_tokens"]}},
{"type": "message_stop"},
])
return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n"
for event in events)
@pytest.fixture
def fake_anthropic(monkeypatch):
state = {"calls": []}
@@ -3122,7 +3150,7 @@ def test_messages_no_backend_leak_when_tool_build_fails(bridge_client,
lambda backend=None: made.append(FakeAnthropic())
or made[-1])
def boom(client, doc_ids=None):
def boom(client, doc_ids=None, **kwargs):
raise PageIndexAPIError("Could not reach the PageIndex MCP server")
monkeypatch.setattr(
@@ -3339,3 +3367,192 @@ def test_chat_answer_lane_forwards_the_promoted_knobs(client, monkeypatch):
lambda c, messages, **kw: streamed.update(kw) or "s")
assert client.chat("q", stream=True, **knobs) == "s"
assert {k: streamed[k] for k in knobs} == knobs
# ── a tool failure that survived the bridge's retries fails the run fast ──
@needs_agents
def test_translate_run_error_unwraps_a_tool_failure():
"""A failure the invoker re-raised leaves the run wrapped in the
framework's exception; the caller gets it back with its status."""
from agents.exceptions import AgentsException
wrapped = AgentsException("Error invoking MCP tool get_document")
wrapped.__cause__ = PageIndexAPIError("MCP request failed: HTTP 429",
status_code=429)
err = local_chat._translate_run_error(wrapped, None, "chat")
assert err.status_code == 429 and "HTTP 429" in str(err)
plain = local_chat._translate_run_error(AgentsException("boom"), None,
"chat")
assert plain.status_code is None and "agent backend failed" in str(plain)
def test_model_backend_error_keeps_the_status_code():
limited = Exception("rate limited")
limited.status_code = 429
assert local_chat._model_backend_error(limited, "chat").status_code == 429
assert local_chat._model_backend_error(
Exception("x"), "chat").status_code is None
def _rate_limited(name, arguments):
raise PageIndexAPIError("MCP request failed: HTTP 429", status_code=429)
@needs_agents
def test_bridge_chat_fails_fast_on_a_rate_limited_tool(bridge_client,
fake_model):
"""A 429 that survived the bridge's retries ends the run with its
status — no second model turn over an error envelope."""
client, bridge = bridge_client
bridge.call_tool = _rate_limited
fake = fake_model([
[_call_item("get_document", {"doc_name": "r.pdf"})],
[_msg_item("never reached")],
])
with pytest.raises(PageIndexAPIError, match="HTTP 429") as info:
client.chat_completions("What?")
assert info.value.status_code == 429
assert len(fake.instructions) == 1
@needs_anthropic
def test_messages_fail_fast_is_quiet(bridge_client, fake_anthropic, caplog):
"""The runner's own tool-error logging never reports the failure the
lane is about to raise."""
client, bridge = bridge_client
bridge.call_tool = _rate_limited
fake_anthropic([_anthropic_message(
[{"type": "tool_use", "id": "tu_1", "name": "get_document",
"input": {"doc_name": "r.pdf"}}], "tool_use")])
with pytest.raises(PageIndexAPIError, match="HTTP 429"):
client.chat("q", protocol="messages", model="claude-test",
extra_body={"max_tokens": 100})
assert not [r for r in caplog.records if r.name.startswith("anthropic")]
@needs_anthropic
@pytest.mark.parametrize("stop_reason", ["tool_use", "max_tokens", "refusal"])
@pytest.mark.parametrize("stream", [False, True])
@pytest.mark.parametrize("max_turns", [1, 2])
def test_messages_fail_fast_on_a_rate_limited_tool(
bridge_client, fake_anthropic, stop_reason, stream, max_turns):
"""Real runners must surface executed tools' failures before another
model call or a max_turns exit, and preserve terminal-turn policy."""
client, bridge = bridge_client
tool_calls = []
def fail(name, arguments):
tool_calls.append((name, arguments))
return _rate_limited(name, arguments)
bridge.call_tool = fail
replies = [
_anthropic_message([{"type": "tool_use", "id": "tu_1",
"name": "get_document",
"input": {"doc_name": "r.pdf"}}], stop_reason),
_anthropic_message([{"type": "text", "text": "never reached"}],
"end_turn"),
]
calls = fake_anthropic([_anthropic_sse(reply) for reply in replies]
if stream else replies)
def run():
result = client.chat("q", protocol="messages", model="claude-test",
extra_body={"max_tokens": 100}, stream=stream,
max_turns=max_turns)
return list(result) if stream else result
executes_tools = (stop_reason == "tool_use"
or (stop_reason == "max_tokens"
and _ANTHROPIC_RUNS_CUT_TOOL_TURNS))
if executes_tools:
with pytest.raises(PageIndexAPIError, match="HTTP 429") as info:
run()
assert info.value.status_code == 429
assert tool_calls == [("get_document", {"doc_name": "r.pdf"})]
else:
run()
assert tool_calls == []
assert len(calls) == 1, "a second model turn ran"
@needs_anthropic
@pytest.mark.parametrize("stream", [False, True])
def test_messages_runs_each_tool_once(bridge_client, fake_anthropic, stream):
"""Failure checks preserve normal tool execution across multiple turns."""
client, bridge = bridge_client
replies = [_anthropic_message([
{"type": "tool_use", "id": tool_id, "name": "get_document",
"input": {"doc_name": "r.pdf"}}], "tool_use")
for tool_id in ("tu_1", "tu_2")]
replies.append(_anthropic_message([{"type": "text", "text": "Done"}],
"end_turn"))
calls = fake_anthropic([_anthropic_sse(reply) for reply in replies]
if stream else replies)
result = client.chat("q", protocol="messages", model="claude-test",
extra_body={"max_tokens": 100}, stream=stream)
if stream:
list(result)
assert len(calls) == 3
assert bridge.calls == [("get_document", {"doc_name": "r.pdf"})] * 2
@needs_anthropic
@pytest.mark.parametrize("stream", [False, True])
@pytest.mark.parametrize("cached", [False, True])
def test_messages_tool_failure_preserves_client_lifecycle(
bridge_client, fake_anthropic, monkeypatch, stream, cached):
"""Failed runs close owned clients and leave cached clients reusable,
with no failure state leaking into the next run."""
client, bridge = bridge_client
replies = [
_anthropic_message([_anthropic_tool_use()], "tool_use"),
_anthropic_message([_anthropic_tool_use()], "tool_use"),
_anthropic_message([{"type": "text", "text": "Recovered"}],
"end_turn"),
]
calls = fake_anthropic([_anthropic_sse(reply) for reply in replies]
if stream else replies)
backend = local_chat._anthropic_client()
monkeypatch.setattr(local_chat, "_ANTHROPIC_CLIENTS",
{"test": backend} if cached else {})
original_call = bridge.call_tool
bridge.call_tool = _rate_limited
def run():
result = client.chat("q", protocol="messages", model="claude-test",
extra_body={"max_tokens": 100}, stream=stream)
return list(result) if stream else result
try:
with pytest.raises(PageIndexAPIError, match="HTTP 429"):
run()
assert len(calls) == 1
assert backend.is_closed() is (not cached)
if cached:
bridge.call_tool = original_call
run()
assert len(calls) == 3
assert len(bridge.calls) == 1
assert not backend.is_closed()
finally:
backend.close()
@needs_agents
def test_bridge_chat_keeps_model_slips_model_visible(bridge_client,
fake_model):
"""Only the invoker's re-raised failures escape: a model-side slip (bad
JSON arguments) still comes back to the model as text and the run
goes on."""
from openai.types.responses import ResponseFunctionToolCall
client, bridge = bridge_client
fake = fake_model([
[ResponseFunctionToolCall(id="fc_1", type="function_call",
call_id="call_1", name="get_document",
arguments="{not json", status="completed")],
[_msg_item("Recovered")],
])
result = client.chat_completions("What?")
assert result["choices"][0]["message"]["content"] == "Recovered"
assert len(fake.instructions) == 2 and bridge.calls == []