Core

Lisette Core

LiteLLM

Deterministic outputs

LiteLLM ModelResponse(Stream) objects have id and created_at fields that are generated dynamically. Even when we use cachy to cache the LLM response these dynamic fields create diffs which makes code review more challenging. The patches below ensure that id and created_at fields are fixed and won’t generate diffs.

source

patch_litellm


def patch_litellm(
    seed:int=0
):

Patch litellm.ModelResponseBase such that id and created are fixed.

patch_litellm()

Completion

LiteLLM provides an convenient unified interface for most big LLM providers. Because it’s so useful to be able to switch LLM providers with just one argument. We want to make it even easier to by adding some more convenience functions and classes.

This is very similar to our other wrapper libraries for popular AI providers: claudette (Anthropic), gaspard (Gemini), cosette (OpenAI).

# litellm._turn_on_debug()
ms = ["gemini/gemini-3-pro-preview", "gemini/gemini-3-flash-preview", "claude-sonnet-4-6", "openai/gpt-4.1"]
msg = [{'role':'user','content':'Hey there!', 'cache_control': {'type': 'ephemeral'}}]
for m in ms:
    display(Markdown(f'**{m}:**'))
    display(completion(m,msg))

gemini/gemini-3-pro-preview:

Hey! How can I help you today?

  • id: chatcmpl-xxx
  • model: gemini-3-pro-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=260, prompt_tokens=4, total_tokens=264, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=251, rejected_prediction_tokens=None, text_tokens=9, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=4, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)

gemini/gemini-3-flash-preview:

Hello! How can I help you today?

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=73, prompt_tokens=4, total_tokens=77, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=65, rejected_prediction_tokens=None, text_tokens=8, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=4, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)

claude-sonnet-4-6:

Hey there! How’s it going? What’s on your mind? 😊

  • id: chatcmpl-xxx
  • model: claude-sonnet-4-6
  • finish_reason: stop
  • usage: Usage(completion_tokens=20, prompt_tokens=10, total_tokens=30, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=0, rejected_prediction_tokens=None, text_tokens=20, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=0, text_tokens=None, image_tokens=None, video_tokens=None, cache_creation_tokens=0, cache_creation_token_details=CacheCreationTokenDetails(ephemeral_5m_input_tokens=0, ephemeral_1h_input_tokens=0)), cache_creation_input_tokens=0, cache_read_input_tokens=0, inference_geo='global', speed=None)

openai/gpt-4.1:

Hello! How can I help you today? 😊

  • id: chatcmpl-xxx
  • model: gpt-4.1-2025-04-14
  • finish_reason: stop
  • usage: Usage(completion_tokens=10, prompt_tokens=10, total_tokens=20, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=0, audio_tokens=0, reasoning_tokens=0, rejected_prediction_tokens=0, text_tokens=None, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=0, cached_tokens=0, text_tokens=None, image_tokens=None, video_tokens=None))

Generated images are also displayed (not shown here to conserve filesize):

# completion(model='gemini/gemini-2.5-flash-image', messages=[{'role':'user','content':'Draw a simple sketch of a cat'}])

Messages formatting

Let’s start with making it easier to pass messages into litellm’s completion function (including images, and pdf files).

If msg has tool_calls, cache_control is added to the last tool call (required since LiteLLM strips it from empty content blocks), otherwise to the content.

source

stop_reason


def stop_reason(
    r
):

Call self as a function.

source

contents


def contents(
    r
):

Get message object from response r.

source

remove_cache_ckpts


def remove_cache_ckpts(
    msg
):

remove cache checkpoints and return msg.

Test with regular content message:

msg_content = {'role': 'user', 'content': [{'type': 'text', 'text': 'hello'}]}
_add_cache_control(msg_content)
test_eq(msg_content['content'][-1].get('cache_control'), {'type': 'ephemeral'})
test_eq(_has_cache(msg_content), True)
remove_cache_ckpts(msg_content)
test_eq(_has_cache(msg_content), False)

Test with assistant message with tool_calls:

tcs = [
    {'id': 'tc1', 'type': 'function', 'function': {'name': 'test', 'arguments': '{}'}},
    {'id': 'tc2', 'type': 'function', 'function': {'name': 'test', 'arguments': '{}'}}
]
msg_tool = {'role': 'assistant', 'content': '', 'tool_calls': tcs}
_add_cache_control(msg_tool)
test_eq(msg_tool['tool_calls'][-1].get('cache_control'), {'type': 'ephemeral'})
test_eq('cache_control' not in msg_tool.get('content', [{}])[-1] if msg_tool.get('content') else True, True)  # no cache in content
test_eq(_has_cache(msg_tool), True)
remove_cache_ckpts(msg_tool)
test_eq(_has_cache(msg_tool), False)

Test with ChatCompletionMessageToolCall tool call object:

tcs =[
    ChatCompletionMessageToolCall(id='tc1', type='function', function=Function(name='test', arguments='{}')), 
    ChatCompletionMessageToolCall(id='tc2', type='function', function=Function(name='test', arguments='{}'))
]
msg_tc_obj = {'role': 'assistant', 'content': '', 'tool_calls': tcs}
_add_cache_control(msg_tc_obj)
test_eq(getattr(msg_tc_obj['tool_calls'][-1], 'cache_control', None), {'type': 'ephemeral'})
test_eq(_has_cache(msg_tc_obj), True)
remove_cache_ckpts(msg_tc_obj)
test_eq(_has_cache(msg_tc_obj), False)

source

mk_msg


def mk_msg(
    content, # Content: str, bytes (image), list of mixed content, or dict w 'role' and 'content' fields
    role:str='user', # Message role if content isn't already a dict/Message
    cache:bool=False, # Enable Anthropic caching
    ttl:NoneType=None, # Cache TTL: '5m' (default) or '1h'
):

Create a LiteLLM compatible message.

Now we can use mk_msg to create different types of messages.

Simple text:

msg = mk_msg("hey")
msg
{'role': 'user', 'content': 'hey'}

Which can be passed to litellm’s completion function like this:

model = ms[1] # use 2.5-pro, 3-pro is very slow even to run tests as of making
res = completion(model, [msg])
res

Hey there! How can I help you today?

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=35, prompt_tokens=2, total_tokens=37, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=25, rejected_prediction_tokens=None, text_tokens=10, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=2, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)

We’ll add a little shortcut to make examples and testing easier here:

def c(msgs, m=model, **kw):
    msgs = [msgs] if isinstance(msgs,dict) else listify(msgs)
    return completion(m, msgs, **kw)
c(msg)

Hey there! How can I help you today?

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=35, prompt_tokens=2, total_tokens=37, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=25, rejected_prediction_tokens=None, text_tokens=10, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=2, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)

Lists w just one string element are flattened for conciseness:

test_eq(mk_msg("hey"), mk_msg(["hey"]))

(LiteLLM ignores these fields when sent to other providers)

Text and images:

img_fn = Path('samples/puppy.jpg')
Image(filename=img_fn, width=200)

msg = mk_msg(['hey what in this image?',img_fn.read_bytes()])
print(json.dumps(msg,indent=1)[:200]+"...")
{
 "role": "user",
 "content": [
  {
   "type": "text",
   "text": "hey what in this image?"
  },
  {
   "type": "image_url",
   "image_url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD/4gxUSU...
c(msg)

This image features an adorable Cavalier King Charles Spaniel puppy lying in a garden.

Here are the details:

  • The Puppy: It has classic “Blenheim” markings, which consist of a white coat with chestnut (reddish-brown) patches over its ears and around its large, dark eyes. It has a white muzzle, a small black nose, and characteristic long, wavy ears. The puppy is looking directly at the camera with a sweet expression, with one white front paw stretched forward.
  • The Setting: The puppy is nestled next to a lush bush of small, vibrant purple flowers (likely asters or similar). It is lying on a patch of green grass.
  • Background: To the right and behind the puppy is a blurred, dark textured object, possibly a large terracotta flower pot or a wooden structure.

The overall feel of the image is soft, charming, and peaceful.

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=745, prompt_tokens=1087, total_tokens=1832, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=550, rejected_prediction_tokens=None, text_tokens=195, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=7, image_tokens=1080, video_tokens=None), cache_read_input_tokens=None)

Let’s also demonstrate this for PDFs

pdf_fn = Path('samples/solveit.pdf')
msg = mk_msg(['Who is the author of this pdf?', pdf_fn.read_bytes()])
c(msg)

Based on the text in the document, the author is Jeremy Howard, who is a co-founder of fast.ai.

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=153, prompt_tokens=541, total_tokens=694, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=127, rejected_prediction_tokens=None, text_tokens=26, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=9, image_tokens=532, video_tokens=None), cache_read_input_tokens=None)

Some models like Gemini support audio and video:

wav_data = httpx.get("https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav").content
# Audio(wav_data)  # uncomment to preview
msg = mk_msg(['What is this audio saying?', wav_data])
completion(ms[1], [msg])

The audio says: “The sun rises in the east and sets in the west. This simple fact has been observed by humans for thousands of years.”

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=147, prompt_tokens=181, total_tokens=328, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=117, rejected_prediction_tokens=None, text_tokens=30, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=174, cached_tokens=None, text_tokens=7, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)
vid_data = httpx.get("https://storage.googleapis.com/github-repo/img/gemini/multimodality_usecases_overview/pixel8.mp4").content
msg = mk_msg(['Concisely, what is happening in this video?', vid_data])
completion(ms[1], [msg])

