Track agent request costs and validate JSON before accepting results

An agent workflow can finish a request, produce valid JSON and still leave two questions unanswered: what did the attempt cost, and is its result safe to use?

Consider a developer building an extraction step that retrieves source material and writes a structured record. A client timeout may leave a tool request's billing state unresolved. A later response may parse correctly while changing a timeout value from 1500 milliseconds to 15000. Retrying can add attempts without resolving either question.

This tutorial separates request accounting from result acceptance. It includes two complete, independent Python examples: a tool-request ledger and a JSON acceptance function. Both use invented fixtures, standard-library code and no network access. They do not connect to each other or implement an end-to-end agent. No model, SDK, customer data or billing account is used.

Preserve the evidence for every attempt

Keep a stable internal attempt ID, its task ID and a reference to the observed evidence. Then record distinct decisions: the supplier response, the billing classification, the acceptance decision and any later write result. One successful retry should not overwrite an earlier attempt.

The first example keeps uncertain charges visible. The second shows why a completed response and a correctly shaped object still need an application-specific source check. Together they provide small controls to review before adding automatic retries or database writes.

Build a tool-request ledger

The dated supplier contract matters. As checked on 2 November 2026, DeepSeek's Search/Fetch guide lists $0.75 per 1,000 successful requests for each service. Billing requires HTTP 200 with a non-empty result; the number of returned results does not change the request price. The guide gives 1 November 2026 at 00:00 Beijing time, equivalent to 31 October at 16:00 UTC, as the billing start for its named server-side integrations through DeepSeek Harness, Claude Code and Codex.

That stated cutoff is now in the past. These are direct-provider documentation terms, not evidence of tool availability or matching charges through IteraCompute. The guide does not resolve start-versus-completion attribution for a request crossing the cutoff, so the fixture leaves that case unknown.

Our fields include endpoint, billing_path, start/end timestamps, http_status and normalized_payload. They are our normalized fixture fields, not a provider response schema. The documented_server_side label identifies the fixture's assumed path; it cannot infer a real integration's eligibility from a model ID.

Here, http_status means an observed supplier-tool response. A local timeout or proxy error is insufficient evidence to populate it. An observed failed response can be classified under the documented rule; an unobserved server outcome remains unknown.

Run the ledger example

Save this first block as tool_ledger.py in an empty scratch directory and run python3 tool_ledger.py with Python 3.9 or newer. It writes tool_ledger.json beside the script, overwriting a file of that name if present. It makes no network requests.

The twelve authored cases include one-result and fifty-result searches, requests on both sides of the cutoff, empty results, observed errors, a crossing request, a client timeout and an unknown billing path.

#!/usr/bin/env python3
"""Offline authored examples, NOT an API response parser or real invoice.
Only stdlib. No keys, network, tool execution, SDK or model calls.
The timestamps and normalized outcomes below are invented fixtures.
"""
import json
from datetime import datetime
from decimal import Decimal
from pathlib import Path

CUTOFF = datetime.fromisoformat('2026-10-31T16:00:00+00:00')
RATE = Decimal('0.75') / 1000
BEFORE = ('2026-10-31T15:59:00+00:00', '2026-10-31T15:59:01+00:00')
AFTER = ('2026-10-31T16:00:01+00:00', '2026-10-31T16:00:02+00:00')
CROSS = (BEFORE[0], AFTER[1])

def fixture(name, endpoint, path, interval, status, payload, expected):
    return dict(id=name, endpoint=endpoint, billing_path=path,
                start=interval[0], end=interval[1], http_status=status,
                normalized_payload=payload, expected_tool_usd=expected)

CASES = [
 fixture('direct_one_result_before', 'search', 'standalone', BEFORE, 200, {'results_count':1}, '0.00075'),
 fixture('direct_fifty_results_before', 'search', 'standalone', BEFORE, 200, {'results_count':50}, '0.00075'),
 fixture('documented_integration_before', 'search', 'documented_server_side', BEFORE, 200, {'results_count':3}, '0'),
 fixture('documented_integration_after', 'search', 'documented_server_side', AFTER, 200, {'results_count':3}, '0.00075'),
 fixture('fetch_content_after', 'fetch', 'documented_server_side', AFTER, 200, {'content':'authored example'}, '0.00075'),
 fixture('empty_search_after', 'search', 'documented_server_side', AFTER, 200, {'results_count':0}, '0'),
 fixture('empty_fetch_after', 'fetch', 'documented_server_side', AFTER, 200, {'content':''}, '0'),
 fixture('confirmed_http_504', 'fetch', 'documented_server_side', AFTER, 504, None, '0'),
 fixture('confirmed_http_429', 'search', 'standalone', AFTER, 429, None, '0'),
 fixture('crosses_cutoff', 'search', 'documented_server_side', CROSS, 200, {'results_count':1}, None),
 fixture('client_timeout_no_server_outcome', 'fetch', 'documented_server_side', AFTER, None, None, None),
 fixture('unknown_billing_path', 'search', 'unknown', AFTER, 200, {'results_count':1}, None),
]

