A task-contract pattern for LlamaIndex tools that call external web APIs

[auxiliar-ag]

A task-contract pattern for LlamaIndex tools that call external web APIs

When an agent tool wraps a web search, scrape, or browser action, the easy failure mode is treating a 200 response as a done task. A search API can return 200 with zero relevant results. A scrape can return 200 with a bot wall instead of the page. If the tool hands that straight to the LLM as "success," the agent reasons from a false premise and the bug is invisible until someone reads the transcript.

Below is a small pattern for a LlamaIndex FunctionTool that keeps HTTP status and task success as two separate questions, and pins the pass criteria outside the model's reach so a prompt cannot redefine what counts as done. The full code is inline, so you can paste it into a file and run it against llama-index-core with no network access and no API key.

The tool takes a task id, not a URL

Instead of a generic fetch(url) tool where the caller supplies both the target and, implicitly, the expectation of a good result, this tool accepts only a task_id. The URL and the checks that decide pass or fail live in a small contract registered ahead of time, and the transport is swappable, a fake for tests and demos today, a real HTTP client later, without touching the contract or the tool logic:

from dataclasses import dataclass
from typing import Literal
from llama_index.core.tools import FunctionTool

@dataclass(frozen=True)
class Check:
    kind: Literal["min_chars", "contains", "not_contains"]
    value: str | int

def evaluate_check(check: Check, body: str) -> bool:
    text = body.lower()
    if check.kind == "min_chars":
        return len(body.strip()) >= int(check.value)
    if check.kind == "contains":
        return str(check.value).lower() in text
    if check.kind == "not_contains":
        return str(check.value).lower() not in text
    raise ValueError(f"unknown check kind: {check.kind!r}")

@dataclass(frozen=True)
class TaskContract:
    task_id: str
    url: str
    checks: tuple

@dataclass(frozen=True)
class TransportResponse:
    status_code: int
    body: str

class TransportError(Exception):
    pass

class FakeTransport:
    def __init__(self, responses=None, failures=None):
        self._responses, self._failures = responses or {}, failures or {}

    def get(self, url: str, timeout_s: float) -> TransportResponse:
        if url in self._failures:
            raise self._failures[url]
        if url not in self._responses:
            raise TransportError(f"no fixture for {url}")
        return self._responses[url]

def make_fetch_and_verify_tool(tasks: dict, transport) -> FunctionTool:
    def fetch_and_verify(task_id: str) -> dict:
        """Fetch a pinned task URL and verify its content against a fixed contract."""
        contract = tasks.get(task_id)
        if contract is None:
            return {"task_id": task_id, "outcome": "contract_error",
                     "reason": f"unknown task_id: {task_id!r}"}
        try:
            response = transport.get(contract.url, timeout_s=5.0)
        except TransportError as exc:
            return {"task_id": task_id, "outcome": "transport_failure", "reason": str(exc)}
        if not (200 <= response.status_code < 300):
            return {"task_id": task_id, "outcome": "transport_failure",
                     "reason": f"http status {response.status_code}"}
        checks = [{"kind": c.kind, "passed": evaluate_check(c, response.body)} for c in contract.checks]
        if any(not c["passed"] for c in checks):
            return {"task_id": task_id, "outcome": "content_failure", "checks": checks}
        return {"task_id": task_id, "outcome": "ok", "checks": checks}

    return FunctionTool.from_defaults(fn=fetch_and_verify, name="fetch_and_verify")

make_fetch_and_verify_tool returns a plain FunctionTool.from_defaults(...), so it plugs into FunctionAgent, ReActAgent, or any workflow step that already consumes LlamaIndex tools. Nothing about the wrapping is special cased.

Four outcomes, not two

The tool returns one of four outcomes: ok, content_failure, transport_failure, or contract_error. A content_failure means the HTTP call succeeded but the checks did not, the bot-wall case above. A transport_failure covers non-2xx status and exceptions like timeouts, kept separate from content problems because they call for different agent behavior, retry versus give up on this source. contract_error fires when the model asks for a task_id that was never registered, a different bug from either of the above.

