Bounded web-tool fallback with failure receipts in AutoGen

[auxiliar-ag]

Sharing a small, self-contained pattern I keep re-deriving whenever an AutoGen agent is given web access: bound the tool, and make its failures structured data the model can reason over.

This is a reusable pattern and example code, not an official AutoGen integration and not a package anyone needs to depend on. It is a couple hundred lines you can paste into your own project and adapt. Everything runs on autogen-core 0.7.5 with no API key and no network — the providers are deterministic scripted transports, while FunctionTool, CancellationToken and AssistantAgent are the real AutoGen ones.

The problem

A naive web tool tends to look like this:

async def web_search(query: str) -> str:
    try:
        return await provider.search(query)
    except Exception as exc:
        return f"Search failed: {exc}"

Three separate things go wrong with it, and they compound.

1. The failure is prose. "Search failed: HTTPSConnectionPool(host='...', port=443): Read timed out" lands in the model's context as an English sentence. The model has to infer whether to retry, answer from memory, or apologise. Different phrasings on different days produce different behaviour, and you cannot write an assertion against it.

2. Transport success is treated as usefulness. HTTP 200 means a server answered. It does not mean the answer helps. Search APIs return empty result sets, consent interstitials, cookie walls and "enable JavaScript" pages with a 200 and a well-formed body. The code above returns all of those as success, and the model — which cannot see that the body is empty of evidence — cites them.

3. There is no bound. Add a fallback provider and a retry loop and the tool call's worst case is now providers × retries × timeout. Two providers at 30s each is a 60s tool call inside a turn the user is watching.

The four rules

1. Transport success and usefulness are separate questions, asked in that order, reported as separate fields.
2. At most one fallback. Not a loop over a provider list; one alternate, structurally.
3. One deadline shared by every attempt, not a timeout per attempt.
4. The tool always returns a receipt, never raises, and the receipt is drawn from a closed vocabulary.

The receipt

The vocabulary is the whole point. An agent's behaviour becomes a function of a small finite set of tokens instead of an open set of English error strings — so a prompt can enumerate it exhaustively and a test can assert on it exactly.

class Outcome(str, Enum):
    """Terminal outcome of one bounded lookup. Exhaustive and mutually exclusive."""

    USEFUL_RESULT = "useful_result"
    """A transport succeeded *and* the payload passed usefulness validation."""

    NO_USEFUL_RESULT = "no_useful_result"
    """Every attempt reached a transport, but no payload was judged useful."""

    TRANSPORT_FAILED = "transport_failed"
    """No attempt produced a payload at all (error, timeout, rate limit, auth)."""

    BUDGET_EXHAUSTED = "budget_exhausted"
    """The shared deadline ran out before an attempt could be completed."""

    BLOCKED_BY_POLICY = "blocked_by_policy"
    """The request was refused locally; no transport was contacted."""

Each outcome maps to exactly one instruction:

class NextAction(str, Enum):
    """The single instruction the calling agent is expected to follow."""

    USE_RESULT = "use_result"
    ANSWER_FROM_PRIOR_KNOWLEDGE = "answer_from_prior_knowledge"
    ASK_USER_TO_NARROW = "ask_user_to_narrow"
    RETRY_LATER = "retry_later"
    STOP = "stop"

The mapping is total, and one branch is worth calling out:

def decide_next_action(outcome: Outcome, attempts: Sequence[AttemptRecord]) -> NextAction:
    """Map a terminal outcome onto exactly one instruction for the agent.

    The mapping is total: every :class:`Outcome` member has a branch. Credential
    failures are singled out because retrying them later cannot help.
    """
    if outcome is Outcome.USEFUL_RESULT:
        return NextAction.USE_RESULT
    if outcome is Outcome.BLOCKED_BY_POLICY:
        return NextAction.STOP
    if outcome is Outcome.BUDGET_EXHAUSTED:
        return NextAction.ANSWER_FROM_PRIOR_KNOWLEDGE
    if outcome is Outcome.NO_USEFUL_RESULT:
        return NextAction.ASK_USER_TO_NARROW
    if outcome is Outcome.TRANSPORT_FAILED:
        if any(a.status is AttemptStatus.AUTH_FAILED for a in attempts):
            return NextAction.STOP
        return NextAction.RETRY_LATER
    raise AssertionError(f"unhandled outcome: {outcome!r}")  # pragma: no cover

