core

Cache your API calls with a single line of code. No mocks, no fixtures. Just faster, cleaner code.

Introduction

We often call APIs while prototyping and testing our code. A single API call (e.g. an Anthropic chat completion) can take 100’s of ms to run. This can really slow down development especially if our notebook contains many API calls 😞.

cachy caches API requests. It does this by saving the result of each API call to a local cachy.jsonl file. Before calling an API (e.g. OpenAI) it will check if the request already exists in cachy.jsonl. If it does it will return the cached result.

How does it work?

Under the hood popular SDK’s like OpenAI and Anthropic use httpx.Client and httpx.AsyncClient.

cachy patches the send method of both clients and injects a simple caching mechanism:

create a cache key from the request
if the key exists in cachy.jsonl return the cached response
if not, call the API and save the response to cachy.jsonl

import tempfile
from httpx import RequestNotRead
from fastcore.test import *

cachy.jsonl contains one API response per line.

Each line has the following format {"key": key, "response": response}

key: hash of the API request
response: the API response.

{
    "key": "afc2be0c", 
    "response": "{\"id\":\"msg_xxx\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"claude-sonnet-4-20250514\",\"content\":[{\"type\":\"text\",\"text\":\"Coordination.\"}],\"stop_reason\":\"end_turn\",\"stop_sequence\":null,\"usage\":{\"input_tokens\":16,\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0,\"cache_creation\":{\"ephemeral_5m_input_tokens\":0,\"ephemeral_1h_input_tokens\":0},\"output_tokens\":6,\"service_tier\":\"standard\"}}"
}

Patching `httpx`

Patching a method is very straightforward.

In our case we want to patch httpx.Client.send and httpx.AsyncClient.send.

These methods are called when running httpx.get, httpx.post, etc.

In the example below we use @patch from fastcore to print calling an API when httpx.Client.send is run.

@patch
def send(self:httpx.Client, r, **kwargs):
    print('calling an API')
    return self._orig_send(r, **kwargs)

Cache Filtering

Now, let’s build up our caching logic piece-by-piece.

The first thing we need to do is ensure that our caching logic only runs on specific urls.

For now, let’s only cache API calls made to popular LLM providers like OpenAI, Anthropic, Google and DeepSeek. We can make this fully customizable later.

Exported source

doms = ("chatgpt.com", "api.openai.com", "api.anthropic.com", "generativelanguage.googleapis.com", "api.deepseek.com",
    'api.fireworks.ai', 'openrouter.ai', 'api.groq.com', 'api.together.xyz', 'api.mistral.ai', 'api.x.ai', 'api.moonshot.ai')

Exported source

def _should_cache(url, doms): return any(dom in str(url) for dom in doms)

We could then use _should_cache like this.

@patch
def send(self:httpx._client.Client, r, **kwargs):
    if not _should_cache(r.url, doms): return self._orig_send(r, **kwargs)
    # insert caching logic
    ...

Cache Key

The next thing we need to do is figure out if a response for the request r already exists in our cache.

Recall that each line in cachy.jsonl has the following format {"key": key, "response": response}.

Our key needs to be unique and deterministic. One way to do this is to concatenate the request URL and content, then generate a hash from the result.

def _key(r): return hashlib.sha256(str(r.url.copy_remove_param('key')).encode() + r.content).hexdigest()[:8]

Some clients pass the Gemini API key as a query param so that’s why we strip the key param from the url.

Let’s test this out.

r1 = httpx.Request('POST', 'https://api.openai.com/v1/chat/completions', content=b'some content')
r1

<Request('POST', 'https://api.openai.com/v1/chat/completions')>

_key(r1)

'2d135d43'

If we run it again we should get the same key.

_key(r1)

'2d135d43'

Let’s modify the url and confirm we get a different key.

_key(httpx.Request('POST', 'https://api.anthropic.com/v1/messages', content=b'some content'))

'8a99b0a9'

Great. Let’s update our patch.

@patch
def send(self:httpx._client.Client, r, **kwargs):
    if not _should_cache(r.url, doms): return self._orig_send(r, **kwargs)
    key = _key(r)
    # if cache hit return the response
    # else run the request, write to response the cache and return it
    ...

Cache Reads/Writes

Now let’s add some methods that will read from and write to cachy.jsonl.

Exported source

def _cache(key, cfp):
    if not Path(cfp).exists(): return None
    with open(cfp) as f:
        line = first(f, lambda l: json.loads(l)["key"] == key)
        return json.loads(line) if line else None

Responses can carry very redundant payloads: OpenAI’s Responses API echoes the request’s configuration – instructions, tools, sampling settings – back inside the response resource, and a streamed response repeats the whole resource in every lifecycle event, so one cached line can carry a host’s entire system prompt three times over (in solveit’s cache that echo was ~90% of the file). resp_keep fixes this by saying what to keep: any JSON object whose object field names an entry is pruned to the listed fields before storing, whether it arrives as a plain body or inside SSE data: lines. Because the pruning parses the JSON, it is immune to formatting (compact vs pretty-printed) and field order, and new echoed fields are dropped automatically – keep-lists fail safe where strip-lists accumulate omissions. Set resp_keep = None to store byte-exact responses. resp_norm_pats handles the app-specific remainder: (pattern, replacement) pairs applied to the stored text, empty by default. Neither affects the key, so cache hits are unchanged.

_write_cache keeps the file sorted by (response, key) rather than appending. Append order is completion order, which is nondeterministic under concurrency, so re-recording identical calls used to reshuffle the whole file; sorted, the order is stable, writing an existing key replaces its line instead of duplicating it, and near-identical responses sit next to each other – so a re-recorded entry lands beside the stale one it supersedes and diffs pair them up. Sorting by key alone wouldn’t do that last part: the key hashes the request, so a prompt tweak moves every affected line somewhere unrelated. The rewrite lands via atomic_save (write-aside, then rename), because parallel test processes share one cache file: _cache readers take no lock, and an atomic replace means they always see a complete file, never a half-written one.

Exported source

def _write_cache(key, content, cfp, hdrs, status_code=200, binary=False):
    if not binary: content = _norm_resp(content)
    with open(cfp, "a+") as lk:
        fcntl.flock(lk, fcntl.LOCK_EX)
        lk.seek(0)
        recs = {(d:=json.loads(l))['key']: d for l in lk if l.strip()}
        recs[key] = dict(key=key, response=content, headers=hdrs, status_code=status_code, binary=binary)
        with atomic_save(Path(cfp), mode='w') as f:
            for d in sorted(recs.values(), key=lambda d: (d['response'], d['key'])): f.write(json.dumps(d)+"\n")

Writes land sorted and replace rather than duplicate:

tcfp = Path(tempfile.mkdtemp())/'cachy.jsonl'
_write_cache('k2', 'resp B', tcfp, None)
_write_cache('k1', 'resp A', tcfp, None)
_write_cache('k2', 'resp B v2', tcfp, None)
test_eq([json.loads(l)['key'] for l in tcfp.read_text().splitlines()], ['k1','k2'])
print(tcfp.read_text())