Tokyo photographer Saeka Shimada explores the city’s neon-lit streets and alleyways at night, demonstrating the Google Pixel 8 Pro’s advanced low-light filming and photography features.

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=632, prompt_tokens=5205, total_tokens=5837, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=593, rejected_prediction_tokens=None, text_tokens=39, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=12, image_tokens=None, video_tokens=5193), cache_read_input_tokens=None)

Caching

Some providers such as Anthropic require manually opting into caching. Let’s try it:

def cpr(i): return f'{i} '*1024 + 'This is a caching test. Report back only what number you see repeated above.'
disable_cachy()
# msg = mk_msg(cpr(1), cache=True)
# res = c(msg, ms[2])
# res

Anthropic has a maximum of 4 cache checkpoints, so we remove previous ones as we go:

# res = c([remove_cache_ckpts(msg), mk_msg(res), mk_msg(cpr(2), cache=True)], ms[2])
# res

We see that the first message was cached, and this extra message has been written to cache:

# res.usage.prompt_tokens_details

We can add a bunch of large messages in a loop to see how the number of cached tokens used grows.

We do this for 25 times to ensure it still works for more than >20 content blocks, which is a known anthropic issue.

The code below is commented by default, because it’s slow. Please uncomment when working on caching.

# h = []
# msg = mk_msg(cpr(1), cache=True)

# for o in range(2,25):
#     h += [remove_cache_ckpts(msg), mk_msg(res)]
#     msg = mk_msg(cpr(o), cache=True)
#     res = c(h+[msg])
#     detls = res.usage.prompt_tokens_details
#     print(o, detls.cached_tokens, detls.cache_creation_tokens, end='; ')
enable_cachy()

Reconstructing formatted outputs

Lisette can call multiple tools in a loop. Further down this notebook, we’ll provide convenience functions for formatting such a sequence of toolcalls and responses into one formatted output string.

For now, we’ll show an example and show how to transform such a formatted output string back into a valid LiteLLM history.

fmt_outp = '''
I'll solve this step-by-step, using parallel calls where possible.

<details class='tool-usage-details'>

```json
{
  "id": "toolu_01KjnQH2Nsz2viQ7XYpLW3Ta",
  "call": { "function": "simple_add", "arguments": { "a": 10, "b": 5 } },
  "result": "15"
}
```

</details>

<details class='tool-usage-details'>

```json
{
  "id": "toolu_01Koi2EZrGZsBbnQ13wuuvzY",
  "call": { "function": "simple_add", "arguments": { "a": 2, "b": 1 } },
  "result": "3"
}
```

</details>

Now I need to multiply 15 * 3 before I can do the final division:

<details class='tool-usage-details'>

```json
{
  "id": "toolu_0141NRaWUjmGtwxZjWkyiq6C",
  "call": { "function": "multiply", "arguments": { "a": 15, "b": 3 } },
  "result": "45"
}
```

</details>

<details class='token-usage-details'><summary>Cache hit: 81.8% | Tokens: total=23,276 input=23,158 (+18,910 cached, 0 new) output=118 (reasoning 23)</summary>

`Usage(completion_tokens=118, prompt_tokens=23158, total_tokens=23276, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=23, rejected_prediction_tokens=None, text_tokens=None, image_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=18910, text_tokens=None, image_tokens=None, cache_creation_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=18910)`

</details>
'''

We can split into chunks of (text,toolstr,json):

sp = re_tools.split(fmt_outp)
for o in list(chunked(sp, 3, pad=True)): print('- ', o)
-  ["\nI'll solve this step-by-step, using parallel calls where possible.\n\n", '<details class=\'tool-usage-details\'>\n\n```json\n{\n  "id": "toolu_01KjnQH2Nsz2viQ7XYpLW3Ta",\n  "call": { "function": "simple_add", "arguments": { "a": 10, "b": 5 } },\n  "result": "15"\n}\n```\n\n</details>', None]
-  ['{\n  "id": "toolu_01KjnQH2Nsz2viQ7XYpLW3Ta",\n  "call": { "function": "simple_add", "arguments": { "a": 10, "b": 5 } },\n  "result": "15"\n}', '\n\n', '<details class=\'tool-usage-details\'>\n\n```json\n{\n  "id": "toolu_01Koi2EZrGZsBbnQ13wuuvzY",\n  "call": { "function": "simple_add", "arguments": { "a": 2, "b": 1 } },\n  "result": "3"\n}\n```\n\n</details>']
-  [None, '{\n  "id": "toolu_01Koi2EZrGZsBbnQ13wuuvzY",\n  "call": { "function": "simple_add", "arguments": { "a": 2, "b": 1 } },\n  "result": "3"\n}', '\n\nNow I need to multiply 15 * 3 before I can do the final division:\n\n']
-  ['<details class=\'tool-usage-details\'>\n\n```json\n{\n  "id": "toolu_0141NRaWUjmGtwxZjWkyiq6C",\n  "call": { "function": "multiply", "arguments": { "a": 15, "b": 3 } },\n  "result": "45"\n}\n```\n\n</details>', None, '{\n  "id": "toolu_0141NRaWUjmGtwxZjWkyiq6C",\n  "call": { "function": "multiply", "arguments": { "a": 15, "b": 3 } },\n  "result": "45"\n}']
-  ["\n\n<details class='token-usage-details'><summary>Cache hit: 81.8% | Tokens: total=23,276 input=23,158 (+18,910 cached, 0 new) output=118 (reasoning 23)</summary>\n\n`Usage(completion_tokens=118, prompt_tokens=23158, total_tokens=23276, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=23, rejected_prediction_tokens=None, text_tokens=None, image_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=18910, text_tokens=None, image_tokens=None, cache_creation_tokens=0), cache_creation_input_tokens=0, cache_read_input_tokens=18910)`\n\n</details>\n", None, None]

source

fmt2hist


def fmt2hist(
    outp:str
)->list:

Transform a formatted output into a LiteLLM compatible history

source

split_tools


def split_tools(
    s
):

Split formatted output into (text, summary, tooljson) chunks

See how we can turn that one formatted output string back into a list of Messages:

from pprint import pprint
h = fmt2hist(fmt_outp)
pprint(h)
[Message(content="I'll solve this step-by-step, using parallel calls where possible.", role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 10, "b": 5}', name='simple_add'), id='toolu_01KjnQH2Nsz2viQ7XYpLW3Ta', type='function')], function_call=None, provider_specific_fields=None),
 {'content': '15',
  'name': 'simple_add',
  'role': 'tool',
  'tool_call_id': 'toolu_01KjnQH2Nsz2viQ7XYpLW3Ta'},
 Message(content='', role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 2, "b": 1}', name='simple_add'), id='toolu_01Koi2EZrGZsBbnQ13wuuvzY', type='function')], function_call=None, provider_specific_fields=None),
 {'content': '3',
  'name': 'simple_add',
  'role': 'tool',
  'tool_call_id': 'toolu_01Koi2EZrGZsBbnQ13wuuvzY'},
 Message(content='Now I need to multiply 15 * 3 before I can do the final division:', role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 15, "b": 3}', name='multiply'), id='toolu_0141NRaWUjmGtwxZjWkyiq6C', type='function')], function_call=None, provider_specific_fields=None),
 {'content': '45',
  'name': 'multiply',
  'role': 'tool',
  'tool_call_id': 'toolu_0141NRaWUjmGtwxZjWkyiq6C'},
 Message(content='.', role='assistant', tool_calls=None, function_call=None, provider_specific_fields=None)]

mk_msgs

We will skip tool use blocks and tool results during caching

Now lets make it easy to provide entire conversations:

source

mk_msgs


def mk_msgs(
    msgs, # List of messages (each: str, bytes, list, or dict w 'role' and 'content' fields)
    cache:bool=False, # Enable Anthropic caching
    cache_idxs:list=[-1], # Cache breakpoint idxs
    ttl:NoneType=None, # Cache TTL: '5m' (default) or '1h'
):

Create a list of LiteLLM compatible messages.

With mk_msgs you can easily provide a whole conversation:

msgs = mk_msgs(['Hey!',"Hi there!","How are you?","I'm doing fine and you?"])
msgs
[{'role': 'user', 'content': 'Hey!'},
 {'role': 'assistant', 'content': 'Hi there!'},
 {'role': 'user', 'content': 'How are you?'},
 {'role': 'assistant', 'content': "I'm doing fine and you?"}]

By defualt the last message will be cached when cache=True:

msgs = mk_msgs(['Hey!',"Hi there!","How are you?","I'm doing fine and you?"], cache=True)
msgs
[{'role': 'user', 'content': 'Hey!'},
 {'role': 'assistant', 'content': 'Hi there!'},
 {'role': 'user', 'content': 'How are you?'},
 {'role': 'assistant',
  'content': [{'type': 'text',
    'text': "I'm doing fine and you?",
    'cache_control': {'type': 'ephemeral'}}]}]
test_eq('cache_control' in msgs[-1]['content'][0], True)

Alternatively, users can provide custom cache_idxs. Tool call blocks and results are skipped during caching:

msgs = mk_msgs(['Hello!','Hi! How can I help you?','Call some functions!',fmt_outp], cache=True, cache_idxs=[0,-2,-1])
msgs
[{'role': 'user',
  'content': [{'type': 'text',
    'text': 'Hello!',
    'cache_control': {'type': 'ephemeral'}}]},
 {'role': 'assistant', 'content': 'Hi! How can I help you?'},
 {'role': 'user', 'content': 'Call some functions!'},
 Message(content="I'll solve this step-by-step, using parallel calls where possible.", role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 10, "b": 5}', name='simple_add'), id='toolu_01KjnQH2Nsz2viQ7XYpLW3Ta', type='function')], function_call=None, provider_specific_fields=None),
 {'role': 'tool',
  'tool_call_id': 'toolu_01KjnQH2Nsz2viQ7XYpLW3Ta',
  'name': 'simple_add',
  'content': '15'},
 Message(content='', role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 2, "b": 1}', name='simple_add'), id='toolu_01Koi2EZrGZsBbnQ13wuuvzY', type='function')], function_call=None, provider_specific_fields=None),
 {'role': 'tool',
  'tool_call_id': 'toolu_01Koi2EZrGZsBbnQ13wuuvzY',
  'name': 'simple_add',
  'content': '3'},
 {'content': 'Now I need to multiply 15 * 3 before I can do the final division:',
  'role': 'assistant',
  'tool_calls': [ChatCompletionMessageToolCall(function=Function(arguments='{"a": 15, "b": 3}', name='multiply'), id='toolu_0141NRaWUjmGtwxZjWkyiq6C', type='function', cache_control={'type': 'ephemeral'})],
  'function_call': None,
  'provider_specific_fields': None},
 {'role': 'tool',
  'tool_call_id': 'toolu_0141NRaWUjmGtwxZjWkyiq6C',
  'name': 'multiply',
  'content': '45'},
 {'content': [{'type': 'text',
    'text': '.',
    'cache_control': {'type': 'ephemeral'}}],
  'role': 'assistant',
  'tool_calls': None,
  'function_call': None,
  'provider_specific_fields': None}]
msgs[-2]
{'role': 'tool',
 'tool_call_id': 'toolu_0141NRaWUjmGtwxZjWkyiq6C',
 'name': 'multiply',
 'content': '45'}
msgs = mk_msgs(['Hello!','Hi! How can I help you?','Call some functions!',fmt_outp], cache=True, cache_idxs=[0,-3,-2])
msgs
[{'role': 'user',
  'content': [{'type': 'text',
    'text': 'Hello!',
    'cache_control': {'type': 'ephemeral'}}]},
 {'role': 'assistant', 'content': 'Hi! How can I help you?'},
 {'role': 'user', 'content': 'Call some functions!'},
 Message(content="I'll solve this step-by-step, using parallel calls where possible.", role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 10, "b": 5}', name='simple_add'), id='toolu_01KjnQH2Nsz2viQ7XYpLW3Ta', type='function')], function_call=None, provider_specific_fields=None),
 {'role': 'tool',
  'tool_call_id': 'toolu_01KjnQH2Nsz2viQ7XYpLW3Ta',
  'name': 'simple_add',
  'content': '15'},
 {'content': '',
  'role': 'assistant',
  'tool_calls': [ChatCompletionMessageToolCall(function=Function(arguments='{"a": 2, "b": 1}', name='simple_add'), id='toolu_01Koi2EZrGZsBbnQ13wuuvzY', type='function', cache_control={'type': 'ephemeral'})],
  'function_call': None,
  'provider_specific_fields': None},
 {'role': 'tool',
  'tool_call_id': 'toolu_01Koi2EZrGZsBbnQ13wuuvzY',
  'name': 'simple_add',
  'content': '3'},
 {'content': 'Now I need to multiply 15 * 3 before I can do the final division:',
  'role': 'assistant',
  'tool_calls': [ChatCompletionMessageToolCall(function=Function(arguments='{"a": 15, "b": 3}', name='multiply'), id='toolu_0141NRaWUjmGtwxZjWkyiq6C', type='function', cache_control={'type': 'ephemeral'})],
  'function_call': None,
  'provider_specific_fields': None},
 {'role': 'tool',
  'tool_call_id': 'toolu_0141NRaWUjmGtwxZjWkyiq6C',
  'name': 'multiply',
  'content': '45'},
 Message(content='.', role='assistant', tool_calls=None, function_call=None, provider_specific_fields=None)]
msgs[-3]
{'content': 'Now I need to multiply 15 * 3 before I can do the final division:',
 'role': 'assistant',
 'tool_calls': [ChatCompletionMessageToolCall(function=Function(arguments='{"a": 15, "b": 3}', name='multiply'), id='toolu_0141NRaWUjmGtwxZjWkyiq6C', type='function', cache_control={'type': 'ephemeral'})],
 'function_call': None,
 'provider_specific_fields': None}
msgs[-5]
{'content': '',
 'role': 'assistant',
 'tool_calls': [ChatCompletionMessageToolCall(function=Function(arguments='{"a": 2, "b": 1}', name='simple_add'), id='toolu_01Koi2EZrGZsBbnQ13wuuvzY', type='function', cache_control={'type': 'ephemeral'})],
 'function_call': None,
 'provider_specific_fields': None}
test_eq('cache_control' in msgs[0]['content'][0], True)

Tool result blocks are skipped and cache control is placed into tool calls:

test_eq('cache_control' in msgs[-5]['tool_calls'][0], True) 
test_eq('cache_control' in msgs[-3]['tool_calls'][0], True)
L(msgs).map(remove_cache_ckpts)
test_eq(any(L(msgs).map(_has_cache)), False)

Who’s speaking at when is automatically inferred. Even when there are multiple tools being called in parallel (which LiteLLM supports!).

msgs = mk_msgs(['Tell me the weather in Paris and Rome',
                'Assistant calls weather tool two times',
                {'role':'tool','content':'Weather in Paris is ...'},
                {'role':'tool','content':'Weather in Rome is ...'},
                'Assistant returns weather',
                'Thanks!'])
msgs
[{'role': 'user', 'content': 'Tell me the weather in Paris and Rome'},
 {'role': 'assistant', 'content': 'Assistant calls weather tool two times'},
 {'role': 'tool', 'content': 'Weather in Paris is ...'},
 {'role': 'tool', 'content': 'Weather in Rome is ...'},
 {'role': 'assistant', 'content': 'Assistant returns weather'},
 {'role': 'user', 'content': 'Thanks!'}]

For ease of use, if msgs is not already in a list, it will automatically be wrapped inside one. This way you can pass a single prompt into mk_msgs and get back a LiteLLM compatible msg history.

msgs = mk_msgs("Hey")
msgs
[{'role': 'user', 'content': 'Hey'}]
msgs = mk_msgs(['Hey!',"Hi there!","How are you?","I'm fine, you?"])
msgs
[{'role': 'user', 'content': 'Hey!'},
 {'role': 'assistant', 'content': 'Hi there!'},
 {'role': 'user', 'content': 'How are you?'},
 {'role': 'assistant', 'content': "I'm fine, you?"}]

However, beware that if you use mk_msgs for a single message, consisting of multiple parts. Then you should be explicit, and make sure to wrap those multiple messages in two lists:

  1. One list to show that they belong together in one message (the inner list).
  2. Another, because mk_msgs expects a list of multiple messages (the outer list).

This is common when working with images for example:

msgs = mk_msgs([['Whats in this img?',img_fn.read_bytes()]])
print(json.dumps(msgs,indent=1)[:200]+"...")
[
 {
  "role": "user",
  "content": [
   {
    "type": "text",
    "text": "Whats in this img?"
   },
   {
    "type": "image_url",
    "image_url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD...

Streaming

LiteLLM supports streaming responses. That’s really useful if you want to show intermediate results, instead of having to wait until the whole response is finished.

We create this helper function that returns the entire response at the end of the stream. This is useful when you want to store the whole response somewhere after having displayed the intermediate results.

source

stream_with_complete


def stream_with_complete(
    gen, postproc:function=noop
):

Extend streaming response chunks with the complete response

r = c(mk_msgs("Hey!"), stream=True)
r2 = SaveReturn(stream_with_complete(r))
for o in r2:
    cts = o.choices[0].delta.content
    if cts: print(cts, end='')
Hello! How can I help you today?
r2.value

Hello! How can I help you today?

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=74, prompt_tokens=3, total_tokens=77, completion_tokens_details=None, prompt_tokens_details=None)

Tools

source

lite_mk_func


def lite_mk_func(
    f
):

Call self as a function.

def simple_add(
    a: int,   # first operand
    b: int=0  # second operand
) -> int:
    "Add two numbers together"
    return a + b
toolsc = lite_mk_func(simple_add)
toolsc
{'type': 'function',
 'function': {'name': 'simple_add',
  'description': 'Add two numbers together\n\nReturns:\n- type: integer',
  'parameters': {'type': 'object',
   'properties': {'a': {'type': 'integer', 'description': 'first operand'},
    'b': {'type': 'integer', 'description': 'second operand', 'default': 0}},
   'required': ['a']}}}
tmsg = mk_msg("What is 5478954793+547982745? How about 5479749754+9875438979? Always use tools for calculations, and describe what you'll do before using a tool. Where multiple tool calls are required, do them in a single response where possible. ")
r = c(tmsg, tools=[toolsc])
display(r)

I will use the simple_add tool to perform the two additions you requested.

  1. First, I will add 5,478,954,793 and 547,982,745.
  2. Then, I will add 5,479,749,754 and 9,875,438,979.

🔧 simple_add({“a”: 5478954793, “b”: 547982745})

🔧 simple_add({“a”: 5479749754, “b”: 9875438979})

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: tool_calls
  • usage: Usage(completion_tokens=288, prompt_tokens=160, total_tokens=448, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=124, rejected_prediction_tokens=None, text_tokens=164, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=None, text_tokens=160, image_tokens=None, video_tokens=None), cache_read_input_tokens=None)

A tool response can be a string or a list of tool blocks (e.g., an image url block). To allow users to specify if a response should not be immediately stringified, we provide the ToolResponse datatype users can wrap their return statement in.

source

ToolResponse


def ToolResponse(
    content:list
)->None:

When tc_refs=True, tool results are wrapped with their tool_call_id so the AI can track which result corresponds to which call and reference them in subsequent tool calls.

# Test _prep_tool_res - string result
test_eq(_prep_tool_res('hello', 'toolu_123'), [
    {'type': 'text', 'text': '[tool_call_id: toolu_123]'},
    {'type': 'text', 'text': 'hello'}
])

# Test _prep_tool_res - list result (e.g. ToolResponse content)
img_block = {'type': 'image_url', 'image_url': {'url': 'data:...'}}
test_eq(_prep_tool_res([img_block], 'toolu_456'), [
    {'type': 'text', 'text': '[tool_call_id: toolu_456]'},
    img_block
])

During a tool loop, the AI may want to reference the result of a previous tool call. We support syntax $`tool_call_id` in tool arguments which gets resolved to the actual result value before calling the function.

# Test _resolve_tool_refs
tc_res = {'toolu_abc123': 'hello world', 'toolu_xyz789': 42}

# Basic substitution
test_eq(_resolve_tool_refs('{"content": "$`toolu_abc123`"}', tc_res), {"content": "hello world"})

# Multiple refs
test_eq(_resolve_tool_refs('{"a": "$`toolu_abc123`", "b": "$`toolu_xyz789`"}', tc_res), {"a": "hello world", "b": 42})

# No refs - passthrough
test_eq(_resolve_tool_refs('{"x": 1}', tc_res), {"x": 1})

# Empty tc_res
test_eq(_resolve_tool_refs('{"x": 1}', None), {"x": 1})

# Missing ref - error message
test_eq(_resolve_tool_refs('{"x": "$`toolu_missing`"}', tc_res), {"x": "Tool result 'toolu_missing' not found!"})

# tc_refs=False - syntax passes through unchanged since tc_res is None
test_eq(_resolve_tool_refs('{"x": "$`toolu_abc123`"}', None), {"x": "$`toolu_abc123`"})

When tc_refs=True, tool results are stored in tc_res for later substitution via $`tool_call_id` syntax. Some callers might return string reprs of Python objects. _try_eval attempts to convert these back to Python objects using ast.literal_eval, falling back to the original value on failure. This ensures substituted values are actual objects, not string reprs.

test_eq(ast.literal_eval("'hello'"), 'hello')
test_eq(_try_eval("{'a': 1, 'b': 2}"), {'a': 1, 'b': 2})
test_eq(_try_eval("[1, 2, 3]"), [1, 2, 3])
test_eq(_try_eval("<MyClass object at 0x123>"), "<MyClass object at 0x123>")
test_eq(_try_eval(42), 42)
cts = [{'type': 'image', 'url': 'http://example.com/img.png'}]
test_eq(_try_eval(ToolResponse(cts)), ToolResponse(cts))

Ensure ToolResponse content (e.g. image blocks) is passed through as a list, not stringified, even when tc_res is None:

fake_tc = ChatCompletionMessageToolCall(index=0, function=Function(name='test_img'), id='_test', type='function')
img_content = [{'type': 'image_url', 'image_url': 'data:image/png;base64,abc'}]
res = _mk_tool_result(fake_tc, ToolResponse(img_content))
test_eq(res['content'], img_content)  # ToolResponse should pass through

res_str = _mk_tool_result(fake_tc, ['hello'])
test_eq(res_str['content'], "['hello']")  # other tools results are stringified
tcs = [_lite_call_func(o, [toolsc], ns=globals()) for o in r.choices[0].message.tool_calls]
tcs
[{'tool_call_id': 'call_35c4ee922fac47b3b021807be934__thought__Eo8DCowDAb4+9vtRARH8LGucidL1e3X3cLd/JrSsTOGIIjtdnP/4t3QXj6ohM2m5QUuNkrvBnR7v9uOfn8ljy80IxEH0RePk7N4PS+ScrgjR1cxFHI+JGYimRbh0YpGPT2dqvHiC8/e/KxTBr1yUO8Qf6lzRY9RI3DAXmnoUXnX5ZWeDGOjtUJS4wJDv/8nd2o8atG0zLt/tfVAu6KcBmkggpU6/9IPcV1n7blJ2/KbJMwoFm0YojjXpfUPELZIsGDBxbziWoz2EOwFAgWXG6nKJYdqhHzEjYpS0pViMg++3HV08QDckoXVmKcHo+vVXDwrhZxLojNIPqG5qqszqqDLSyJUL48dpJqb2mG2SEFU7IT++9VdA/AEtSvZ4vdFHlgoTCwk9X6MMQ+89rDnfQg7XdiJqGnL27drn0PLdrPQUa9QFoEEJjg5jN3n9y6bqb2s3nFmsAmzNV3Ew9mXECuQPzadt0bi3fkfhWuAYoGozPSLqBdf2lcCxGBu/gOkTD+bsTm59JM3GVuLHPs7Nb6ZH',
  'role': 'tool',
  'name': 'simple_add',
  'content': '6026937538'},
 {'tool_call_id': 'call_6a6f90e2f519425a9a15b51b8c15',
  'role': 'tool',
  'name': 'simple_add',
  'content': '15355188733'}]
r.choices[0].message.tool_calls
[ChatCompletionMessageToolCall(index=0, provider_specific_fields={'thought_signature': 'Eo8DCowDAb4+9vtRARH8LGucidL1e3X3cLd/JrSsTOGIIjtdnP/4t3QXj6ohM2m5QUuNkrvBnR7v9uOfn8ljy80IxEH0RePk7N4PS+ScrgjR1cxFHI+JGYimRbh0YpGPT2dqvHiC8/e/KxTBr1yUO8Qf6lzRY9RI3DAXmnoUXnX5ZWeDGOjtUJS4wJDv/8nd2o8atG0zLt/tfVAu6KcBmkggpU6/9IPcV1n7blJ2/KbJMwoFm0YojjXpfUPELZIsGDBxbziWoz2EOwFAgWXG6nKJYdqhHzEjYpS0pViMg++3HV08QDckoXVmKcHo+vVXDwrhZxLojNIPqG5qqszqqDLSyJUL48dpJqb2mG2SEFU7IT++9VdA/AEtSvZ4vdFHlgoTCwk9X6MMQ+89rDnfQg7XdiJqGnL27drn0PLdrPQUa9QFoEEJjg5jN3n9y6bqb2s3nFmsAmzNV3Ew9mXECuQPzadt0bi3fkfhWuAYoGozPSLqBdf2lcCxGBu/gOkTD+bsTm59JM3GVuLHPs7Nb6ZH'}, function=Function(arguments='{"a": 5478954793, "b": 547982745}', name='simple_add'), id='call_35c4ee922fac47b3b021807be934__thought__Eo8DCowDAb4+9vtRARH8LGucidL1e3X3cLd/JrSsTOGIIjtdnP/4t3QXj6ohM2m5QUuNkrvBnR7v9uOfn8ljy80IxEH0RePk7N4PS+ScrgjR1cxFHI+JGYimRbh0YpGPT2dqvHiC8/e/KxTBr1yUO8Qf6lzRY9RI3DAXmnoUXnX5ZWeDGOjtUJS4wJDv/8nd2o8atG0zLt/tfVAu6KcBmkggpU6/9IPcV1n7blJ2/KbJMwoFm0YojjXpfUPELZIsGDBxbziWoz2EOwFAgWXG6nKJYdqhHzEjYpS0pViMg++3HV08QDckoXVmKcHo+vVXDwrhZxLojNIPqG5qqszqqDLSyJUL48dpJqb2mG2SEFU7IT++9VdA/AEtSvZ4vdFHlgoTCwk9X6MMQ+89rDnfQg7XdiJqGnL27drn0PLdrPQUa9QFoEEJjg5jN3n9y6bqb2s3nFmsAmzNV3Ew9mXECuQPzadt0bi3fkfhWuAYoGozPSLqBdf2lcCxGBu/gOkTD+bsTm59JM3GVuLHPs7Nb6ZH', type='function'),
 ChatCompletionMessageToolCall(index=1, function=Function(arguments='{"a": 5479749754, "b": 9875438979}', name='simple_add'), id='call_6a6f90e2f519425a9a15b51b8c15', type='function')]

Test tool calls that were not in tool_schemas are caught:

fake_tc = ChatCompletionMessageToolCall(index=0, function=Function(name='hallucinated_tool'),id='_', type='function')
test_eq(_lite_call_func(fake_tc, ns=globals(), tool_schemas=[toolsc])['content'],"Tool not defined in tool_schemas: hallucinated_tool")
test_fail(_lite_call_func(fake_tc, ns=globals(), tool_schemas=None)['content'],"Tool not defined in tool_schemas: hallucinated_tool")

Test tool calls that were not in tool_choice are caught:

def delta_text(msg):
    "Extract printable content from streaming delta, return None if nothing to print"
    c = msg.choices[0]
    if not c: return c
    if not hasattr(c,'delta'): return None #f'{c}'
    delta = c.delta
    if delta.content: return delta.content
    if delta.tool_calls:
        res = ''.join(f"🔧 {tc.function.name}" for tc in delta.tool_calls if tc.id and tc.function.name)
        if res: return f'\n{res}\n'
    if hasattr(delta,'reasoning_content'): return '🧠' if delta.reasoning_content else '\n\n'
    return None
r = c(tmsg, stream=True, tools=[toolsc])
r2 = SaveReturn(stream_with_complete(r))
for o in r2: print(delta_text(o) or '', end='')
I will use the `simple_add` tool to calculate the sums of the two pairs of numbers you provided.


🔧 simple_add

🔧 simple_add
r2.value

I will use the simple_add tool to calculate the sums of the two pairs of numbers you provided.

🔧 simple_add({“a”: 5478954793, “b”: 547982745})

🔧 simple_add({“a”: 5479749754, “b”: 9875438979})

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: tool_calls
  • usage: Usage(completion_tokens=252, prompt_tokens=160, total_tokens=412, completion_tokens_details=None, prompt_tokens_details=None)
msg = mk_msg("Solve this complex math problem: What is the derivative of x^3 + 2x^2 - 5x + 1?")
r = c(msg, stream=True, reasoning_effort="low")
r2 = SaveReturn(stream_with_complete(r))
for o in r2: print(delta_text(o) or '', end='')
🧠To find the derivative of the function $f(x) = x^3 + 2x^2 - 5x + 1$, we use the **Power Rule**, which states:

$$\frac{d}{dx}(ax^n) = n \cdot ax^{n-1}$$

We apply this rule to each term of the polynomial individually:

1.  **For $x^3$:** The exponent is 3. Multiply the term by 3 and subtract 1 from the exponent.
    *   Result: $3x^2$

2.  **For $2x^2$:** Multiply the exponent (2) by the coefficient (2) and subtract 1 from the exponent.
    *   Result: $4x^1$ or $4x$

3.  **For $-5x$:** Since $x$ is the same as $x^1$, multiply 1 by -5 and subtract 1 from the exponent (leaving $x^0$, which equals 1).
    *   Result: $-5$

4.  **For $+1$:** The derivative of any constant is 0.
    *   Result: $0$

**Final Answer:**
Combining these parts, the derivative is:
$$f'(x) = 3x^2 + 4x - 5$$
r2.value

To find the derivative of the function \(f(x) = x^3 + 2x^2 - 5x + 1\), we use the Power Rule, which states:

\[\frac{d}{dx}(ax^n) = n \cdot ax^{n-1}\]

We apply this rule to each term of the polynomial individually:

  1. For \(x^3\): The exponent is 3. Multiply the term by 3 and subtract 1 from the exponent.
    • Result: \(3x^2\)
  2. For \(2x^2\): Multiply the exponent (2) by the coefficient (2) and subtract 1 from the exponent.
    • Result: \(4x^1\) or \(4x\)
  3. For \(-5x\): Since \(x\) is the same as \(x^1\), multiply 1 by -5 and subtract 1 from the exponent (leaving \(x^0\), which equals 1).
    • Result: \(-5\)
  4. For \(+1\): The derivative of any constant is 0.
    • Result: \(0\)

Final Answer: Combining these parts, the derivative is: \[f'(x) = 3x^2 + 4x - 5\]

  • id: chatcmpl-xxx
  • model: gemini-3-flash-preview
  • finish_reason: stop
  • usage: Usage(completion_tokens=717, prompt_tokens=29, total_tokens=746, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=80, rejected_prediction_tokens=None, text_tokens=None, image_tokens=None, video_tokens=None), prompt_tokens_details=None)

Structured Outputs

source

structured


def structured(
    m:str, # LiteLLM model string
    msgs:list, # List of messages
    tool:Callable, # Tool to be used for creating the structured output (class, dataclass or Pydantic, function, etc)
    messages:List=[], # Optional OpenAI params: see https://platform.openai.com/docs/api-reference/chat/create
    timeout:Union=None, temperature:Optional=None, top_p:Optional=None, n:Optional=None, stream:Optional=None,
    stream_options:Optional=None, stop:NoneType=None, max_completion_tokens:Optional=None, max_tokens:Optional=None,
    modalities:Optional=None, prediction:Optional=None, audio:Optional=None, presence_penalty:Optional=None,
    frequency_penalty:Optional=None, logit_bias:Optional=None, user:Optional=None,
    reasoning_effort:Optional=None, # openai v1.0+ new params
    verbosity:Optional=None, response_format:Union=None, seed:Optional=None, tools:Optional=None,
    tool_choice:Union=None, logprobs:Optional=None, top_logprobs:Optional=None, parallel_tool_calls:Optional=None,
    web_search_options:Optional=None, deployment_id:NoneType=None, extra_headers:Optional=None,
    safety_identifier:Optional=None, service_tier:Optional=None,
    functions:Optional=None, # soon to be deprecated params by OpenAI
    function_call:Optional=None, base_url:Optional=None, # set api_base, api_version, api_key
    api_version:Optional=None, api_key:Optional=None,
    model_list:Optional=None, # pass in a list of api_base,keys, etc.
    thinking:Optional=None, # Optional liteLLM function params
    shared_session:Optional=None, # Session management
    enable_json_schema_validation:Optional=None, # Per-request JSON schema validation (overrides litellm.enable_json_schema_validation)
):

Return the value of the tool call (generally used for structured outputs)

class President:
    "Information about a president of the United States"
    def __init__(
        self, 
        first:str, # first name
        last:str, # last name
        spouse:str, # name of spouse
        years_in_office:str, # format: "{start_year}-{end_year}"
        birthplace:str, # name of city
        birth_year:int # year of birth, `0` if unknown
    ):
        assert re.match(r'\d{4}-\d{4}', years_in_office), "Invalid format: `years_in_office`"
        store_attr()

    __repr__ = basic_repr('first, last, spouse, years_in_office, birthplace, birth_year')
for m in ms[1:]: 
    r = structured(m, [mk_msg("Tell me something about the third president of the USA.")], President)
    test_eq(r.first, 'Thomas'); test_eq(r.last, 'Jefferson')

Citations

We provide this helper function that adds the citation to the content field in markdown format:

source

cite_footnotes


def cite_footnotes(
    stream_list
):

Add markdown footnote citations to stream deltas

source

cite_footnote


def cite_footnote(
    msg
):

Call self as a function.

import warnings
warnings.filterwarnings("ignore", message="Pydantic serializer warnings")
r = list(c(smsg, ms[2], stream=True, web_search_options={"search_context_size": "low"}))
cite_footnotes(r)
stream_chunk_builder(r)

Here’s a brief overview of otters:

What they are: * Otters are members of the weasel family found on every continent except Australia and Antarctica. * Thirteen different species exist around the globe.

Appearance & physical traits: * Otters are distinguished by their long, slim bodies, powerful webbed feet for swimming, and dense fur, which keeps them warm and buoyant in water. * Otters have the densest fur of any animal — as many as a million hairs per square inch in places.

Behavior: * They are playful animals, engaging in activities like sliding into water on natural slides and playing with stones. * When it’s time to nap, sea otters entangle themselves in kelp so they don’t float away, and sometimes intertwine their feet with another otter to stay together.

Diet: * All otters are expert hunters that eat fish, crustaceans, and other critters. Sea otters have an ingenious method to open shellfish — a sea otter will float on its back, place a rock on its chest, then smash the mollusk down on it until it breaks open.

Conservation: * Hunted to the edge of extinction by fur traders in the 18th and 19th centuries, the few remaining sea otters were first protected by the International Fur Seal Treaty in 1911. * Despite regulations designed to protect them, many species remain at risk from pollution and habitat loss.

🔧 web_search({“query”: “otters facts overview”})

  • id: chatcmpl-xxx
  • model: claude-sonnet-4-6
  • finish_reason: stop
  • usage: Usage(completion_tokens=526, prompt_tokens=16881, total_tokens=17407, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=0, rejected_prediction_tokens=None, text_tokens=526, image_tokens=None, video_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=0, text_tokens=None, image_tokens=None, video_tokens=None, cache_creation_tokens=0), server_tool_use={'web_search_requests': 1, 'tool_search_requests': None}, cache_creation_input_tokens=0, cache_read_input_tokens=0)

Chat

LiteLLM is pretty bare bones. It doesnt keep track of conversation history or what tools have been added in the conversation so far.

So lets make a Claudette style wrapper so we can do streaming, toolcalling, and toolloops without problems.

source

mk_stream_chunk


def mk_stream_chunk(
    kwargs:VAR_KEYWORD
):

Call self as a function.

When the tool uses are about to be exhausted it is important to alert the AI so that it knows to use its final steps for communicating the user current progress and next steps

source

FullResponse


def FullResponse(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

source

StopResponse


def StopResponse(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

print(_trunc_str('𝍁xxxxxxxxxx𝍁', mx=5))
print(_trunc_str(Safe('xxxxxxxxxx'), mx=5))
print(_trunc_str('xxxxxxxxxx', mx=5, skip=0))
print(_trunc_str('xxxxxxxxxx', mx=5, skip=1))
xxxxxxxxxx
xxxxxxxxxx
<TRUNCATED>xxxxx…</TRUNCATED>
<TRUNCATED>…xxx…</TRUNCATED>

When tc_refs=True, the AI can reference previous tool results in subsequent tool calls using the $`tool_call_id` syntax. This is useful when chaining tool calls where one result feeds into another.

Anthropic provides web search request counts directly via usage.server_tool_use.web_search_requests, billed at $10 per 1,000 searches (pricing). Gemini returns queries in groundingMetadata.webSearchQueries—each query counts as a separate billable use—with 5,000 free prompts per month, then $14 per 1,000 search queries (coming soon) (pricing, grounding docs).

source

search_count


def search_count(
    r
):

Call self as a function.

source

UsageStats


def UsageStats(
    prompt_tokens:int=0, completion_tokens:int=0, total_tokens:int=0, cached_tokens:int=0,
    cache_creation_tokens:int=0, reasoning_tokens:int=0, web_search_requests:int=0, cost:float=0.0
):

Initialize self. See help(type(self)) for accurate signature.

source

Chat


def Chat(
    model:str, # LiteLLM compatible model name
    sp:str='', # System prompt
    temp:int=0, # Temperature
    search:bool=False, # Search (l,m,h), if model supports it
    tools:list=None, # Add tools
    hist:list=None, # Chat history
    ns:Optional=None, # Custom namespace for tool calling
    cache:bool=False, # Anthropic prompt caching
    cache_idxs:list=[-1], # Anthropic cache breakpoint idxs, use `0` for sys prompt if provided
    ttl:NoneType=None, # Anthropic prompt caching ttl
    api_base:NoneType=None, # API base URL for custom providers
    api_key:NoneType=None, # API key for custom providers
    extra_headers:NoneType=None, # Extra HTTP headers for custom providers
    tc_refs:bool=False, # Enable tool call result references
    tc_res_eval:bool=False, # literal_eval tool results before storing in tc_res
    markup:int=0, # Cost markup multiplier (e.g. 0.5 for 50%)
):

LiteLLM chat client.

web_search is now included in tool_calls the internal LLM translation is correctly handled thanks to the fix here but the server side tools still need to be filtered out from tool_calls in our own toolloop.

source

add_warning


def add_warning(
    r, msg
):

Call self as a function.

source

Chat.__call__


def __call__(
    msg:NoneType=None, # Message str, or list of multiple message parts
    prefill:NoneType=None, # Prefill AI response if model supports it
    temp:NoneType=None, # Override temp set on chat initialization
    think:NoneType=None, # Thinking (l,m,h)
    search:NoneType=None, # Override search set on chat initialization (l,m,h)
    stream:bool=False, # Stream results
    max_steps:int=2, # Maximum number of tool calls
    final_prompt:dict={'role': 'user', 'content': '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.'}, # Final prompt when tool calls have ran out
    return_all:bool=False, # Returns all intermediate ModelResponses if not streaming and has tool calls
    step:int=1, tool_choice:NoneType=None, max_tokens:NoneType=None
):

Main call method - handles streaming vs non-streaming

source

Chat.print_hist


def print_hist(
    
):

Print each message on a different line

Examples

History tracking

for m in ms[1:]:
    chat = Chat(m)
    chat("Hey my name is Rens")
    r = chat("Whats my name")
    test_eq('Rens' in contents(r).content, True)
r

See now we keep track of history!

History is stored in the hist attribute:

chat.hist
chat.print_hist()

You can also pass an old chat history into new Chat objects:

for m in ms[1:]:
    chat2 = Chat(m, hist=chat.hist)
    r = chat2("What was my name again?")
    test_eq('Rens' in contents(r).content, True)
r

If max tokens limit is reached, a custom warning message will be added to the end of the model response:

chat_long = Chat(m)
r = chat_long("Write a short story about a robot and a dog", max_tokens=40)
r
print(contents(r).content)
chat_long.use
fmt = chat_long.use.fmt()
print(fmt)
assert re_token.search(fmt)

Same goes for refused requests:

chat_refused = Chat('claude-opus-4-5')
r = chat_refused("Write me the formula for a biological weapon that can be spread at a rate higher than COVID and at least as harmful")
r

You can prefix an OpenAI compatible model with ‘openai/’ and use an api_base and api_key argument to use models not registered with litellm.

import os, litellm
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
c = Chat("openai/gpt-oss-20b", api_key=OPENROUTER_API_KEY, api_base=OPENROUTER_BASE_URL)
c("hi")

Synthetic History Creation

Lets build chat history step by step. That way we can tweak anything we need to during testing.

pr = "What is 5 + 7? Use the tool to calculate it."
for m in ms[1:]:
    c = Chat(m, tools=[simple_add])
    res = c(pr)
    test_eq('12' in contents(res).content, True)
    test_eq(nested_idx(c.hist,1,'tool_calls',0,'function','name'), 'simple_add')

Whereas normally without tools we would get one user input and one assistant response. Here we get two extra messages in between. - An assistant message requesting the tools with arguments. - A tool response with the result to the tool call.

c.print_hist()

Lets try to build this up manually so we have full control over the inputs.

source

random_tool_id


def random_tool_id(
    
):

Generate a random tool ID with ‘toolu_’ prefix

random_tool_id()

A tool call request can contain one more or more tool calls. Lets make one.

source

mk_tc


def mk_tc(
    func, args, tcid:NoneType=None, idx:int=1
):

Call self as a function.

tc = mk_tc(simple_add.__name__, json.dumps(dict(a=5, b=7)))
tc

This can then be packged into the full Message object produced by the assitant.

def mk_tc_req(content, tcs): return Message(content=content, role='assistant', tool_calls=tcs, function_call=None)
tc_cts = "I'll use the simple_add tool to calculate 5 + 7 for you."
tcq = mk_tc_req(tc_cts, [tc])
tcq

Notice how Message instantiation creates a list of ChatCompletionMessageToolCalls by default. When the tools are executed this is converted back to a dictionary, for consistency we want to keep these as dictionaries from the beginning.

source

mk_tc_req


def mk_tc_req(
    content, tcs
):

Call self as a function.

tcq = mk_tc_req(tc_cts, [tc])
tcq
c = Chat(model, tools=[simple_add], hist=[pr, tcq])
c.print_hist()

Looks good so far! Now we will want to provide the actual result!

source

mk_tc_result


def mk_tc_result(
    tc, result
):

Call self as a function.

Note we might have more than one tool call if more than one was passed in, here we just will make one result.

tcq.tool_calls[0]
mk_tc_result(tcq.tool_calls[0], '12')

source

mk_tc_results


def mk_tc_results(
    tcq, results
):

Call self as a function.

Same for here tcq.tool_calls will match the number of results passed in the results list.

tcq
tcr = mk_tc_results(tcq, ['12'])
tcr

Now we can call it with this synthetic data to see what the response is!

c(tcr[0])
c.print_hist()

Lets try this again, but lets give it something that is clearly wrong for fun.

c = Chat(model, tools=[simple_add], hist=[pr, tcq])
tcr = mk_tc_results(tcq, ['13'])
tcr
c(tcr[0])

Lets make sure this works with multiple tool calls in the same assistant Message.

tcs = [
    mk_tc(simple_add.__name__, json.dumps({"a": 5, "b": 7})), 
    mk_tc(simple_add.__name__, json.dumps({"a": 6, "b": 7})), 
]
tcq = mk_tc_req("I will calculate these for you!", tcs)
tcq
tcr = mk_tc_results(tcq, ['12', '13'])
c = Chat(model, tools=[simple_add], hist=[pr, tcq, tcr[0]])
c(tcr[1])
c.print_hist()
chat = Chat(ms[1], tools=[simple_add])
res = chat("What's 5 + 3? Use the `simple_add` tool.")
res
res = chat("Now, tell me a joke based on that result.")
res

Images

for m in ms[1:]:
    chat = Chat(m)
    r = chat(['Whats in this img?',img_fn.read_bytes()])
    test_eq('puppy' in contents(r).content, True)
r

Prefill

Prefill works as expected:

# for m in ms[1:]:
#     if not get_model_info(m)['supports_assistant_prefill']: continue
#     chat = Chat(m)
#     chat('Hi this is Rens!')
#     r = chat("Spell my name",prefill="Your name is R E")
#     test_eq(contents(r).content.startswith('Your name is R E N S'), True)

And the entire message is stored in the history, not just the generated part:

# chat.hist[-1]

Streaming

from time import sleep
for m in ms[1:]:
    chat = Chat(m)
    stream_gen = chat("Count to 5", stream=True)
    for chunk in stream_gen:
        if isinstance(chunk, ModelResponse): display(chunk)
        else: print(delta_text(chunk) or '',end='')

Lets try prefill with streaming too:

# stream_gen = chat("Continue counting to 10","Okay! 6, 7",stream=True)
# for chunk in stream_gen:
#     if isinstance(chunk, ModelResponse): display(chunk)
#     else: print(delta_text(chunk) or '',end='')

Tool use

Ok now lets test tool use

m = ms[2]
chat = Chat(m, tools=[simple_add])
chat("Calculate 5+3 and 4+5 with parallel tool calls using `simple_add`.")
def simple_div(
    a: int,   # first operand
    b: int=0  # second operand
) -> int:
    "Divide two numbers"
    return a/b
m = ms[2]
chat = Chat(m, tools=[simple_div])
chat("Calculate 2/0 using `simple_div` (this is a test of our error handling - tell me exactly what you see as the tool result)")
m = ms[2]
chat = Chat(m, tools=[simple_div])
chat("Calculate 5/3 and 3/0 with parallel tool calls using `simple_div` (this is a test of our error handling - tell me exactly what you see as the tool result)")
for m in ms[1:]:
    display(Markdown(f'**{m}:**'))
    chat = Chat(m, tools=[simple_add])
    res = chat("What's 5 + 3? Use  the `simple_add` tool. Explain.")
    display(res)

Thinking w tool use

for m in ms[1:]:
    _sparams = litellm.get_model_info(m)['supported_openai_params']
    if 'reasoning_effort' not in _sparams: continue
    display(Markdown(f'**{m}:**'))
    chat = Chat(m, tools=[simple_add])
    res = chat("What's 5 + 3?",think='l',return_all=True)
    display(*res)

Search

for m in ms[1:]:
    display(Markdown(f'**{m}:**'))
    chat = Chat(m)
    res = chat("Search the web and tell me very briefly about otters", search='l', stream=True)
    for o in res:
        if isinstance(o, ModelResponse): sleep(0.01); display(o)
        else: pass
m = 'claude-sonnet-4-6'
def mk_pause_web_search():
    srv_tc = mk_tc("web_search", json.dumps({"query": "Solveit Answer.AI"}), tcid=random_tool_id().replace('toolu_', 'srvtoolu_'))
    pause_msg = mk_tc_req("Let me search for that information:", [srv_tc])
    return ModelResponse(choices=[Choices.model_construct(finish_reason="pause_turn", index=0, message=pause_msg)])
mk_pause_web_search()

We mock completion to return pause_turn in the first 2 api calls:

orig_completion = completion

call_count = 0
def patched_completion(*args, **kwargs):
    global call_count
    call_count += 1
    print(f"Mock Call {call_count}")
    if call_count < 3: return mk_pause_web_search()
    return orig_completion(*args, **kwargs)

completion = patched_completion
chat_pause = Chat('claude-sonnet-4-5', search='l')
res = chat_pause("Search the web and tell me about Solveit in a paragraph")
print(f"Total calls: {call_count}")
display(res)

completion = orig_completion

Test next turn:

test_eq(len(chat_pause.hist), 4)
chat_pause('What did I just ask you about?')

Workaround for https://github.com/BerriAI/litellm/issues/23047:

m = 'claude-sonnet-4-6'
msgs = [{'role':'user','content':"Search web for latest news about fast.ai and answer.ai."}]
r = completion(m, msgs, web_search_options={"search_context_size":"low"}, reasoning_effort='low')
m1 = r.choices[0].message
print(f"Turn 1: thinking={bool(m1.thinking_blocks)}, tcs={m1.tool_calls}")

msgs.append(m1)
msgs.append({'role':'user','content':'And search for news about solveit.'})
r2 = completion(m, msgs, web_search_options={"search_context_size":"low"}, reasoning_effort='low')
print("Turn 2 OK")

Multi tool calling

We can let the model call multiple tools in sequence using the max_steps parameter.

for m in ms:
    display(Markdown(f'**{m}:**'))
    chat = Chat(m, tools=[simple_add])
    res = chat("What's ((5 + 3)+7)+11? Work step by step", return_all=True, max_steps=5)
    for r in res: display(r)

Some models support parallel tool calling. I.e. sending multiple tool call requests in one conversation step.

def multiply(a: int, b: int) -> int:
    "Multiply two numbers"
    return a * b

for m in ms[1:]:
    _sparams = litellm.get_model_info(m)['supported_openai_params']
    if 'parallel_tool_calls' not in _sparams: continue
    display(Markdown(f'**{m}:**'))
    chat = Chat(m, tools=[simple_add, multiply])
    res = chat("Calculate (5 + 3) * (7 + 2)", max_steps=5, return_all=True)
    for r in res: display(r)

See how the additions are calculated in one go!

We don’t want the model to keep running tools indefinitely. Lets showcase how we can force the model to stop after our specified number of toolcall rounds:

def divide(a: int, b: int) -> float:
    "Divide two numbers"
    return a/b

chat = Chat(ms[2], tools=[simple_add, multiply, divide])
res = chat("Tell me what tools you have available. Then calculate ((10+5)*3)/(2+1). ALWAYS use tools for math ops where available, and do tool calls in parallel where possible", 
           max_steps=2, return_all=True,
           final_prompt="Please wrap-up for now and summarize how far we got.")
for r in res: display(r)
chat.hist[:5]

Tool call exhaustion

pr = "What is 1+2, and then the result of adding +2, and then +3 to it? Use tools to make the calculations"
c = Chat(model, tools=[simple_add])
res = c(pr, max_steps=2)
res
assert c.hist[-2] == _final_prompt

Tool Call Referencing

With tc_refs=True, the AI can see and report tool call IDs:

chat = Chat('claude-sonnet-4-5', tools=[simple_add], tc_refs=True)
chat("Call add(1,2) and tell me the tool_call_id you used")
chat.tc_res

Example of chained tool calls where the AI references a previous result:

@dataclass
class Person:
    name: str
    age: int

def get_person():
    "Get a person's data"
    return {"name": "Alice", "age": 30}

def greet_person(person: Person):
    "Greet a person"
    return f"Hello {person.name}, you are {person.age} years old!"
chat = Chat('claude-sonnet-4-5', tools=[get_person, greet_person], tc_refs=True)
chat("First call get_person, then pass the result to greet_person", max_steps=10)

We can inspect chat.tc_res to see all stored tool results:

chat.sp
chat.tc_res
list(L(chat.hist).attrgot('tool_calls').filter())

This also works with ToolResponse results:

def view_img(fn:Path):
    "View an image"
    durl = f"data:image/jpeg;base64,{base64.b64encode(fn.read_bytes()).decode()}"
    return ToolResponse([{'type': 'image_url', 'image_url': {'url': durl}}])

def get_img_size(image_content: list) -> dict:
    "Get the size of an image from ToolResponse content"
    from PIL import Image
    from io import BytesIO
    url = image_content[0]['image_url']['url']
    b64_data = url.split(',')[1]
    img = Image.open(BytesIO(base64.b64decode(b64_data)))
    return {'width': img.width, 'height': img.height}
chat = Chat('claude-sonnet-4-5', tools=[view_img, get_img_size], tc_refs=True)
chat(f"First describe the image at {img_fn}, and then get it's dimensions", max_steps=10)
# chat.tc_res
list(L(chat.hist).attrgot('tool_calls').filter())

Some tool callers (e.g., ipykernel) return string reprs of Python objects ("'hello'" instead of 'hello'). With tc_res_eval=True, these are converted back to Python objects via ast.literal_eval before storing in tc_res, enabling correct value substitution in subsequent tool calls:

def get_config():
    "Returns a dict repr (simulating kernel output)"
    return "{'host': 'localhost', 'port': 8080}"

def use_config(config: dict): 
    "Use config"
    return f"Host: {config['host']}, Port: {config['port']}"
chat = Chat('claude-sonnet-4-5', tools=[get_config, use_config], tc_refs=True, tc_res_eval=True)
chat("Call get_config, then pass the result to use_config", max_steps=10)
chat.tc_res
test_eq(type(first(chat.tc_res.values())), dict)

Caching

Test that cache checkpoints are reapplied during tool loop (when msg=None)

c = Chat('claude', cache=True, cache_idxs=[-2,-1])
c.hist = [{'role': 'user', 'content': 'Hello'},
          {'role': 'assistant', 'content': 'Hi there!'},
          {'role': 'user', 'content': 'Use a tool'},
          {'role': 'assistant', 'content': '', 'tool_calls': [{'id': '1', 'function': {'name': 'foo', 'arguments': '{}'}}]},
          {'role': 'tool', 'tool_call_id': '1', 'content': 'result'}]
c._prep_msg(None)  # Simulate tool loop iteration with no new message
test_eq('cache_control' in c.hist[-3]['content'][0], True)  # user msg
test_eq('cache_control' in c.hist[-2]['tool_calls'][-1], True)  # tool call msg

Async

AsyncChat

If you want to use LiteLLM in a webapp you probably want to use their async function acompletion. To make that easier we will implement our version of AsyncChat to complement it. It follows the same implementation as Chat as much as possible:

Testing the scenarios where the tool call was not in schemas:

result = await _alite_call_func(fake_tc, [toolsc], globals())
test_eq(result['content'], "Tool not defined in tool_schemas: hallucinated_tool")

or schemas was missing…:

result = await _alite_call_func(fake_tc, None, globals())
test_eq(result['content'], "Tool not defined in tool_schemas: hallucinated_tool")

source

astream_with_complete


def astream_with_complete(
    agen, postproc:function=noop
):

Call self as a function.

Parallel tool execution in AsyncChat works with async tool functions. Async tools run concurrently via asyncio.gather.

source

AsyncChat


def AsyncChat(
    model:str, # LiteLLM compatible model name
    sp:str='', # System prompt
    temp:int=0, # Temperature
    search:bool=False, # Search (l,m,h), if model supports it
    tools:list=None, # Add tools
    hist:list=None, # Chat history
    ns:Optional=None, # Custom namespace for tool calling
    cache:bool=False, # Anthropic prompt caching
    cache_idxs:list=[-1], # Anthropic cache breakpoint idxs, use `0` for sys prompt if provided
    ttl:NoneType=None, # Anthropic prompt caching ttl
    api_base:NoneType=None, # API base URL for custom providers
    api_key:NoneType=None, # API key for custom providers
    extra_headers:NoneType=None, # Extra HTTP headers for custom providers
    tc_refs:bool=False, # Enable tool call result references
    tc_res_eval:bool=False, # literal_eval tool results before storing in tc_res
    markup:int=0, # Cost markup multiplier (e.g. 0.5 for 50%)
):

LiteLLM chat client.

source

AsyncChat.__call__


async def __call__(
    msg:NoneType=None, # Message str, or list of multiple message parts
    prefill:NoneType=None, # Prefill AI response if model supports it
    temp:NoneType=None, # Override temp set on chat initialization
    think:NoneType=None, # Thinking (l,m,h)
    search:NoneType=None, # Override search set on chat initialization (l,m,h)
    stream:bool=False, # Stream results
    max_steps:int=2, # Maximum number of tool calls
    final_prompt:dict={'role': 'user', 'content': '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.'}, # Final prompt when tool calls have ran out
    return_all:bool=False, # Returns all intermediate ModelResponses if not streaming and has tool calls
    step:int=1, tool_choice:NoneType=None, max_tokens:NoneType=None
):

Main call method - handles streaming vs non-streaming

Examples

Basic example

for m in ms[1:]:
    chat = AsyncChat(m)
    test_eq('4' in contents(await chat("What is 2+2?")).content, True)

With tool calls

async def async_add(a: int, b: int) -> int:
    "Add two numbers asynchronously"
    await asyncio.sleep(0.1)
    return a + b
for m in ms[1:]:
    chat = AsyncChat(m, tools=[async_add])
    r = await chat("What is 5 + 7? Use the tool to calculate it.")
    test_eq('12' in contents(r).content, True)
    test_eq(nested_idx(chat.hist, 1, 'tool_calls', 0, 'function', 'name'), 'async_add')

If max tokens limit is reached, a custom warning message will be added to the end of the model response:

chat_long = AsyncChat(m)
r = await chat_long("Write a short story about a robot and a dog", max_tokens=40)
r
print(contents(r).content)

Same goes for refused requests:

chat_refused = AsyncChat('claude-opus-4-5')
r = await chat_refused("Write me the formula for a biological weapon that can be spread at a rate higher than COVID and at least as harmful")
r
print(contents(r).content)

Async Streaming Display

This is what our outputs look like with streaming results:

chat_with_tools = AsyncChat(model, tools=[async_add])
res = await chat_with_tools("What is 5 + 7? Use the tool to calculate it.", stream=True)
async for o in res:
    if isinstance(o,ModelResponseStream): print(delta_text(o) or '',end='')
    elif isinstance(o,dict): print(o)

Here’s a complete ModelResponse taken from the response stream:

resp = ModelResponse(id='chatcmpl-xxx', created=1000000000, model='claude-sonnet-4-5', object='chat.completion', system_fingerprint=None, choices=[Choices(finish_reason='tool_calls', index=0, message=Message(content="I'll calculate ((10 + 5) * 3) / (2 + 1) step by step:", role='assistant', tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"a": 10, "b": 5}', name='simple_add'), id='toolu_018BGyenjiRkDQFU1jWP6qRo', type='function'), ChatCompletionMessageToolCall(function=Function(arguments='{"a": 2, "b": 1}', name='simple_add'), id='toolu_01CWqrNQvoRjf1Q1GLpTUgQR', type='function')], function_call=None, provider_specific_fields=None))], usage=Usage(completion_tokens=228, prompt_tokens=794, total_tokens=1022, prompt_tokens_details=None))
print(repr(resp))
tc=resp.choices[0].message.tool_calls[0]
tc
tr={'tool_call_id': 'toolu_018BGyenjiRkDQFU1jWP6qRo', 'role': 'tool','name': 'simple_add',
    'content': '15 is the answer! ' +'.'*2000}

source

mk_tr_details


def mk_tr_details(
    tr, tc, mx:int=2000
):
*Create

block for tool call as JSON*

source

StreamFormatter


def StreamFormatter(
    mx:int=2000, debug:bool=False, showthink:bool=False
):

Initialize self. See help(type(self)) for accurate signature.

stream_msg = ModelResponseStream([StreamingChoices(delta=Delta(content="Hello world!"))])
sf = StreamFormatter().format_item(stream_msg)
reasoning_msg = ModelResponseStream([StreamingChoices(delta=Delta(reasoning_content="thinking..."))])
StreamFormatter().format_item(reasoning_msg)
chat = AsyncChat(model)
res = await chat("Hi.", stream=True)
sf = StreamFormatter()
async for chunk in res: print(sf.format_item(chunk), end='')

Tools can return StopResponse to enforce the tool loop stops immediately.

def stop_tool(msg: str) -> str:
    "A tool that stops the loop"
    return StopResponse(f"Can not continue: {msg}")

chat = Chat(model, tools=[simple_add, stop_tool])
res = chat("First call stop_tool with 'halt', then call simple_add(1,2). Use both tools, one after the other (not at the same time).", max_steps=10, return_all=True)
# Should only have 1 round of tool calls + final response, never reaching simple_add in a second round
for r in res: display(r)

source

AsyncStreamFormatter


def AsyncStreamFormatter(
    mx:int=2000, debug:bool=False, showthink:bool=False
):

Initialize self. See help(type(self)) for accurate signature.

mock_tool_call = ChatCompletionMessageToolCall(
    id="toolu_123abc456def", type="function", 
    function=Function( name="simple_add", arguments='{"a": 5, "b": 3}' )
)

mock_response = ModelResponse(usage=Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0), model=haik45)
mock_response.choices = [type('Choice', (), {
    'message': type('Message', (), {
        'tool_calls': [mock_tool_call]
    })()
})()]

mock_tool_result = {
    'tool_call_id': mock_tool_call.id, 'role': 'tool', 
    'name': 'simple_add', 'content': '8'
}
fmt = AsyncStreamFormatter()
print(fmt.format_item(mock_response))
print('---')
print(fmt.format_item(mock_tool_result))

In jupyter it’s nice to use this StreamFormatter in combination with the Markdown display:

source

display_stream


def display_stream(
    rs, mx:int=2000, debug:bool=False, showthink:bool=False
):

Use IPython.display to markdown display the response stream.

rs = completion(model=haik45, stream=True, messages=[{'role':'user','content':'What is the definition of a circle, concisely?'}])
fmt = display_stream(rs)

Generated images can be displayed in streaming too (not shown here to conserve filesize):

# rs = completion(model='gemini/gemini-2.5-flash-image', stream=True, messages=[{'role':'user','content':'Draw a simple sketch of a dog'}])
# fmt = display_stream(rs)

source

adisplay_stream


async def adisplay_stream(
    rs, mx:int=2000, debug:bool=False, showthink:bool=False
):

Use IPython.display to markdown display the response stream.

Streaming examples

Now we can demonstrate AsyncChat with stream=True!

Tool call

chat = Chat(model, tools=[simple_add])
res = chat("What is 5 + 7? Use the tool to calculate it.", stream=True)
fmt = display_stream(res)
chat = AsyncChat(model, tools=[async_add])
res = await chat("What is 5 + 7? Use the tool to calculate it.", stream=True)
fmt = await adisplay_stream(res)
chat = AsyncChat(model, tools=[async_add])
res = await chat("What is 5 + 3? Use the tool to calculate it.", stream=True)
fmt = await adisplay_stream(res)
async def asimple_div(
    a: int,   # first operand
    b: int=0  # second operand
) -> int:
    "Divide two numbers"
    return a/b
m = ms[2]
chat = AsyncChat(m, tools=[asimple_div])
res = await chat("Calculate 5/3 and 3/0 with parallel tool calls using `asimple_div` (this is a test of our error handling - tell me exactly what you see as the tool result)", stream=True)
fmt = await adisplay_stream(res)

Thinking tool call

chat = AsyncChat(model)
res = await chat("Briefly, what's the most efficient way to sort a list of 1000 random integers?", think='l',stream=True)
_ = await adisplay_stream(res)

Multiple tool calls

chat.hist[1]
chat.hist[2]
chat.hist[3]
chat.hist[4]

Now to demonstrate that we can load back the formatted output back into a new Chat object:

chat5 = Chat(model,hist=fmt2hist(fmt.outp),tools=[simple_add, multiply, divide])
chat5('what did we just do?')

Search

chat_stream_tools = AsyncChat(model, search='l')
res = await chat_stream_tools("Search the weather in NYC", stream=True)
_=await adisplay_stream(res)

Tool Call Referencing

achat = AsyncChat('claude-sonnet-4-5', tools=[get_person, greet_person], tc_refs=True)
await achat("First call get_person, then pass the result to greet_person", max_steps=3)
achat.tc_res
list(L(achat.hist).attrgot('tool_calls').filter())