auth_failed collapses to stop rather than retry_later, because a rejected credential will still be rejected in thirty seconds. Telling the agent to retry there just buys a second failed turn.

The Receipt model is frozen=True, extra="forbid" and has no field capable of holding a response body — no body, no html, no headers, no raw. Bodies cannot leak through it by accident, because there is nowhere to put them. A test asserts that property directly rather than trusting the reviewer to notice:

def test_attempt_record_has_no_body_field() -> None:
    """A receipt cannot leak a response body it has nowhere to store."""
    forbidden = {"body", "raw", "payload", "text", "html", "headers", "response"}
    assert forbidden.isdisjoint(AttemptRecord.model_fields)
    assert forbidden.isdisjoint(Receipt.model_fields)

Rule 1: arriving is not the same as being useful

The engine asks two independent questions. Did a payload arrive — a TransportResponse rather than a TransportError? And is that payload usable? The second question belongs to a validator, and its answer is a closed-vocabulary reason:

    def validate(self, query: str, response: TransportResponse) -> Verdict:
        if not response.parse_ok:
            return Verdict(False, NotUsefulReason.UNPARSEABLE_PAYLOAD)

        if len(response.hits) < self.min_hits:
            return Verdict(False, NotUsefulReason.EMPTY_RESULT_SET)

        blob = " ".join(f"{h.title} {h.snippet}" for h in response.hits).lower()

        for marker in self.login_wall_markers:
            if marker in blob:
                return Verdict(False, NotUsefulReason.LOGIN_WALL)

        if max((len(h.snippet.strip()) for h in response.hits), default=0) < self.min_snippet_chars:
            return Verdict(False, NotUsefulReason.CONTENT_TOO_SHORT)

        if self.require_term_overlap:
            wanted = _terms(query)
            if wanted and not (wanted & _terms(blob)):
                return Verdict(False, NotUsefulReason.NO_QUERY_TERMS_MATCHED)

        return USEFUL

The ordering is fixed so that the reported reason is deterministic — that matters when you are diffing evaluation runs. The thresholds (min_snippet_chars, the login-wall markers, whether query terms must overlap) are the part you should expect to tune against your own traffic rather than inherit; more on that below.

The practical consequence is that a 200 carrying zero hits is recorded as ok_not_useful with not_useful_reason=empty_result_set, and it triggers the fallback. Naive code returns it as a success.

Rule 3: one shared deadline

The budget is deliberately not per-attempt. A single Budget instance is threaded through the entire plan, and each attempt may only borrow a slice of what remains:

    def slice_ms(self, cap_ms: int) -> int:
        """The largest per-attempt timeout allowed right now."""
        if cap_ms < 0:
            raise ValueError("cap_ms must be non-negative")
        return min(int(cap_ms), self.remaining_ms())

An attempt that cannot get a meaningful slice is never started. It is recorded as skipped_no_budget instead — a distinct term, so the receipt distinguishes "the fallback was tried and failed" from "the fallback never got a chance":

    target = redact_url(transport.target_for(query))
    started_ms = budget.elapsed_ms()
    slice_ms = budget.slice_ms(policy.per_attempt_ms)

    if slice_ms < policy.min_attempt_ms:
        return (
            AttemptRecord(
                index=index,
                role=role,
                transport=transport.name,
                status=AttemptStatus.SKIPPED_NO_BUDGET,
                elapsed_ms=0,
                target=target,
            ),
            None,
        )