Pruning applies to whole-body JSON and to each SSE event alike; delta events carry no object: response resource, so streamed content passes through untouched. resp_norm_pats remains for content the keep-list can’t describe – an app-specific token to stub, say. Both rewrite the stored bytes only, and a JSON body that nothing prunes is stored byte-exact – re-serialization happens only where a keep-list actually removed something, so ordinary JSON responses replay identically:

sse = '''event: response.created
data: {"type":"response.created","response":{"id":"resp_1","object":"response","status":"in_progress","model":"gpt-x","instructions":"an enormous system prompt","tools":[{"name":"big_schema"}],"top_p":0.98}}

event: response.output_text.delta
data: {"type":"response.output_text.delta","delta":"hi"}
'''
_write_cache('k3', sse, tcfp, None)
stored = _cache('k3', tcfp)['response']
assert 'enormous' not in stored and 'big_schema' not in stored
assert 'data: {"type":"response.output_text.delta","delta":"hi"}' in stored
resp_norm_pats.append((r'sk-[\w-]+', 'sk-XXX'))
_write_cache('k4', 'used key sk-abc123', tcfp, None)
resp_norm_pats.clear()
test_eq(_cache('k4', tcfp)['response'], 'used key sk-XXX')
_write_cache('k5', '{\n  "pretty":  "spacing kept"\n}', tcfp, None)
test_eq(_cache('k5', tcfp)['response'], '{\n  "pretty":  "spacing kept"\n}')
print(stored)

Multipart Requests

_key will throw the following error for multipart requests (e.g. file uploads).

RequestNotRead: Attempted to access streaming request content, without having calledread().

rfu = httpx.Request('POST', 'https://api.openai.com/v1/chat/completions', files={"file": ("test.txt", b"hello")})
rfu

<Request('POST', 'https://api.openai.com/v1/chat/completions')>

with expect_fail(RequestNotRead): _key(rfu)

rfu.read(); _key(rfu);

Each part of a multipart request is separated by a delimiter called a boundary with this structure --b{RANDOM_ID}. Here’s an example for rfu.

b'--f9ee33966b45cc8c80952bb57cc728c4\r\nContent-Disposition: form-data; name="file"; filename="test.txt"\r\nContent-Type: text/plain\r\n\r\nhello\r\n--f9ee33966b45cc8c80952bb57cc728c4--\r\n'

As the boundary is a random id, two identical multipart requests will produce different boundaries. As the boundary is part of the request content, _key will generate different keys leading to cache misses 😞.

Let’s create a helper method _content that will extract content from any request and remove the non-deterministic boundary.

rfu = httpx.Request('POST', 'https://api.openai.com/v1/chat/completions', files={"file": ("test.txt", b"hello")})
rfu

<Request('POST', 'https://api.openai.com/v1/chat/completions')>

_content(rfu)

b'--cachy-boundary\r\nContent-Disposition: form-data; name="file"; filename="test.txt"\r\nContent-Type: text/plain\r\n\r\nhello\r\n--cachy-boundary--\r\n'

def _key(r): return hashlib.sha256(str(r.url.copy_remove_param('key')).encode() + _content(r)).hexdigest()[:8]

Let’s confirm that running _key multiple times on the same multipart request now returns the same key.

_key(rfu), _key(rfu)

('9ae79ac5', '9ae79ac5')

Streaming

Let’s add support for streaming.

First let’s include an is_stream bool in our hash so that a non-streamed request will generate a different key to the same request when streamed.

norm_pats is public: append your own (pattern, replacement) pairs (strings or compiled) to normalize app-specific ephemeral content out of the cache key. For example solveit embeds random message ids in its requests, so its tests do norm_pats.append((r'\b_[0-9a-f]{8}\b', '_MSGID')) – re-running a test with fresh ids then still hits the cache. Only the key is affected; responses are stored as-is (see resp_norm_pats above for the stored side).

c1 = b'{"model":"kimi-k2.5","messages":[{"content":"Say hello in French","role":"user"}],"max_tokens":64,"temperature":1.0}'
c2 = b'{"messages":[{"role":"user","content":"Say hello in French"}],"model":"kimi-k2.5","max_tokens":64,"temperature":1.0}'
test_eq(_norm_content(SimpleNamespace(headers={'content-type': 'application/json'}, content=c1, _content='done reading')), 
        _norm_content(SimpleNamespace(headers={'content-type': 'application/json'}, content=c2, _content='done reading')))

In the patch we need to consume the entire stream before writing it to the cache.

Some libraries use requests rather than httpx – e.g. cloudscraper, which solveit’s read_url uses to fetch pages. Patching requests.adapters.HTTPAdapter.send catches every requests.Session, including subclasses like cloudscraper’s (whose adapter inherits send).

@patch
def send(self:httpx._client.Client, request, **kwargs):
    return _send('cachy.json', doms, self, request, **kwargs)

doms = doms + ('httpbingo.org',)
origdir = os.getcwd()
os.chdir(tempfile.mkdtemp())
r1 = httpx.post('https://httpbingo.org/post', json={'a':1})
r1.json()['headers']

{'Accept': ['*/*'],
 'Accept-Encoding': ['gzip, deflate, br, zstd'],
 'Connection': ['keep-alive'],
 'Content-Length': ['7'],
 'Content-Type': ['application/json'],
 'Host': ['httpbingo.org'],
 'User-Agent': ['python-httpx/0.28.1'],
 'Via': ['1.1 fly.io, 1.1 fly.io'],
 'X-Forwarded-For': ['159.196.29.66, 66.241.125.232'],
 'X-Forwarded-Port': ['443'],
 'X-Forwarded-Proto': ['https'],
 'X-Forwarded-Ssl': ['on'],
 'X-Request-Start': ['t=1786014913976728']}

r2 = httpx.post('https://httpbingo.org/post', json={'a':1})
assert r2.text==r1.text

`enable_cachy`

To make cachy as user friendly as possible let’s make it so that we can apply our patch by running a single method at the top of our notebook.

from cachy import enable_cachy

enable_cachy()

def enable_cachy(cache_dir=None, doms=doms):
    cfp = Path(cache_dir or find_file_parents("pyproject.toml") or ".") / "cachy.jsonl"
    cfp.touch(exist_ok=True)
    _apply_patch(cfp, doms)

Async

Now let’s add support for async requests.

enable_cachy

def enable_cachy(
    cache_dir:NoneType=None,
    doms:tuple=('chatgpt.com', 'api.openai.com', 'api.anthropic.com', 'generativelanguage.googleapis.com', 'api.deepseek.com', 'api.fireworks.ai', 'openrouter.ai', 'api.groq.com', 'api.together.xyz', 'api.mistral.ai', 'api.x.ai', 'api.moonshot.ai'),
    hdrs:NoneType=None, debug:bool=False
):