def classify(row):
    status = row['http_status']
    if status is None:
        return None, 'server_outcome_unknown_client_timeout_is_not_confirmed_failure'
    if status != 200:
        return Decimal(0), 'confirmed_unsuccessful_request'
    payload = row['normalized_payload']
    if payload is None:
        return None, 'payload_not_observed'
    present = (payload['results_count'] > 0 if row['endpoint']=='search'
               else bool(payload['content']))
    if not present:
        return Decimal(0), 'confirmed_empty_result'
    path = row['billing_path']
    if path == 'standalone':
        return RATE, 'standalone_per_successful_request_not_per_result'
    if path != 'documented_server_side':
        return None, 'supplier_or_integration_mapping_unknown'
    start, end = datetime.fromisoformat(row['start']), datetime.fromisoformat(row['end'])
    assert start <= end
    if end < CUTOFF:
        return Decimal(0), 'documented_integration_before_billing_start'
    if start >= CUTOFF:
        return RATE, 'documented_integration_after_billing_start'
    return None, 'start_vs_completion_attribution_not_documented'

def main():
    rows=[]
    for row in CASES:
        cost, reason=classify(row)
        assert cost == (None if row['expected_tool_usd'] is None else Decimal(row['expected_tool_usd'])), row['id']
        rows.append({**row, 'calculated_tool_usd':None if cost is None else str(cost), 'reason':reason})
    known=sum((Decimal(r['calculated_tool_usd']) for r in rows if r['calculated_tool_usd'] is not None),Decimal(0))
    assert known == Decimal('0.00300')
    assert rows[0]['calculated_tool_usd'] == rows[1]['calculated_tool_usd']
    assert sum(r['calculated_tool_usd'] is None for r in rows)==3
    out={'kind':'synthetic_offline_controls_not_service_execution',
         'cutoff_utc':CUTOFF.isoformat(),'cutoff_beijing':'2026-11-01T00:00:00+08:00',
         'source':'https://api-docs.deepseek.com/guides/search/',
         'normalized_fields_are_ours_not_provider_schema':True,
         'cases':rows, 'case_controls_passed':len(rows),'additional_aggregate_controls_passed':3,
         'known_hypothetical_tool_subtotal_usd':str(known),
         'complete_tool_total_usd':None,'model_tokens_cost_usd':None,'all_in_task_usd':None,
         'actual_gateway_tool_support':None,'actual_invoice':None}
    target=Path(__file__).with_suffix('.json');target.write_text(json.dumps(out,ensure_ascii=False,indent=2)+'\n')
    print(json.dumps({'cases_passed':len(rows),'aggregate_controls_passed':3,'known_hypothetical_tool_subtotal_usd':str(known),'unknown_rows':3,'complete_total':None}))
if __name__=='__main__':main()

The expected output reports twelve case checks and three aggregate checks. Four priced requests contribute a known hypothetical subtotal of $0.00300, five cases contribute zero and three remain unresolved. The complete tool total, model-token cost and all-in task total stay null.

These fixtures check arithmetic and classification, not the provider's billing implementation. The known subtotal is neither a full invoice nor a spending ceiling. Keep a separate model-token ledger and reconcile unresolved tool rows against authorized supplier records before presenting a complete total.

A later successful retry can complete the task while the earlier request's charge remains unknown. Retain both attempt records. This example does not execute retries, collect model usage, account for taxes or fees, or compare suppliers.

Validate a result before accepting it

Once a response arrives, billing evidence cannot tell us whether the extracted data is correct. A separate acceptance function needs its own contract.