The slice is handed to the transport explicitly, the way any real HTTP client takes a timeout, with wait_for as a backstop for a transport that ignores it. CancellationToken.link_future is what wires AutoGen's cancellation through to the in-flight request:

    task = asyncio.ensure_future(transport.fetch(request, slice_ms, cancellation_token))
    cancellation_token.link_future(task)
    try:
        # The transport is given the slice explicitly, the way a real HTTP client
        # takes a timeout. wait_for is the backstop for a transport that ignores it.
        response = await asyncio.wait_for(task, timeout=slice_ms / 1000.0)
    except (TransportError, asyncio.TimeoutError, asyncio.CancelledError) as error:
        status = _classify_error(error)
    else:
        http_status = response.http_status
        verdict = validator.validate(query, response)
        status = AttemptStatus.OK_USEFUL if verdict.useful else AttemptStatus.OK_NOT_USEFUL
        reason = verdict.reason

The clock is injectable (clock: Clock = time.monotonic). That is what makes the tests and the demo deterministic: a ManualClock advances only by the latency a scripted transport declares, so simulated 5-second providers cost zero real seconds and the timing assertions are exact rather than flaky.

Rule 2: at most one fallback, structurally

There is no provider list and no loop. The fallback is a single optional argument, FallbackPolicy rejects max_fallbacks > 1 at construction time, and the engine body contains exactly one conditional second attempt:

    # Exactly one fallback slot. No loop, no third attempt, no per-attempt budget reset.
    if useful_response is None and fallback_allowed:
        assert fallback is not None
        record, response = await _run_attempt(
            index=1,
            role=AttemptRole.FALLBACK,
            transport=fallback,
            query=normalised,
            budget=budget,
            policy=policy,
            validator=validator,
            cancellation_token=token,
        )
        attempts.append(record)
        if record.status is AttemptStatus.OK_USEFUL:
            useful_response = response

Bounding by construction rather than by convention means the invariant survives the next person editing the file.

Reducing the trail to one outcome has one subtle branch, which I got wrong first and only noticed because a test disagreed with me:

def _terminal_outcome(attempts: Sequence[AttemptRecord]) -> Outcome:
    """Reduce the attempt trail to one terminal outcome.

    Order matters, and the second branch is the subtle one. If the plan was cut
    short because the shared budget ran out, that is reported *even when an
    earlier attempt did return a useless payload*. The two outcomes lead the
    agent to different behaviour -- ``no_useful_result`` asks the user to narrow
    the query, ``budget_exhausted`` falls back to prior knowledge -- and asking
    someone to rephrase is misleading when the real story is that the plan never
    finished. ``no_useful_result`` therefore means "the plan ran to completion
    and nothing usable came back", which is the only reading that justifies
    blaming the query.
    """
    if any(a.status is AttemptStatus.OK_USEFUL for a in attempts):
        return Outcome.USEFUL_RESULT
    if attempts and attempts[-1].status is AttemptStatus.SKIPPED_NO_BUDGET:
        return Outcome.BUDGET_EXHAUSTED
    if any(a.status in _PAYLOAD_STATUSES for a in attempts):
        return Outcome.NO_USEFUL_RESULT
    return Outcome.TRANSPORT_FAILED

My first version checked "a payload arrived" before "the budget ran out", which produced no_useful_result → ask_user_to_narrow for a run where the primary returned an empty page and the fallback was then starved of runway. Asking the user to rephrase their question is actively misleading when the real story is that the plan never finished.

Wiring it to AutoGen

This is the only module that imports AutoGen. FunctionTool derives the argument schema from the wrapped function's annotations, and BaseTool.return_value_as_string serialises a pydantic return value with model_dump() + json.dumps. So returning a Receipt puts the entire receipt into the model's context as JSON with no formatting code of my own:

def make_bounded_web_search_tool(
    primary: Transport,
    fallback: Transport | None = None,
    *,
    policy: FallbackPolicy = DEFAULT_POLICY,
    validator: UsefulnessValidator = DEFAULT_VALIDATOR,
    clock: Clock = time.monotonic,
    name: str = DEFAULT_TOOL_NAME,
    description: str = TOOL_DESCRIPTION,
) -> FunctionTool:
    """Wrap :func:`bounded_web_lookup` as an AutoGen ``FunctionTool``.

    The transports, policy, validator and clock are captured in the closure, so
    the model-visible schema is exactly one argument: ``query``.
    """

    async def bounded_web_search(query: str) -> Receipt:
        return await bounded_web_lookup(
            query,
            primary,
            fallback,
            policy=policy,
            validator=validator,
            clock=clock,
        )

    return FunctionTool(bounded_web_search, description=description, name=name)