Call self as a function.

disable_cachy

def disable_cachy():

Call self as a function.

enable_cachy(debug=True)
r1 = httpx.post('https://httpbingo.org/post', json={'a':1})
r2 = httpx.post('https://httpbingo.org/post', json={'a':1})
test_eq(r1.text, r2.text)

/private/var/folders/51/b2_szf2945n072c0vj2cyty40000gn/T/cachy.jsonl
🟢 HIT 12fdbd07
b'{"a":1}'
🟢 HIT 12fdbd07
b'{"a":1}'

async with AsyncClient() as c:
    r1 = await c.post('https://httpbingo.org/post', json={'a':2})
    r2 = await c.post('https://httpbingo.org/post', json={'a':2})
test_eq(r1.text, r2.text)

🟢 HIT 1be4aa2f
b'{"a":2}'
🟢 HIT 1be4aa2f
b'{"a":2}'

with httpx.stream('POST', 'https://httpbingo.org/post', json={'a':3}) as r1: t1 = r1.read()
with httpx.stream('POST', 'https://httpbingo.org/post', json={'a':3}) as r2: t2 = r2.read()
test_eq(t1, t2)

🟢 HIT 331a6125
b'{"a":3}'
🟢 HIT 331a6125
b'{"a":3}'

disable_cachy()

Tests

Let’s test enable_cachy on the OpenAI and Anthropic SDKs for the scenarios below:

sync requests with(out) streaming
async requests with(out) streaming

Add some helper functions.

class mods: ant="claude-sonnet-5"; oai="gpt-4o"

def mk_msgs(m): return [{"role": "user", "content": f"write 1 word about {m}"}]

enable_cachy(debug=True)

OpenAI

from openai import OpenAI

cli = OpenAI()

r = cli.responses.create(model=mods.oai, input=mk_msgs("openai sync"))
r

🟢 HIT 57d94e97
b'{"input":[{"role":"user","content":"write 1 word about openai sync"}],"model":"gpt-4o"}'

Response(id='resp_0698fd4066445997006a746409f670819aa420d72f27e6cf90', created_at=1786012682.0, error=None, incomplete_details=None, instructions=None, metadata={}, model='gpt-4o-2024-08-06', object='response', output=[ResponseOutputMessage(id='msg_0698fd4066445997006a74640b4cb8819ab9c33ec83c08f1c0', content=[ResponseOutputText(annotations=[], text='Collaboration', type='output_text', logprobs=[])], role='assistant', status='completed', type='message', phase=None)], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, completed_at=1786012683.0, conversation=None, max_output_tokens=None, max_tool_calls=None, moderation=None, previous_response_id=None, prompt=None, prompt_cache_key=None, prompt_cache_options=None, prompt_cache_retention='in_memory', reasoning=Reasoning(context=None, effort=None, generate_summary=None, mode=None, summary=None), safety_identifier=None, service_tier='default', status='completed', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=ResponseUsage(input_tokens=15, input_tokens_details=InputTokensDetails(cache_write_tokens=0, cached_tokens=0), output_tokens=3, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=18), user=None, billing={'payer': 'developer'}, frequency_penalty=0.0, presence_penalty=0.0, store=True, tool_usage={'image_gen': {'input_tokens': 0, 'input_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'output_tokens': 0, 'output_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'total_tokens': 0}, 'web_search': {'num_requests': 0}})

r = cli.responses.create(model=mods.oai, input=mk_msgs("openai sync"))
r

🟢 HIT 57d94e97
b'{"input":[{"role":"user","content":"write 1 word about openai sync"}],"model":"gpt-4o"}'

Response(id='resp_0698fd4066445997006a746409f670819aa420d72f27e6cf90', created_at=1786012682.0, error=None, incomplete_details=None, instructions=None, metadata={}, model='gpt-4o-2024-08-06', object='response', output=[ResponseOutputMessage(id='msg_0698fd4066445997006a74640b4cb8819ab9c33ec83c08f1c0', content=[ResponseOutputText(annotations=[], text='Collaboration', type='output_text', logprobs=[])], role='assistant', status='completed', type='message', phase=None)], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, completed_at=1786012683.0, conversation=None, max_output_tokens=None, max_tool_calls=None, moderation=None, previous_response_id=None, prompt=None, prompt_cache_key=None, prompt_cache_options=None, prompt_cache_retention='in_memory', reasoning=Reasoning(context=None, effort=None, generate_summary=None, mode=None, summary=None), safety_identifier=None, service_tier='default', status='completed', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=ResponseUsage(input_tokens=15, input_tokens_details=InputTokensDetails(cache_write_tokens=0, cached_tokens=0), output_tokens=3, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=18), user=None, billing={'payer': 'developer'}, frequency_penalty=0.0, presence_penalty=0.0, store=True, tool_usage={'image_gen': {'input_tokens': 0, 'input_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'output_tokens': 0, 'output_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'total_tokens': 0}, 'web_search': {'num_requests': 0}})

Let’s test streaming.

r = cli.responses.create(model=mods.oai, input=mk_msgs("openai sync streaming"), stream=True)
for ch in r: print(str(ch)[:60])