As checked on 2 November 2026, DeepSeek's JSON Output guide requires JSON-object response format, a prompt requesting JSON with an example, and a suitable output budget. It also documents space-only content even with a stop completion reason. Check blank content before parsing. This is documentation review, not a live API or gateway conformance test.

For the offline example, the source is exactly:

Ticket S-17: timeout_ms=1500; retry_limit=2.

The accepted object must contain exactly ticket_id, timeout_ms and retry_limit. The identifier must be a string, the timeout an integer from 1 to 60000 milliseconds, and the retry limit an integer from 0 to 3. These are invented application bounds. Units belong to the contract even though JSON itself does not encode them.

Separate completion, parsing, contract and source checks

First require stop and non-blank text. Treating every other completion reason as incomplete is this example's conservative choice, not a universal rule for all endpoints.

Next reject malformed JSON, duplicate keys and non-finite constants. Python's JSON decoder documentation describes the hooks used here. Duplicate-key rejection is stricter than the default last-value behavior; a later timeout field should not silently replace an earlier one in this workflow.

Then check the exact fields, types and ranges. Python's Boolean values pass an isinstance(value, int) check, so this contract uses type(value) is int. It also rejects floating-point values such as 1500.0. These are deliberate application policies.

Finally compare with trusted source evidence. For this single invented line, exact dictionary equality is a small oracle. It cannot judge arbitrary documents. Without an expected answer, a well-shaped record returns source_unverified rather than accepted. A production source check might require deterministic extraction, reviewed citations or a human decision.

Run the acceptance example

Save this second block separately as json_acceptance.py and run python3 json_acceptance.py with Python 3.9 or newer. It uses only the standard library and prints to standard output. The two scripts are independent; do not concatenate them into one file.

#!/usr/bin/env python3
"""Authored offline counterexamples only. No network, SDK, model or customer data.
Narrow acceptance contract, not a general JSON Schema validator or API adapter.
"""
import json
from collections import Counter

SOURCE = 'Ticket S-17: timeout_ms=1500; retry_limit=2.'
EXPECTED = {'ticket_id': 'S-17', 'timeout_ms': 1500, 'retry_limit': 2}

def reject_constant(value):
    raise ValueError('non-finite number')

def unique_object(pairs):
    obj = {}
    for key, value in pairs:
        if key in obj:
            raise ValueError('duplicate key')
        obj[key] = value
    return obj

def assess(content, finish_reason, expected=EXPECTED):
    # This is an application-specific, non-streaming content/finish pair.
    if finish_reason != 'stop':
        return 'incomplete'
    if not isinstance(content, str) or not content.strip():
        return 'blank_or_missing'
    try:
        value = json.loads(content, object_pairs_hook=unique_object,
                           parse_constant=reject_constant)
    except (ValueError, TypeError):
        return 'parse_policy_rejected'
    if (type(value) is not dict or set(value) != set(EXPECTED)
            or type(value.get('ticket_id')) is not str
            or type(value.get('timeout_ms')) is not int
            or not 1 <= value.get('timeout_ms', 0) <= 60000
            or type(value.get('retry_limit')) is not int
            or not 0 <= value.get('retry_limit', -1) <= 3):
        return 'contract_rejected'
    if expected is None:
        return 'source_unverified'
    return 'accepted' if value == expected else 'source_mismatch'