Transports, policy, validator and clock live in the closure, so the model-visible schema is exactly one field:

def test_schema_exposes_exactly_one_query_argument(clock: ManualClock) -> None:
    """Transports, policy and clock live in the closure, so the model sees one field."""
    tool = make_bounded_web_search_tool(scripted("primary", [ok(100)], clock), clock=clock)
    schema = tool.schema
    assert schema["name"] == DEFAULT_TOOL_NAME
    parameters = schema["parameters"]
    assert parameters["type"] == "object"
    assert list(parameters["properties"]) == ["query"]
    assert parameters["properties"]["query"]["type"] == "string"
    assert parameters["required"] == ["query"]

The system-message fragment that teaches the model the vocabulary is generated from the enums, so the prompt cannot drift away from the code — a test iterates Outcome and NextAction and asserts every member appears in it.

A complete runnable example

This is self-contained; it needs autogen-core and this package, no key and no network:

import asyncio

from bounded_web_tool import (
    FallbackPolicy,
    ManualClock,
    ScriptedStep,
    ScriptedTransport,
    TransportResponse,
    bounded_web_lookup,
    hits,
)

GOOD = hits(
    (
        "FunctionTool reference",
        "https://microsoft.github.io/autogen/stable/reference/python/autogen_core.tools.html",
        "FunctionTool derives its argument schema from the wrapped function's type annotations.",
    )
)


async def main() -> None:
    clock = ManualClock()
    # A perfectly valid HTTP 200 that happens to contain nothing usable.
    primary = ScriptedTransport(
        "primary",
        [ScriptedStep(300, response=TransportResponse(200, ()))],
        clock,
        base_url="https://primary.invalid/search",
    )
    fallback = ScriptedTransport(
        "fallback",
        [ScriptedStep(400, response=TransportResponse(200, GOOD))],
        clock,
        base_url="https://fallback.invalid/search",
    )

    receipt = await bounded_web_lookup(
        "autogen functiontool schema",
        primary,
        fallback,
        policy=FallbackPolicy(budget_ms=2000, per_attempt_ms=1200),
        clock=clock,
    )
    print(receipt.explain())
    assert receipt.outcome.value == "useful_result"
    assert receipt.fallback_used is True
    assert receipt.elapsed_ms == 700  # 300ms primary + 400ms fallback, one shared budget


asyncio.run(main())

which prints:

outcome=useful_result next_action=use_result attempts=2/2 elapsed_ms=700/2000 trail=[primary:ok_not_useful -> fallback:ok_useful]

What the demo shows

python examples/demo.py walks eight scenarios. Abridged transcript (real output, byte-for-byte reproducible because the clock is injected):

--- 1. primary succeeds, fallback never dialled ---
outcome=useful_result next_action=use_result attempts=1/2 elapsed_ms=300/2000 trail=[primary:ok_useful]
    fallback fetch() calls: 0

--- 2. HTTP 200 with an empty result set is a failure, so fall back ---
outcome=useful_result next_action=use_result attempts=2/2 elapsed_ms=700/2000 trail=[primary:ok_not_useful -> fallback:ok_useful]

--- 3. both providers answered, neither was usable ---
outcome=no_useful_result next_action=ask_user_to_narrow attempts=2/2 elapsed_ms=600/2000 trail=[primary:ok_not_useful -> fallback:ok_not_useful]

--- 4. two 5s providers under a 2s shared budget ---
outcome=transport_failed next_action=retry_later attempts=2/2 elapsed_ms=2000/2000 trail=[primary:transport_timeout -> fallback:transport_timeout]

--- 5. the fallback is skipped rather than started without runway ---
outcome=budget_exhausted next_action=answer_from_prior_knowledge attempts=1/2 elapsed_ms=1150/1300 trail=[primary:ok_not_useful -> fallback:skipped_no_budget]
    fallback fetch() calls: 0