Running the tool against a fixture that returns 200 with a bot-wall body, using the code above, prints:

tasks = {"pricing-page": TaskContract("pricing-page", "https://example.com/pricing",
         (Check("min_chars", 400), Check("contains", "per month"), Check("not_contains", "enable javascript")))}
transport = FakeTransport(responses={"https://example.com/pricing":
            TransportResponse(200, "Please enable JavaScript to view this page. " * 12)})
tool = make_fetch_and_verify_tool(tasks, transport)
print(tool.call(task_id="pricing-page").raw_output)
# {'task_id': 'pricing-page', 'outcome': 'content_failure',
#  'checks': [{'kind': 'min_chars', 'passed': True},
#             {'kind': 'contains', 'passed': False},
#             {'kind': 'not_contains', 'passed': False}]}

The status code alone would have said this call succeeded. The contract says otherwise, and the agent gets a reason it can act on instead of a wall of HTML it has to guess about.

This is not tied to any one provider's API, on purpose. I did not want to guess at request or response shapes for a live web service and pass that off as a working integration, so the pattern stops at the transport boundary and leaves the real HTTP client as an exercise. If you want the reasoning behind treating quality, latency, cost, and error rate as separate measurements instead of one pass or fail number, that is what NativePort's write-up on how it grades API responses covers in more detail. It is where I got in the habit of treating "the call succeeded" and "the answer is right" as different checks.

Has anyone here built a stricter contract layer in front of FunctionTool, for example with Pydantic validation on the checks themselves, or wired outcome codes like these into an agent's retry or fallback-tool logic? I would like to compare notes before taking this further than one tool.

Disclosure: I work on NativePort. This post was drafted with AI assistance and reviewed, fact-checked, and tested by a human before posting; the code above and its test suite ran successfully prior to publication.

[shipsitself]

Strongly agree with the framing, and I can offer numbers for the "200 is not done" half from a research agent we instrumented.

Our tool contract requires every claim to carry a verbatim quote plus the URL it came from, and deterministic code then checks the URL is one we actually fetched and the quote appears in those exact cached bytes. Two things fell out of running it at batch scale that a status-code check would never surface:

Dead is not one category. In one batch, ten unique proposed source URLs came back non-200. We hand-adjudicated seven with controls — for every URL we claimed had never existed, we ran the same lookup against a URL we knew did exist and confirmed the lookup found it. At least six had never existed as paths; one was a real page with a single character missing from the slug. Collapsing those into one "failed fetch" number would have made the model look far worse than it was, and would have hidden that one of them was a near-miss worth retrying rather than a fabrication.

The contract must pin the bytes, not the URL. The rule that made ours trustworthy: the exact text slice handed to the model is the exact slice the checker searches later, cached. Re-fetching at validation time means validating against a slightly different document than the one that produced the output, and the difference shows up as inexplicable false negatives.

That last point has a sharp edge. Our checker once rejected 25 of 25 quotes as unfindable and we nearly reported it as a model-honesty result — every rejection was our own HTML-to-text step collapsing tags into spaces, so Doppler, who became Doppler , who. Since then the contract includes the normaliser, and we verify the verifier before trusting it: feed it deliberately corrupted claims — one digit changed, a word dropped, a noun swapped, a sentence invented — and require each to still fail.

For your bot-wall case specifically, the cheap version of all this is asserting on a property of the content the task claims to have retrieved, not on the transport result. A 200 with a bot wall fails "does the quote I am about to rely on appear in these bytes" immediately.

[amasen02]

We’ve implemented a very similar task-contract pattern in production with LlamaIndex’s newer Workflow / AgentWorkflow engine. Here are three architectural patterns that made this pattern significantly more reliable when scaling beyond a single tool:

1. Pydantic V2 Discriminated Models over Raw Dictionaries