🟢 HIT 24e0bc9b
b'{"input":[{"role":"user","content":"write 1 word about openai sync streaming"}],"model":"gpt-4o","stream":true}'
ResponseCreatedEvent(response=Response(id='resp_083278a55a8e
ResponseInProgressEvent(response=Response(id='resp_083278a55
ResponseOutputItemAddedEvent(item=ResponseOutputMessage(id='
ResponseContentPartAddedEvent(content_index=0, item_id='msg_
ResponseTextDeltaEvent(content_index=0, delta='Innov', item_
ResponseTextDeltaEvent(content_index=0, delta='ative', item_
ResponseTextDoneEvent(content_index=0, item_id='msg_083278a5
ResponseContentPartDoneEvent(content_index=0, item_id='msg_0
ResponseOutputItemDoneEvent(item=ResponseOutputMessage(id='m
ResponseCompletedEvent(response=Response(id='resp_083278a55a

r = cli.responses.create(model=mods.oai, input=mk_msgs("openai sync streaming"), stream=True)
for ch in r: print(str(ch)[:60])

🟢 HIT 24e0bc9b
b'{"input":[{"role":"user","content":"write 1 word about openai sync streaming"}],"model":"gpt-4o","stream":true}'
ResponseCreatedEvent(response=Response(id='resp_083278a55a8e
ResponseInProgressEvent(response=Response(id='resp_083278a55
ResponseOutputItemAddedEvent(item=ResponseOutputMessage(id='
ResponseContentPartAddedEvent(content_index=0, item_id='msg_
ResponseTextDeltaEvent(content_index=0, delta='Innov', item_
ResponseTextDeltaEvent(content_index=0, delta='ative', item_
ResponseTextDoneEvent(content_index=0, item_id='msg_083278a5
ResponseContentPartDoneEvent(content_index=0, item_id='msg_0
ResponseOutputItemDoneEvent(item=ResponseOutputMessage(id='m
ResponseCompletedEvent(response=Response(id='resp_083278a55a

Let’s test async.

from openai import AsyncOpenAI

cli = AsyncOpenAI()

r = await cli.responses.create(model=mods.oai, input=mk_msgs("openai async"))
r

🟢 HIT b823478a
b'{"input":[{"role":"user","content":"write 1 word about openai async"}],"model":"gpt-4o"}'

Response(id='resp_072e22650967c250006a74640d6980819ba84b2fa564f4bace', created_at=1786012685.0, error=None, incomplete_details=None, instructions=None, metadata={}, model='gpt-4o-2024-08-06', object='response', output=[ResponseOutputMessage(id='msg_072e22650967c250006a74640f9298819b841612d691525ed6', content=[ResponseOutputText(annotations=[], text='Innovative', type='output_text', logprobs=[])], role='assistant', status='completed', type='message', phase=None)], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, completed_at=1786012687.0, conversation=None, max_output_tokens=None, max_tool_calls=None, moderation=None, previous_response_id=None, prompt=None, prompt_cache_key=None, prompt_cache_options=None, prompt_cache_retention='in_memory', reasoning=Reasoning(context=None, effort=None, generate_summary=None, mode=None, summary=None), safety_identifier=None, service_tier='default', status='completed', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=ResponseUsage(input_tokens=15, input_tokens_details=InputTokensDetails(cache_write_tokens=0, cached_tokens=0), output_tokens=3, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=18), user=None, billing={'payer': 'developer'}, frequency_penalty=0.0, presence_penalty=0.0, store=True, tool_usage={'image_gen': {'input_tokens': 0, 'input_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'output_tokens': 0, 'output_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'total_tokens': 0}, 'web_search': {'num_requests': 0}})

r = await cli.responses.create(model=mods.oai, input=mk_msgs("openai async"))
r

🟢 HIT b823478a
b'{"input":[{"role":"user","content":"write 1 word about openai async"}],"model":"gpt-4o"}'

Response(id='resp_072e22650967c250006a74640d6980819ba84b2fa564f4bace', created_at=1786012685.0, error=None, incomplete_details=None, instructions=None, metadata={}, model='gpt-4o-2024-08-06', object='response', output=[ResponseOutputMessage(id='msg_072e22650967c250006a74640f9298819b841612d691525ed6', content=[ResponseOutputText(annotations=[], text='Innovative', type='output_text', logprobs=[])], role='assistant', status='completed', type='message', phase=None)], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, completed_at=1786012687.0, conversation=None, max_output_tokens=None, max_tool_calls=None, moderation=None, previous_response_id=None, prompt=None, prompt_cache_key=None, prompt_cache_options=None, prompt_cache_retention='in_memory', reasoning=Reasoning(context=None, effort=None, generate_summary=None, mode=None, summary=None), safety_identifier=None, service_tier='default', status='completed', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=ResponseUsage(input_tokens=15, input_tokens_details=InputTokensDetails(cache_write_tokens=0, cached_tokens=0), output_tokens=3, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=18), user=None, billing={'payer': 'developer'}, frequency_penalty=0.0, presence_penalty=0.0, store=True, tool_usage={'image_gen': {'input_tokens': 0, 'input_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'output_tokens': 0, 'output_tokens_details': {'image_tokens': 0, 'text_tokens': 0}, 'total_tokens': 0}, 'web_search': {'num_requests': 0}})

Let’s test async streaming.

r = await cli.responses.create(model=mods.oai, input=mk_msgs("openai async streaming"), stream=True)
async for ch in r: print(str(ch)[:60])

🟢 HIT b3ff4fe2
b'{"input":[{"role":"user","content":"write 1 word about openai async streaming"}],"model":"gpt-4o","stream":true}'
ResponseCreatedEvent(response=Response(id='resp_0f35c1ced55e
ResponseInProgressEvent(response=Response(id='resp_0f35c1ced
ResponseOutputItemAddedEvent(item=ResponseOutputMessage(id='
ResponseContentPartAddedEvent(content_index=0, item_id='msg_
ResponseTextDeltaEvent(content_index=0, delta='Innov', item_
ResponseTextDeltaEvent(content_index=0, delta='ative', item_
ResponseTextDoneEvent(content_index=0, item_id='msg_0f35c1ce
ResponseContentPartDoneEvent(content_index=0, item_id='msg_0
ResponseOutputItemDoneEvent(item=ResponseOutputMessage(id='m
ResponseCompletedEvent(response=Response(id='resp_0f35c1ced5

r = await cli.responses.create(model=mods.oai, input=mk_msgs("openai async streaming"), stream=True)
async for ch in r: print(str(ch)[:60])

🟢 HIT b3ff4fe2
b'{"input":[{"role":"user","content":"write 1 word about openai async streaming"}],"model":"gpt-4o","stream":true}'
ResponseCreatedEvent(response=Response(id='resp_0f35c1ced55e
ResponseInProgressEvent(response=Response(id='resp_0f35c1ced
ResponseOutputItemAddedEvent(item=ResponseOutputMessage(id='
ResponseContentPartAddedEvent(content_index=0, item_id='msg_
ResponseTextDeltaEvent(content_index=0, delta='Innov', item_
ResponseTextDeltaEvent(content_index=0, delta='ative', item_
ResponseTextDoneEvent(content_index=0, item_id='msg_0f35c1ce
ResponseContentPartDoneEvent(content_index=0, item_id='msg_0
ResponseOutputItemDoneEvent(item=ResponseOutputMessage(id='m
ResponseCompletedEvent(response=Response(id='resp_0f35c1ced5

Anthropic

from anthropic import Anthropic

cli = Anthropic()

r = cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant sync"))
r

🟢 HIT 2e3c4bee
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant sync"}],"model":"claude-sonnet-5"}'

Message(id='msg_011CdmMH1fopuA8wcTAR2967', container=None, content=[TextBlock(citations=None, text='**Pheromones**\n\nThis single word captures how ants synchronize behavior—using chemical trails to coordinate movement, foraging, and collective decision-making across the colony.', type='text')], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason='end_turn', stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=16, output_tokens=56, output_tokens_details=OutputTokensDetails(thinking_tokens=0), server_tool_use=None, service_tier='standard'))

r = cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant sync"))
r

🟢 HIT 2e3c4bee
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant sync"}],"model":"claude-sonnet-5"}'

Message(id='msg_011CdmMH1fopuA8wcTAR2967', container=None, content=[TextBlock(citations=None, text='**Pheromones**\n\nThis single word captures how ants synchronize behavior—using chemical trails to coordinate movement, foraging, and collective decision-making across the colony.', type='text')], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason='end_turn', stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=16, output_tokens=56, output_tokens_details=OutputTokensDetails(thinking_tokens=0), server_tool_use=None, service_tier='standard'))

Let’s test streaming.

r = cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant sync streaming"), stream=True)
for ch in r: print(ch)

🟢 HIT bd7e38a1
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant sync streaming"}],"model":"claude-sonnet-5","stream":true}'
RawMessageStartEvent(message=Message(id='msg_011CdmMHC1QVjJmCRX8r7SDU', container=None, content=[], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason=None, stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=19, output_tokens=7, output_tokens_details=None, server_tool_use=None, service_tier='standard')), type='message_start')
RawContentBlockStartEvent(content_block=ThinkingBlock(signature='', thinking='', type='thinking'), index=0, type='content_block_start')
RawContentBlockDeltaEvent(delta=ThinkingDelta(thinking='', type='thinking_delta'), index=0, type='content_block_delta')
RawContentBlockDeltaEvent(delta=SignatureDelta(signature='ErwKCokBCBAYAipAlLKje2iCLzXKl9YaSa0wYBIExpnfXRekzdcOMnjfunczcV5Cp5mgIc7XtWw8Mk1J+QG+KxV2wE4NaxtLMtEzQzIPY2xhdWRlLXNvbm5ldC01OABCCHRoaW5raW5nWiQ4ODk4YTFkYy0yMTNkLTRhNmYtOTljYi03ZTBlNTUzZDc0NWISDCUkX89g7OKpqp2O2BoMxGiDbKpbg6TusXxBIjBK5V0gbt+ZKGut6wWOECgAvIioBVqurCAtGcx6Mpa50dkgSqGDJAumHQnvjDmT5S8q3whVmPG0IbRFQB8Iw0dyim+Ybw8PVGr2YXiZJQL9UouJjVgwaLecw3nWmQUnLBI8jSQNKLn6XhdWJwxma70sSfcSHVxAXAukIOQC1SHlILeG9eO7XaTfr5MyxBABvahj4Re+nvGZl6Oy5fRB9NFSTsHQwwgO/2eLFg75ndnxglE1Svgztue/ecWf4lcn6X15W5sdCd7HxK86BD3rX8z8ngt0yQawqgO+2i3POLtjBysFDYxV5UU+TVEMPfi36aehmoJOvvLiTRoCwwWUK9GIQvqzc2ucyEHxScUdVJ5yTkA2O5ZBgilzvpcGKrqktMy+hHdZPhjJBQi63E+OQe25j8MxLnxbyt0bpyiuPQXQbOhoJo20xjk9BMGUAB/bUOTu+rRuKPmX9xaa4t+vVhvDxfbA/NF1r8gXS7QIJzxSEQq0KwGsDMOFynwI939ZdEk8HMvk6VCLwk7PXhWdLmw4+Jge8BLHKDuZevsz60pg9LTTIR2FqMq3iCMapLCQedo9UpJcJQT4jGgTn6lFbLov8JQM6bFAN1ZyEJvbVhr1pEHnRZXBJ1jXUK1h0bt+5VO642fbn0c5Z9EAeYCe0zOT/B3uX6NJFI2bGRGLuq+yjFeE3sJ8FHUfXPLi/lTLbL1CvCji2hEvJd1rruC723znyuvgv/fFiXpCwSqYG8b5nXkxhXqKBKTWhmgzntlT9bVw6OX4FJWNZeDzoEyWmBbd2owNkhZQgWvqJUViK9k1vTkZeNjvBSX5c/PHdKmLvJY2/l/yZFKnM7GJ71Rh+rLR0/pXTsILY8EAUw0LcoTA42dGEw15HQLRGoIdbhY4245M19ohrMuBxKXdPLXTN079rVd7pTF/o0KMRuKPOxkKgbe8Mnnk7V4qqTenQlQMiZIGmJy+WQyo2eagzHAgE/fXyIyJKhQSV/HS4Luf+k/D/fuiaXQ5qN9hp/U8k+1xX16bvROfLWeNcV0rVWblS0S+fxmH8kB5lGxCRZretcrcEMFzShEoAzFPrssGhVuOZJMvVjJ+PYKpCyA7gX3LWaNq+o7ybmoOj4KIt/9J2KNsdvV0bdwlGOnfaGO178slBMsNAVqjLoqN2ONbEKkFpVwuanM9g5mgfnjVEUG5BtZCKM5BlGOgPlwpeb7b0f7CaYunoAFxel5xZEK9RTIzIoOxdiUXR6kiFQgGHMpYvseci2Aoa/8gYf5JnJt+grcslC+UvDPyb62Qjox3ZYlsBSjkLneXiYHmH93k5CNAy1i+2W6dtHXKeuVVWYBEdbvJKRl5bByYvTKWWQ4eB+oIWv/MrDVLHIi3+aJDErbByq5RCQ4VsbUPt4gAUZ/7mvTSgi1ZNlh8lflWcpTBtPyyc4XkuFWRnOoQzpSiomVJZ4y2TjO46oVZPGi7drzxFoWwrWE4QLobCpKyyTghEQRlio/HUFoG9b5FUoAhPeaBXVcuQafMv0iKNFUTq/BpPQKVSMmqvd1lkkfVeBH+3kMzTVddGs4YAQ==', type='signature_delta'), index=0, type='content_block_delta')
RawContentBlockStopEvent(index=0, type='content_block_stop')
RawContentBlockStartEvent(content_block=TextBlock(citations=None, text='', type='text'), index=1, type='content_block_start')
RawContentBlockDeltaEvent(delta=TextDelta(text='**Synchronization**', type='text_delta'), index=1, type='content_block_delta')
RawContentBlockStopEvent(index=1, type='content_block_stop')
RawMessageDeltaEvent(delta=Delta(container=None, stop_details=None, stop_reason='end_turn', stop_sequence=None), type='message_delta', usage=MessageDeltaUsage(cache_creation_input_tokens=0, cache_read_input_tokens=0, input_tokens=19, output_tokens=400, output_tokens_details=OutputTokensDetails(thinking_tokens=390), server_tool_use=None))
RawMessageStopEvent(type='message_stop')

r = cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant sync streaming"), stream=True)
for ch in r: print(ch)

🟢 HIT bd7e38a1
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant sync streaming"}],"model":"claude-sonnet-5","stream":true}'
RawMessageStartEvent(message=Message(id='msg_011CdmMHC1QVjJmCRX8r7SDU', container=None, content=[], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason=None, stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=19, output_tokens=7, output_tokens_details=None, server_tool_use=None, service_tier='standard')), type='message_start')
RawContentBlockStartEvent(content_block=ThinkingBlock(signature='', thinking='', type='thinking'), index=0, type='content_block_start')
RawContentBlockDeltaEvent(delta=ThinkingDelta(thinking='', type='thinking_delta'), index=0, type='content_block_delta')
RawContentBlockDeltaEvent(delta=SignatureDelta(signature='ErwKCokBCBAYAipAlLKje2iCLzXKl9YaSa0wYBIExpnfXRekzdcOMnjfunczcV5Cp5mgIc7XtWw8Mk1J+QG+KxV2wE4NaxtLMtEzQzIPY2xhdWRlLXNvbm5ldC01OABCCHRoaW5raW5nWiQ4ODk4YTFkYy0yMTNkLTRhNmYtOTljYi03ZTBlNTUzZDc0NWISDCUkX89g7OKpqp2O2BoMxGiDbKpbg6TusXxBIjBK5V0gbt+ZKGut6wWOECgAvIioBVqurCAtGcx6Mpa50dkgSqGDJAumHQnvjDmT5S8q3whVmPG0IbRFQB8Iw0dyim+Ybw8PVGr2YXiZJQL9UouJjVgwaLecw3nWmQUnLBI8jSQNKLn6XhdWJwxma70sSfcSHVxAXAukIOQC1SHlILeG9eO7XaTfr5MyxBABvahj4Re+nvGZl6Oy5fRB9NFSTsHQwwgO/2eLFg75ndnxglE1Svgztue/ecWf4lcn6X15W5sdCd7HxK86BD3rX8z8ngt0yQawqgO+2i3POLtjBysFDYxV5UU+TVEMPfi36aehmoJOvvLiTRoCwwWUK9GIQvqzc2ucyEHxScUdVJ5yTkA2O5ZBgilzvpcGKrqktMy+hHdZPhjJBQi63E+OQe25j8MxLnxbyt0bpyiuPQXQbOhoJo20xjk9BMGUAB/bUOTu+rRuKPmX9xaa4t+vVhvDxfbA/NF1r8gXS7QIJzxSEQq0KwGsDMOFynwI939ZdEk8HMvk6VCLwk7PXhWdLmw4+Jge8BLHKDuZevsz60pg9LTTIR2FqMq3iCMapLCQedo9UpJcJQT4jGgTn6lFbLov8JQM6bFAN1ZyEJvbVhr1pEHnRZXBJ1jXUK1h0bt+5VO642fbn0c5Z9EAeYCe0zOT/B3uX6NJFI2bGRGLuq+yjFeE3sJ8FHUfXPLi/lTLbL1CvCji2hEvJd1rruC723znyuvgv/fFiXpCwSqYG8b5nXkxhXqKBKTWhmgzntlT9bVw6OX4FJWNZeDzoEyWmBbd2owNkhZQgWvqJUViK9k1vTkZeNjvBSX5c/PHdKmLvJY2/l/yZFKnM7GJ71Rh+rLR0/pXTsILY8EAUw0LcoTA42dGEw15HQLRGoIdbhY4245M19ohrMuBxKXdPLXTN079rVd7pTF/o0KMRuKPOxkKgbe8Mnnk7V4qqTenQlQMiZIGmJy+WQyo2eagzHAgE/fXyIyJKhQSV/HS4Luf+k/D/fuiaXQ5qN9hp/U8k+1xX16bvROfLWeNcV0rVWblS0S+fxmH8kB5lGxCRZretcrcEMFzShEoAzFPrssGhVuOZJMvVjJ+PYKpCyA7gX3LWaNq+o7ybmoOj4KIt/9J2KNsdvV0bdwlGOnfaGO178slBMsNAVqjLoqN2ONbEKkFpVwuanM9g5mgfnjVEUG5BtZCKM5BlGOgPlwpeb7b0f7CaYunoAFxel5xZEK9RTIzIoOxdiUXR6kiFQgGHMpYvseci2Aoa/8gYf5JnJt+grcslC+UvDPyb62Qjox3ZYlsBSjkLneXiYHmH93k5CNAy1i+2W6dtHXKeuVVWYBEdbvJKRl5bByYvTKWWQ4eB+oIWv/MrDVLHIi3+aJDErbByq5RCQ4VsbUPt4gAUZ/7mvTSgi1ZNlh8lflWcpTBtPyyc4XkuFWRnOoQzpSiomVJZ4y2TjO46oVZPGi7drzxFoWwrWE4QLobCpKyyTghEQRlio/HUFoG9b5FUoAhPeaBXVcuQafMv0iKNFUTq/BpPQKVSMmqvd1lkkfVeBH+3kMzTVddGs4YAQ==', type='signature_delta'), index=0, type='content_block_delta')
RawContentBlockStopEvent(index=0, type='content_block_stop')
RawContentBlockStartEvent(content_block=TextBlock(citations=None, text='', type='text'), index=1, type='content_block_start')
RawContentBlockDeltaEvent(delta=TextDelta(text='**Synchronization**', type='text_delta'), index=1, type='content_block_delta')
RawContentBlockStopEvent(index=1, type='content_block_stop')
RawMessageDeltaEvent(delta=Delta(container=None, stop_details=None, stop_reason='end_turn', stop_sequence=None), type='message_delta', usage=MessageDeltaUsage(cache_creation_input_tokens=0, cache_read_input_tokens=0, input_tokens=19, output_tokens=400, output_tokens_details=OutputTokensDetails(thinking_tokens=390), server_tool_use=None))
RawMessageStopEvent(type='message_stop')

Let’s test async.

from anthropic import AsyncAnthropic

cli = AsyncAnthropic()

r = await cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant async"))
r

🟢 HIT 8853e499
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant async"}],"model":"claude-sonnet-5"}'

Message(id='msg_011CdmMHb4jwgt9k3jNUuqC4', container=None, content=[TextBlock(citations=None, text='**Concurrency**', type='text')], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason='end_turn', stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=15, output_tokens=9, output_tokens_details=OutputTokensDetails(thinking_tokens=0), server_tool_use=None, service_tier='standard'))

r = await cli.messages.create(model=mods.ant, max_tokens=1024, messages=mk_msgs("ant async"))
r

🟢 HIT 8853e499
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant async"}],"model":"claude-sonnet-5"}'

Message(id='msg_011CdmMHb4jwgt9k3jNUuqC4', container=None, content=[TextBlock(citations=None, text='**Concurrency**', type='text')], model='claude-sonnet-5', role='assistant', stop_details=None, stop_reason='end_turn', stop_sequence=None, type='message', usage=Usage(cache_creation=CacheCreation(ephemeral_1h_input_tokens=0, ephemeral_5m_input_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', input_tokens=15, output_tokens=9, output_tokens_details=OutputTokensDetails(thinking_tokens=0), server_tool_use=None, service_tier='standard'))

Let’s test async streaming.

r = await cli.messages.create(model=mods.ant,max_tokens=1024,messages=mk_msgs("ant async streaming"), stream=True)
async for ch in r.response.aiter_bytes(): print(ch.decode())

🟢 HIT 49ccbedd
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant async streaming"}],"model":"claude-sonnet-5","stream":true}'
event: message_start
data: {"type":"message_start","message":{"model":"claude-sonnet-5","id":"msg_011CdmMHoDyj2hbgzrBQgXgu","type":"message","role":"assistant","content":[],"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"output_tokens":1,"service_tier":"standard","inference_geo":"global"}}            }

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}             }

event: ping
data: {"type": "ping"}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"**"}        }

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Reactive**"}           }

event: content_block_stop
data: {"type":"content_block_stop","index":0   }

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null,"stop_details":null},"usage":{"input_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":8,"output_tokens_details":{"thinking_tokens":0}}  }

event: message_stop
data: {"type":"message_stop"           }

r = await cli.messages.create(model=mods.ant,max_tokens=1024,messages=mk_msgs("ant async streaming"), stream=True)
async for ch in r.response.aiter_bytes(): print(ch.decode())

🟢 HIT 49ccbedd
b'{"max_tokens":1024,"messages":[{"role":"user","content":"write 1 word about ant async streaming"}],"model":"claude-sonnet-5","stream":true}'
event: message_start
data: {"type":"message_start","message":{"model":"claude-sonnet-5","id":"msg_011CdmMHoDyj2hbgzrBQgXgu","type":"message","role":"assistant","content":[],"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"output_tokens":1,"service_tier":"standard","inference_geo":"global"}}            }

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}             }

event: ping
data: {"type": "ping"}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"**"}        }

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Reactive**"}           }

event: content_block_stop
data: {"type":"content_block_stop","index":0   }

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null,"stop_details":null},"usage":{"input_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":8,"output_tokens_details":{"thinking_tokens":0}}  }

event: message_stop
data: {"type":"message_stop"           }

Tool Calls

As a sanity check let’s confirm that tool calls work. fastllm’s AsyncChat runs the whole tool loop over httpx – the tool-call request and the follow-up carrying the tool result – so both API calls get cached.

from fastllm.chat import AsyncChat

def get_current_weather(
    location:str, # City and country, e.g. "Reims, France"
):
    "Get the current weather in a given location"
    return "rainy"

chat = AsyncChat(mods.ant, tools=[get_current_weather])
r1 = await chat("Is it raining in Reims?")
r1

🟢 HIT cbbd5dd6
b'{"model":"claude-sonnet-5","messages":[{"role":"user","content":[{"type":"text","text":"Is it raining in Reims?"}]}],"max_tokens":128000,"tools":[{"name":"get_current_weather","description":"Get the current weather in a given location","input_schema":{"type":"object","properties":{"location":{"description":"City and country, e.g. \\"Reims, France\\"","type":"string"}},"required":["location"]}}]}'
🟢 HIT c80e444e
b'{"model":"claude-sonnet-5","messages":[{"role":"user","content":[{"type":"text","text":"Is it raining in Reims?"}]},{"role":"assistant","content":[{"type":"thinking","thinking":"","signature":"EpsCCokBCBAYAipAi72hDNnvmk2CmelOemYkgBTdjl44Rb0TIaqElgULwJdN82ea9UX4xjwwkaAFzBMqNwwoHCeCoKv1zKHjHkYJ6zIPY2xhdWRlLXNvbm5ldC01OABCCHRoaW5raW5nWiQ4ODk4YTFkYy0yMTNkLTRhNmYtOTljYi03ZTBlNTUzZDc0NWISDNZAjysmkiSd6aI4NxoMLZDLIoCVLtRfNUQzIjD1Ku3OqdF9f0t7ZEoz5RbAubB+kab059kC3FqKoX5IUWWJwCZe2nDUBhYB/Ugxh84qP4Z4NN4CZV8hw0u1To96nLHjl/fv/1qumFdv3bjWx8Hf8F66t+HDUFcHpi6E7roWzpTLQaMZ0X3WACJyTgLT5BgB"},{"type":"tool_use","id":"toolu_01CzcMUVvBaQdVZrZaD91XMv","name":"get_current_weather","input":{"location":"Reims, France"},"caller":{"type":"direct"}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_01CzcMUVvBaQdVZrZaD91XMv","content":"rainy"}]},{"role":"user","content":[{"type":"text","text":"You have used all your tool calls for this turn. Please summarize your findings. If you did not complete your goal, tell the user what further work is needed. You may use tools again on the next user message."}]}],"max_tokens":128000,"tools":[{"name":"get_current_weather","description":"Get the current weather in a given location","input_schema":{"type":"object","properties":{"location":{"description":"City and country, e.g. \\"Reims, France\\"","type":"string"}},"required":["location"]}}],"tool_choice":{"type":"none"}}'

Yes, it’s currently raining in Reims, France. ☔

If you’re heading out, you’ll want an umbrella or rain jacket!

model: claude-sonnet-5
finish_reason: stop
usage: Usage(prompt_tokens=603, completion_tokens=46, total_tokens=649, cached_tokens=0, cache_creation_tokens=0, reasoning_tokens=0, raw={'input_tokens': 603, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'cache_creation': {'ephemeral_5m_input_tokens': 0, 'ephemeral_1h_input_tokens': 0}, 'output_tokens': 46, 'output_tokens_details': {'thinking_tokens': 0}, 'service_tier': 'standard', 'inference_geo': 'global'})

chat = AsyncChat(mods.ant, tools=[get_current_weather])
r2 = await chat("Is it raining in Reims?")
test_eq(str(r1), str(r2))

🟢 HIT cbbd5dd6
b'{"model":"claude-sonnet-5","messages":[{"role":"user","content":[{"type":"text","text":"Is it raining in Reims?"}]}],"max_tokens":128000,"tools":[{"name":"get_current_weather","description":"Get the current weather in a given location","input_schema":{"type":"object","properties":{"location":{"description":"City and country, e.g. \\"Reims, France\\"","type":"string"}},"required":["location"]}}]}'
🟢 HIT c80e444e
b'{"model":"claude-sonnet-5","messages":[{"role":"user","content":[{"type":"text","text":"Is it raining in Reims?"}]},{"role":"assistant","content":[{"type":"thinking","thinking":"","signature":"EpsCCokBCBAYAipAi72hDNnvmk2CmelOemYkgBTdjl44Rb0TIaqElgULwJdN82ea9UX4xjwwkaAFzBMqNwwoHCeCoKv1zKHjHkYJ6zIPY2xhdWRlLXNvbm5ldC01OABCCHRoaW5raW5nWiQ4ODk4YTFkYy0yMTNkLTRhNmYtOTljYi03ZTBlNTUzZDc0NWISDNZAjysmkiSd6aI4NxoMLZDLIoCVLtRfNUQzIjD1Ku3OqdF9f0t7ZEoz5RbAubB+kab059kC3FqKoX5IUWWJwCZe2nDUBhYB/Ugxh84qP4Z4NN4CZV8hw0u1To96nLHjl/fv/1qumFdv3bjWx8Hf8F66t+HDUFcHpi6E7roWzpTLQaMZ0X3WACJyTgLT5BgB"},{"type":"tool_use","id":"toolu_01CzcMUVvBaQdVZrZaD91XMv","name":"get_current_weather","input":{"location":"Reims, France"},"caller":{"type":"direct"}}]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_01CzcMUVvBaQdVZrZaD91XMv","content":"rainy"}]},{"role":"user","content":[{"type":"text","text":"You have used all your tool calls for this turn. Please summarize your findings. If you did not complete your goal, tell the user what further work is needed. You may use tools again on the next user message."}]}],"max_tokens":128000,"tools":[{"name":"get_current_weather","description":"Get the current weather in a given location","input_schema":{"type":"object","properties":{"location":{"description":"City and country, e.g. \\"Reims, France\\"","type":"string"}},"required":["location"]}}],"tool_choice":{"type":"none"}}'

Multipart Request

cli = Anthropic()
r = cli.beta.files.upload(file=("ex.txt", b"hello world", "text/plain"))
r

🟢 HIT 461a2b1c
b'--8b86e4dca32893a9b488e35e9745211b\r\nContent-Disposition: form-data; name="file"; filename="ex.txt"\r\nContent-Type: text/plain\r\n\r\nhello world\r\n--8b86e4dca32893a9b488e35e9745211b--\r\n'

FileMetadata(id='file_011CdmMJQxJgZnzfxDtdrmjB', created_at=datetime.datetime(2026, 8, 6, 10, 38, 51, 323000, tzinfo=datetime.timezone.utc), filename='ex.txt', mime_type='text/plain', size_bytes=11, type='file', downloadable=False, scope=None)

cli = Anthropic()
r = cli.beta.files.upload(file=("ex.txt", b"hello world", "text/plain"))
r

🟢 HIT 461a2b1c
b'--19565653bdae34d44d5ec19901e38dd1\r\nContent-Disposition: form-data; name="file"; filename="ex.txt"\r\nContent-Type: text/plain\r\n\r\nhello world\r\n--19565653bdae34d44d5ec19901e38dd1--\r\n'

FileMetadata(id='file_011CdmMJQxJgZnzfxDtdrmjB', created_at=datetime.datetime(2026, 8, 6, 10, 38, 51, 323000, tzinfo=datetime.timezone.utc), filename='ex.txt', mime_type='text/plain', size_bytes=11, type='file', downloadable=False, scope=None)

Requests Library

The requests patch covers any requests.Session (httpbingo echoes a unique X-Request-Start timestamp per real call, so equal bodies prove the second response came from cache).

r1 = requests.post('https://httpbingo.org/post', json={'a': 'req'})
r1

🟢 HIT 64cdef80
b'{"a": "req"}'

<Response [200]>

r2 = requests.post('https://httpbingo.org/post', json={'a': 'req'})
test_eq(r1.text, r2.text)
test_eq(r2.headers['content-type'], r1.headers['content-type'])

🟢 HIT 64cdef80
b'{"a": "req"}'

Gemini File Upload

The google-genai SDK’s files.upload() relies on x-goog-upload-url and x-goog-upload-status response headers

from google import genai

cli = genai.Client()

When no hdrs is provided the request fails:

tfw = tempfile.NamedTemporaryFile(suffix='.txt')

f = tfw.__enter__()
fn = Path(f.name)
fn.write_text("test content");

try: gfile = cli.files.upload(file=fn)
except Exception as e: print(e)

🟢 HIT 74953d92
b'{"file": {"mime_type": "text/plain", "size_bytes": 12}}'
🟢 HIT 5fb8c337
b'test content'

When caching Gemini file uploads, by default request content only includes mime_type and size_bytes. This means different files with the same mime type and size produce identical cache keys, causing incorrect cache hits. The fix is to pass a file content fingerprint (a hash of the file bytes) as the display_name in the upload config: cli.files.upload(file=fn, config={"display_name": _fingerprint(fn)}). This ensures the request body is unique per file content, generating distinct cache keys.

def _fingerprint(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest()[:16]

enable_cachy(hdrs=['x-goog-upload-url', 'x-goog-upload-status'])

gfile = cli.files.upload(file=fn, config={"display_name": _fingerprint(fn)})
gfile

File(
  create_time=datetime.datetime(2026, 8, 6, 10, 38, 59, 130188, tzinfo=TzInfo(0)),
  display_name='6ae8a75555209fd6',
  expiration_time=datetime.datetime(2026, 8, 8, 10, 38, 57, 980758, tzinfo=TzInfo(0)),
  mime_type='text/plain',
  name='files/no9whcrqkdk5',
  sha256_hash='NmFlOGE3NTU1NTIwOWZkNmM0NDE1N2MwYWVkODAxNmU3NjNmZjQzNWExOWNmMTg2Zjc2ODYzMTQwMTQzZmY3Mg==',
  size_bytes=12,
  source=<FileSource.UPLOADED: 'UPLOADED'>,
  state=<FileState.ACTIVE: 'ACTIVE'>,
  update_time=datetime.datetime(2026, 8, 6, 10, 38, 59, 130188, tzinfo=TzInfo(0)),
  uri='https://generativelanguage.googleapis.com/v1beta/files/no9whcrqkdk5'
)

tfw.__exit__(None, None, None)

httpx.stream

headers = {"x-api-key": os.environ["ANTHROPIC_API_KEY"], "anthropic-version": "2023-06-01", "content-type": "application/json"}
url = "https://api.anthropic.com/v1/messages"
payload = json.dumps({"model": mods.ant, "max_tokens": 16, "messages": mk_msgs("ant sync")}).encode()

with httpx.stream("POST", url, headers=headers, content=payload) as r1: c1 = json.loads(b''.join(r1.iter_bytes()).decode())
with httpx.stream("POST", url, headers=headers, content=payload) as r2: c2 = json.loads(b''.join(r2.iter_bytes()).decode())

c1['content'][0]['text']

'**Pheromones**\n\nThis single word captures the essence'

test_eq(c1,c2)

Binary Content

headers = {"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}", "content-type": "application/json"}
url = "https://api.openai.com/v1/audio/speech"
payload = json.dumps({"model": "tts-1", "input": "cachy binary test", "voice": "alloy"}).encode()

r1 = httpx.post(url, headers=headers, content=payload)
r2 = httpx.post(url, headers=headers, content=payload)

test_eq(r1.content, r2.content)
test_eq(isinstance(r1.content, bytes), True)

os.chdir(origdir)

Introduction

Patching httpx

Cache Filtering

Cache Key

Cache Reads/Writes

Multipart Requests

Streaming

enable_cachy

Async

enable_cachy

disable_cachy

Tests

OpenAI

Anthropic

Tool Calls

Multipart Request

Requests Library

Gemini File Upload

httpx.stream

Binary Content

Patching `httpx`

`enable_cachy`