--- 6. rejected credentials advise stop, not retry_later ---
outcome=transport_failed next_action=stop attempts=2/2 elapsed_ms=160/2000 trail=[primary:auth_failed -> fallback:auth_failed]

--- 7. local refusal, no transport contacted ---
outcome=blocked_by_policy next_action=stop attempts=0/1 elapsed_ms=0/2000 trail=[none]
    primary fetch() calls: 0

Scenario 4 is the one I care about most: two providers that each take five seconds, under a two-second budget, return in exactly 2000ms of simulated time — 1200ms for the primary's slice, then only the 800ms that was left.

Scenario 8 runs the same tool inside a real AssistantAgent, driven by ReplayChatCompletionClient from autogen-ext so the model turns are pre-recorded and no key is needed. That exercises AutoGen's actual tool-execution path, and the tool result the model receives is the receipt JSON:

"outcome": "useful_result",
"next_action": "use_result",
"fallback_used": true,
"attempts_used": 2,
"elapsed_ms": 700,
"budget_ms": 2000,

To be precise about what that proves: the model there is scripted, so it demonstrates the plumbing — that a receipt survives tool execution intact and arrives in the model context as JSON the next turn can condition on. It does not demonstrate that a live model obeys next_action. Measuring that needs real models and a real corpus.

Redaction

Two mechanisms, because one is not enough:

• The receipt has no field that can hold a body, so bodies cannot leak structurally.
• Everything that does reach the context — the recorded request target and the bounded content extract of a useful result — goes through redaction. Sensitive query parameters are masked, URL userinfo is dropped, and credential-shaped substrings (bearer tokens, JWTs, sk-/AKIA/ghp_/xox prefixes, long hex blobs, key=value assignments for sensitive names) are replaced.

Redaction deliberately over-matches. A redacted URL keeps its non-sensitive parameters so it stays recognisable in a receipt, but ?q=x&api_key=abcd1234&page=2 becomes ?q=x&api_key=[REDACTED]&page=2.

Trade-offs and limitations

• One fallback is a real ceiling. If your primary is flaky enough that two attempts routinely fail, this pattern surfaces that honestly rather than papering over it with a third attempt. That is the intent, but it is a constraint, not a free win.
• The validator is the hard part, and it is domain-specific. EvidenceValidator is a starting point, not a general answer. Marker lists and length thresholds are exactly the kind of heuristic that looks fine on your fixtures and misclassifies real traffic; treat the defaults here as a scaffold to replace. Calibrating thresholds like these against real pages is its own exercise — I've written up a longer public web-access evaluation methodology covering how to assemble that kind of corpus.
• A closed vocabulary is a versioned interface. Adding an Outcome member is a breaking change for every prompt and evaluation that enumerates it, which is why schema_version is on the receipt and a test pins the exact member lists.
• Scripted transports are not real networks. They model latency, timeouts, rate limits, auth failures and useless-but-valid payloads. They do not model partial reads, TLS problems, redirect chains, or content that is useless in ways my validator does not check for.

Running it

python -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest          # deterministic, offline
.venv/bin/python examples/demo.py   # no API key, no network

Verification

Rather than assert that this is well tested, here is what was actually run:

• autogen-core, autogen-agentchat, autogen-ext 0.7.5, Python 3.11.15, in a clean venv.
• 105 tests pass, pytest exit code 0. The suite is fully offline and uses no clock-dependent sleeps.
• ruff check and mypy --strict both clean.
• The demo is executed by a test that runs it twice and asserts the output is byte-identical, and that the transcript lines quoted above are present.
• Every Python block in this post is checked by a test that either matches it byte-for-byte against the corresponding source file or executes it in a subprocess and requires exit 0. The excerpts above are not retyped or paraphrased.

Disclosure

I work at NativePort, which builds tooling in the web-access space, so the methodology link above is to my employer's site; the pattern and code here are framework-agnostic and carry no dependency on anything we sell. This post and the accompanying code were drafted with AI assistance. I am deliberately not making claims about model behaviour that the offline test suite cannot support — the verification section above lists what was actually executed, and nothing beyond that has been measured.

Happy to hear where this breaks down, particularly on the validator design — that is the piece I am least confident generalises.