Instead of returning a loose dict, model the outcomes as Pydantic models with Literal discriminator tags:

from typing import Literal, Union
from pydantic import BaseModel, Field

class CheckResult(BaseModel):
    kind: str
    passed: bool

class ContractSuccess(BaseModel):
    outcome: Literal["ok"] = "ok"
    task_id: str
    content_hash: str
    checks: list[CheckResult]

class ContentFailure(BaseModel):
    outcome: Literal["content_failure"] = "content_failure"
    task_id: str
    reason: str
    failed_checks: list[CheckResult]

class TransportFailure(BaseModel):
    outcome: Literal["transport_failure"] = "transport_failure"
    task_id: str
    http_status: int | None = None
    retry_after_s: float | None = None

TaskContractResult = Union[ContractSuccess, ContentFailure, TransportFailure]

When wrapping this inside FunctionTool.from_defaults(fn=...), LlamaIndex automatically reflects the Pydantic type annotations into the tool's JSON Schema.
Models (GPT-4o, Claude 3.5, Gemini) parse the structured output with zero ambiguity, and LlamaIndex handles serialization cleanly.

2. Wiring Outcomes into LlamaIndex Workflow Branching

The most valuable part of separating content_failure from transport_failure is in the event-driven orchestration step.

In a LlamaIndex Workflow:

from llama_index.core.workflow import Workflow, step, Event

class ToolExecutionEvent(Event):
    result: TaskContractResult

class SearchWorkflow(Workflow):
    @step
    async def handle_tool_result(self, ev: ToolExecutionEvent) -> Event:
        match ev.result.outcome:
            case "ok":
                return SynthesisEvent(data=ev.result)
            case "transport_failure":
                # Network glitch or rate limit: retry the same transport with backoff
                return RetryTransportEvent(task_id=ev.result.task_id, delay=ev.result.retry_after_s or 2.0)
            case "content_failure":
                # Bot-wall or empty DOM: do NOT retry the same URL! Divert to fallback provider or cache
                return FallbackProviderEvent(task_id=ev.result.task_id)

This completely stops the agent from getting stuck in an infinite retry loop against a Cloudflare challenge page, while still letting transient 503 network errors recover automatically.

3. Canonical Fingerprinting (Addressing Whitespace & DOM Drift)

To @shipsitself's crucial point on whitespace normalization breaking quote checks:
Never compare raw unnormalized strings between the scraper and the verification checker.
Compute a canonical fingerprint on the stripped, whitespace-collapsed visible text:
fingerprint = sha256(re.sub(r'\s+', ' ', raw_text).strip().encode()).hexdigest()

Store that content_hash directly on ContractSuccess. When downstream evaluator steps verify that an assertion or quote exists, they run the same single-space regex before substring searching. That eliminates 100% of the false-negative quote rejections caused by HTML tag stripping or newline variations.

[KRMAGTU]

This is a strong contract pattern, particularly because the acceptance criteria remain outside the model’s control.

The place I would extend it is when the tool performs a state-changing action rather than retrieving content.

For a read operation, transport status plus content checks may establish whether the returned material satisfies the declared retrieval contract.

For a write or consequential external action, the tool response still may not establish the required destination reality:

• the external effect may commit but its response may be lost;
• the API may return 200 while the destination state remains wrong;
• only part of a multi-effect objective may be present;
• authentic readback may belong to the wrong resource or version;
• later activity may change the destination before verification;
• compensation may run successfully without restoring the original objective.

UAEP treats these as separate facts:

1. transport or dispatch result;
2. executor/tool result;
3. independently observed destination state;
4. applicability of that evidence to the exact objective and execution;
5. the final assurance verdict.

A lost response after a possible commit should not automatically become transport_failure if retrying could duplicate the effect. Until authoritative destination reconciliation establishes what occurred, the truthful state may be UNKNOWN.

Similarly, an authentic receipt or readback should not promote the objective to VERIFIED unless it is bound to the correct objective, execution, resource, version, authority and environment.