def main():
    good = json.dumps(EXPECTED)
    variants = [
        ('correct_authored_record', good, 'stop', 'accepted'),
        ('spaces_even_with_stop', '   ', 'stop', 'blank_or_missing'),
        ('empty_even_with_stop', '', 'stop', 'blank_or_missing'),
        ('missing_content', None, 'stop', 'blank_or_missing'),
        ('valid_object_but_length', good, 'length', 'incomplete'),
        ('missing_finish', good, None, 'incomplete'),
        ('truncated_json', '{"ticket_id":', 'stop', 'parse_policy_rejected'),
        ('array_root', '[' + good + ']', 'stop', 'contract_rejected'),
        ('extra_key', good[:-1] + ', "approved": true}', 'stop', 'contract_rejected'),
        ('missing_key', '{"ticket_id":"S-17","timeout_ms":1500}', 'stop', 'contract_rejected'),
        ('bool_not_int', '{"ticket_id":"S-17","timeout_ms":1500,"retry_limit":true}', 'stop', 'contract_rejected'),
        ('duplicate_key', '{"ticket_id":"S-17","timeout_ms":900,"timeout_ms":1500,"retry_limit":2}', 'stop', 'parse_policy_rejected'),
        ('nonfinite', '{"ticket_id":"S-17","timeout_ms":NaN,"retry_limit":2}', 'stop', 'parse_policy_rejected'),
        ('valid_shape_wrong_source', '{"ticket_id":"S-17","timeout_ms":15000,"retry_limit":2}', 'stop', 'source_mismatch'),
    ]
    rows = []
    for name, content, finish, expected_outcome in variants:
        actual = assess(content, finish)
        assert actual == expected_outcome, (name, actual, expected_outcome)
        rows.append(dict(id=name, content=content, finish_reason=finish,
                         expected=expected_outcome, actual=actual))
    # A well-formed record without a source oracle cannot be called verified.
    assert assess(good, 'stop', expected=None) == 'source_unverified'
    # Example retry grouping: both attempts remain, only the second is accepted.
    task_attempts = ['spaces_even_with_stop', 'correct_authored_record']
    selected = [r for r in rows if r['id'] in task_attempts]
    assert len(selected) == 2 and sum(r['actual'] == 'accepted' for r in selected) == 1
    print(json.dumps(dict(kind='authored_offline_contract_controls',
        source=SOURCE, expected_record=EXPECTED, cases=rows,
        controls_passed=len(rows)+2,
        counts=dict(Counter(r['actual'] for r in rows)),
        retry_example=dict(attempt_ids=task_attempts, attempts=2, accepted_tasks=1,
                           real_tokens=None, actual_bill=None),
        limits=['No model or API requested', 'No observed failure frequency',
                'No Gateway behavior verified', 'No production parser or streaming adapter',
                'Fixed source oracle covers only the authored record',
                'No automatic retries executed']), ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()

The expected output contains 16 passed checks: 14 authored records and two controls. The record outcomes are one accepted, three blank or missing, two incomplete, three parse-policy rejections, four contract rejections and one source mismatch. These deliberately selected inputs do not estimate any model's accuracy or production failure frequency.

The 15000-millisecond record is the key counterexample. Its syntax, field set, types and ranges pass. Only comparison with the trusted source rejects it. Tightening the range around this single example would not solve the general problem of plausible but incorrect values.

The no-oracle control keeps source_unverified distinct from both an accepted record and a confirmed extraction error. The retry control groups a blank first attempt and an accepted second attempt under one synthetic task. It preserves both records but performs no retry, supplies no usage data and proves nothing about whether repeating a real write is safe. Actual tokens and billing remain null.

Join the decisions without collapsing them

In a real application, connect evidence using stable task and attempt IDs rather than a single success flag. Keep these questions separate:

- Did the supplier return an observed result, or is the outcome unknown?
- Is the charge known, confirmed zero under the applicable contract, or unresolved?
- Did the content pass the application contract and trusted source check?
- Was a subsequent write actually committed, rejected or left uncertain?

For example, an accepted extraction does not establish that a database write succeeded. A client timeout does not establish that a supplier request was free. A successful retry does not settle the earlier attempt's invoice or eliminate duplicate-write risk.

Before connecting automatic writes, freeze a non-sensitive source example and a human-reviewed expected record. Include plausible values with the wrong unit or magnitude, a source that needs manual review, and retry cases with unresolved earlier outcomes. Record the actual model route, prompt, parser policy and application-contract version when live testing is separately undertaken.

The runtime owner should preserve response evidence. The integration owner should identify the supplier and billing path. The budget owner should reconcile charges in the correct units. The application owner should define acceptance and write policies, including a finite retry budget and protection against duplicate side effects.

Scope of these examples

The original tutorials' preparation records report local assertion checks. For this combined edition, both code blocks were preserved unchanged and checked for Python syntax; they were not rerun. The examples establish no model quality, production reliability, latency, retry savings, total task cost or supplier equivalence.

The ledger is not a production billing engine or provider parser. The acceptance function consumes a normalized content/completion pair, not HTTP responses or SSE events. It does not enforce input-size or nesting limits, implement a general JSON Schema validator, or provide a production security boundary. Its fixed fields and source oracle cover only the authored ticket. Neither script accesses a database or verifies IteraCompute tool support.

When preparing your actual model connection, start with the public IteraCompute API documentation, then verify the selected route's response and billing contract before applying these controls.