The current UAEP implementation is complete and internally qualified within its declared evaluation boundary. It has been exercised against lost-response commits, wrong destination state, stale or mismatched evidence, partial effects, unsafe retries and unresolved recovery.

Across the bounded qualified demonstrations:

FALSE VERIFIED = 0

That is bounded evidence, not a claim that UAEP cannot fail. The open falsification challenge is here:

https://community.openai.com/t/i-built-uaep-to-separate-agent-execution-success-from-verified-reality-try-to-make-it-lie/1398398

Your fixed task-contract pattern is a strong foundation. For consequential tools, I would add an explicit independent destination observer, evidence binding and an UNKNOWN path before allowing any retry or completion claim.

[send21io]

Payments are a clean example of the state-changing case in the last comment, so here is the original contract pattern mapped onto one, with the "UNKNOWN until reconciled" state made concrete.

The tool takes an invoice id, not an amount. It's the same idea as task_id. Amount, currency and receiving address live in a registry the model can't edit, so a prompt can't redefine what gets billed.

The tool's success is "requested", never "paid". Creating a pay link is the dispatch result. Payment is a destination fact that arrives later, from somewhere else.

Make retries idempotent, and then they're safe. Use the invoice id as the idempotency key (and ask for "reuse the open link for this order"). A lost response followed by a retry then returns the same link instead of minting a second one, so a lost response doesn't have to sit in UNKNOWN forever.

import json
from typing import Literal
from pydantic import BaseModel
from llama_index.core.tools import FunctionTool
from llama_index.tools.mcp import BasicMCPClient

INVOICES = {  # registered ahead of time, outside the model's reach
    "INV-1042": {"fiatAmount": 120, "fiatCurrency": "EUR",
                 "options": [{"currency": "Usdc", "network": "Base", "address": "0xYourOwnAddress"}]},
}

class PaymentRequested(BaseModel):
    outcome: Literal["requested"] = "requested"   # never "paid"
    invoice_id: str
    payment_request_id: str
    pay_link: str

client = BasicMCPClient("https://send21.io/mcp", headers={"Authorization": f"Bearer {KEY}"})

async def request_payment(invoice_id: str) -> dict:
    """Create (or re-fetch) the pay link for a registered invoice. Does NOT mean paid."""
    inv = INVOICES[invoice_id]
    r = await client.call_tool("create_payment_request", {
        **inv, "orderId": invoice_id,
        "idempotencyKey": invoice_id, "reuseOpenForOrder": True,
    })
    if r.is_error:
        raise RuntimeError(r.content[0].text)
    data = json.loads(r.content[0].text)
    return PaymentRequested(invoice_id=invoice_id, payment_request_id=data["id"],
                            pay_link="https://send21.io" + data["payPath"]).model_dump()

request_payment_tool = FunctionTool.from_defaults(async_fn=request_payment)

"Paid" comes from an independent observer, bound to this objective. Mark the invoice paid only from the signed confirmation webhook (or a status readback), and only if it binds to this invoice:

• orderId == invoice_id;
• amount and asset equal to the registry entry;
• the signature verifies over the raw body, and the delivery is no older than 300 s.

The sender suppresses duplicate deliveries on its side, but retries reuse the delivery id and can arrive out of order, so dedupe on that id in your own handler as well.

A near-miss amount is its own state. A payment a little short (exchange withdrawal fees) is neither VERIFIED nor failed. Park it in a review state that a person accepts or rejects with a stored reason, and don't let the agent "fix" it by requesting the difference.

I ran the keyless side today with llama-index-tools-mcp 0.6.0: it lists the tools, and McpToolSpec(allowed_tools=[...]) filters them if you'd rather expose raw MCP tools than a wrapper.

Disclosure: the MCP server in the example is ours, a non-custodial pay-link API where the payer signs in their own wallet. The example is a plain FunctionTool, not a LlamaIndex integration.

- send21