pip install git+https://github.com/DrChrisLevy/liteagent.git

uv pip install git+https://github.com/DrChrisLevy/liteagent.git

from dotenv import load_dotenv
load_dotenv()

from liteagent import Agent, Tool, ToolResult
from liteagent.convert import make_default_convert

# Default model for all examples
MODEL = "anthropic/claude-sonnet-4-6"

# The Agent has a built-in default converter (make_default_convert) that:
# - Strips liteagent metadata (usage, timestamp, stop_reason, etc.)
# - Passes everything else through (thinking_blocks, reasoning_content,
#   provider_specific_fields — new litellm fields survive automatically)
# - For OpenAI models: hoists images from tool results into user messages
#   (OpenAI ignores image blocks in tool result content)
#
# Most examples below use the default — no convert_to_llm needed.
# We'll explore the converter in detail in section 16.


# Simple echo tool — reused across examples
async def echo_execute(tool_call_id, params, signal=None, on_update=None):
    return ToolResult(content=[{"type": "text", "text": params["message"]}])


echo_tool = Tool(
    name="echo",
    description="Echo back a message exactly",
    parameters={
        "type": "object",
        "properties": {
            "message": {"type": "string", "description": "The message to echo"}
        },
        "required": ["message"],
    },
    execute=echo_execute,
)

print("Setup complete.")

agent = Agent(
    model=MODEL,
    system_prompt="Be concise. One sentence max.",
)

await agent.prompt("What is 2 + 2?")

agent.messages

agent.state.messages

from pprint import pprint
pprint(agent.state)

agent.state.messages

# The default converter strips liteagent metadata, keeps LLM-compatible fields
convert = make_default_convert(MODEL)
convert(agent.state.messages)

# User message — our input, wrapped in a dict by prompt()
agent.messages[0]

# Assistant message — enriched with usage, stop_reason, timestamp, model, etc.
agent.messages[1]

# This is why we usd @property --> Read-only access. With @property, it prevents replacing the state object itself.
agent.state = "BREAK THE STATE"

unsub = agent.subscribe(my_callback)  # returns unsubscribe function

agent = Agent(
    model=MODEL,
    system_prompt="Be concise.",
)

# Collect all events
events = []
unsub = agent.subscribe(lambda e: events.append(e))

await agent.prompt("Say hello!")

for e in events:
    print(e)

count_before = len(events)
count_before

unsub()  # stop receiving events

agent._subscribers

await agent.prompt("Say goodbye.")

count_after = len(events)
print(f"Events before unsub: {count_before}")
print(f"Events after second prompt: {count_after}")
print(f"Unsubscribe worked: {count_before == count_after}")

agent.state.messages

# Overload 1: string
agent = Agent(model=MODEL)
await agent.prompt("Hello from a string")
agent.messages[0]

# Overload 2: dict (used as-is)
agent = Agent(model=MODEL)
await agent.prompt(
    {"role": "user", "content": "Hello from a dict", "custom_field": "preserved"}
)
agent.messages[0]

convert = make_default_convert(MODEL)
convert(agent.messages)

# Overload 3: list of messages
agent = Agent(model=MODEL, system_prompt="Be concise.")
await agent.prompt(
    [
        {"role": "user", "content": "My name is Alice."},
        {"role": "user", "content": "What is my name?"},
    ]
)

agent.messages

# Overload 4: string + images (multimodal)
# Send a real image and ask the LLM about it

image_block = {
    "type": "image_url",
    "image_url": {
        "url": "https://upload.wikimedia.org/wikipedia/commons/thumb/7/7a/LeBron_James_%2851959977144%29_%28cropped2%29.jpg/250px-LeBron_James_%2851959977144%29_%28cropped2%29.jpg"
    },
}

agent = Agent(
    model=MODEL,
    system_prompt="Be concise. One sentence max.",
)
await agent.prompt("What is in this image?", images=[image_block])

print(f"Response: {agent.messages[-1].get('content')}")

agent.messages

agent.state.is_streaming       # True while loop is running
agent.state.stream_message     # current partial message being streamed (or None)
agent.state.pending_tool_calls # set of tool call IDs currently executing
agent.state.error              # last error message (or None)
agent.state.model              # current model string
agent.state.system_prompt      # current system prompt
agent.state.tools              # current tool list
agent.state.thinking_level     # "off", "minimal", "low", "medium", "high", "xhigh"
agent.messages                 # shorthand for agent.state.messages

agent = Agent(
    model=MODEL,
    system_prompt="Be concise.",
    tools=[echo_tool],
)

# Track state transitions
state_log = []

print(
    f"{'Event':<26} {'Streaming':<10} {'StreamMsg':<10} {'PendTools':<10} {'MsgCount':<10}"
)


def track_state(event):
    t = event["type"]
    entry = {
        "event": t,
        "is_streaming": agent.state.is_streaming,
        "stream_msg": agent.state.stream_message is not None,
        "pending_tools": len(agent.state.pending_tool_calls),
        "msg_count": len(agent.messages),
    }
    state_log.append(entry)
    print("-" * 66)
    print(
        f"{entry['event']:<26} {str(entry['is_streaming']):<10} {str(entry['stream_msg']):<10} {entry['pending_tools']:<10} {entry['msg_count']:<10}"
    )


agent.subscribe(track_state)
await agent.prompt("Echo 'hello world'")

agent = Agent(
    model=MODEL,
    system_prompt="Be concise. Remember everything the user says.",
)

# Turn 1: tell the agent something
await agent.prompt("My favorite color is blue.")
agent.messages

# Turn 2: ask about it — the agent should remember
await agent.prompt("What is my favorite color?")
agent.messages

# Pre-queued steering: steer() before prompt()
agent = Agent(
    model=MODEL,
    system_prompt="Be concise.",
)

agent.steer("Actually, tell me a joke instead.")
await agent.prompt("What is the capital of France?")

agent.messages

# Mid-tool steering: steer() during tool execution
# When tool_a executes, it queues a steering message.
# tool_b should be SKIPPED.

call_log = []
steering_agent = None


async def tool_a_exec(tool_call_id, params, signal=None, on_update=None):
    call_log.append("a")
    steering_agent.steer("Stop! Do something else.")  # interrupt!
    return ToolResult(content=[{"type": "text", "text": "tool_a done"}])


async def tool_b_exec(tool_call_id, params, signal=None, on_update=None):
    call_log.append("b")
    return ToolResult(content=[{"type": "text", "text": "tool_b done"}])


tool_a = Tool(
    name="tool_a",
    description="Tool A",
    parameters={"type": "object", "properties": {}},
    execute=tool_a_exec,
)
tool_b = Tool(
    name="tool_b",
    description="Tool B",
    parameters={"type": "object", "properties": {}},
    execute=tool_b_exec,
)

steering_agent = Agent(
    model=MODEL,
    system_prompt="When asked, call both tool_a and tool_b in a single response. Be concise.",
    tools=[tool_a, tool_b],
)

await steering_agent.prompt("Call both tool_a and tool_b now.")

steering_agent.messages

agent = Agent(
    model=MODEL,
    system_prompt="Be concise. One sentence.",
)

# Queue a follow-up BEFORE the first prompt
agent.follow_up("Now tell me a fun fact about cats.")

await agent.prompt("What is 2 + 2?")

agent.messages

# one-at-a-time (default): queue 3, dequeue returns 1
agent = Agent(model=MODEL)
agent.steer("msg1")
agent.steer("msg2")
agent.steer("msg3")

batch = agent._dequeue_steering()
print(
    f"one-at-a-time: got {len(batch)} message(s), {len(agent._steering_queue)} remaining"
)
print(f"  dequeued: '{batch[0]['content']}'")

# all mode: queue 3, dequeue returns all 3
agent = Agent(model=MODEL, steering_mode="all")
agent.steer("msg1")
agent.steer("msg2")
agent.steer("msg3")

batch = agent._dequeue_steering()
print(f"all mode: got {len(batch)} message(s), {len(agent._steering_queue)} remaining")
for m in batch:
    print(f"  '{m['content']}'")

# Case: continue from a tool result (manually built context)
weather_tool = Tool(
    name="weather",
    description="Get the weather in a location",
    parameters={"type": "object", "properties": {"location": {"type": "string"}}},
    execute=lambda *args: ToolResult(content=[{"type": "text", "text": "72°F and sunny in San Francisco"}]),
)
agent = Agent(
    model=MODEL,
    system_prompt="Be concise.",
    tools=[weather_tool],
)

# Simulate: user asked about weather → assistant called tool → we have the result
agent._state.messages = [
    {"role": "user", "content": "What's the weather?"},
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [
            {
                "id": "c0",
                "type": "function",
                "function": {"name": "weather", "arguments": "{}"},
            }
        ],
        "stop_reason": "tool_calls",
    },
    {
        "role": "tool",
        "tool_call_id": "c0",
        "content": [{"type": "text", "text": "72°F and sunny in San Francisco"}],
        "is_error": False,
    },
]

await agent.continue_run()

agent.messages

# Case: continue from assistant + steering queue
agent = Agent(model=MODEL)
agent._state.messages = [
    {"role": "user", "content": "Hi"},
    {"role": "assistant", "content": "Hello!", "stop_reason": "stop"},
]
agent.steer("Now tell me a joke.")

await agent.continue_run()

agent.messages

# Case: continue from assistant + empty queues → error
agent = Agent(model=MODEL)
agent._state.messages = [
    {"role": "user", "content": "Hi"},
    {"role": "assistant", "content": "Hello!", "stop_reason": "stop"},
]

try:
    await agent.continue_run()
except ValueError as e:
    print(f"Got expected error: {e}")

# abort() after receiving some text — partial should be preserved
agent = Agent(
    model=MODEL,
    system_prompt="Write a very long essay about the history of computing. At least 5000 words.",
)

chunk_count = 0


def abort_after_chunks(event):
    print("signal is set: " + str(agent._signal.is_set()))
    if agent.state.stream_message:
        print(agent.state.stream_message["content"])
    global chunk_count
    if event["type"] == "message_update" and event.get("delta_type") == "text_delta":
        chunk_count += 1
        if chunk_count >= 5:
            agent.abort()


agent.subscribe(abort_after_chunks)
await agent.prompt("Go ahead.")

print(f"is_streaming: {agent.state.is_streaming}")
print(f"signal cleaned up: {agent._signal is None}")

agent.messages

# Some callback triggered a run
agent.follow_up("do something")
asyncio.create_task(agent.continue_run())

# Later, elsewhere:
await agent.wait_for_idle()   # block until that run finishes
# now safe to inspect agent.messages

If you're always doing await agent.prompt() sequentially, you'll never need this.

# When idle, returns immediately
agent = Agent(model=MODEL)
await agent.wait_for_idle()  # should not hang
print("wait_for_idle() returned immediately (agent is idle)")

# After a prompt, also returns immediately (prompt already blocks)
await agent.prompt("Hi")
await agent.wait_for_idle()
print("wait_for_idle() returned immediately (prompt already completed)")

import asyncio

agent = Agent(model=MODEL)

async def background_task():
    agent.follow_up("Summarize what 2+2 is")
    await agent.continue_run()

await agent.prompt("Hi")

# Kick off a run from a background task
asyncio.create_task(background_task())
await asyncio.sleep(0)  # yield to event loop — lets background_task enter _run_loop and set _running_future
await agent.wait_for_idle()

agent.messages

agent = Agent(
    model=MODEL,
    system_prompt="you are a nice agent",
    tools=[echo_tool],
)
agent.append_message({"role": "user", "content": "old message"})
agent.steer("queued steering")
agent.follow_up("queued follow-up")
agent._state.error = "some error"

print("Before clear_messages():")
agent.messages

agent.clear_messages()

print("After clear_messages():")
print(f"  messages: {agent.messages}")
print(f"  queued: {agent.has_queued_messages()}")  # True — queues survived
print(f"  error: {agent.state.error}")              # "some error" — survived

# Now reset — clears everything
agent.append_message({"role": "user", "content": "new message"})
agent.reset()

print("After reset():")
print(
    f"  messages: {len(agent.messages)}, queued: {agent.has_queued_messages()}, error: {agent.state.error}"
)
print(f"  model: {agent.state.model}  ← preserved")
print(f"  system_prompt: '{agent.state.system_prompt}'  ← preserved")
print(f"  tools: {len(agent.state.tools)}  ← preserved")

agent = Agent(model=MODEL)


# Track which model gets called
def event_handler(e):
    if e["type"] == "message_end":
        print(f"Model Used: {agent.state.model}")


agent.subscribe(event_handler)

await agent.prompt("hey")

# Switch model mid-conversation
agent.set_model("gemini/gemini-3-flash-preview")

await agent.prompt("Bye")
agent.messages

# Force an error by using a non-existent model
agent = Agent(model="fake-provider/nonexistent-model")
await agent.prompt("This will fail.")

agent.messages

async def failing_fn(tool_call_id, args, signal, on_update):
    raise ValueError("Something went wrong!")

failing_tool = Tool(
    name="failing_tool",
    description="A tool that always fails",
    parameters={"type": "object", "properties": {}},
    execute=failing_fn,
)

agent = Agent(model=MODEL, tools=[failing_tool], system_prompt="Use the failing_tool when asked.")
await agent.prompt("Use your tool")
agent.messages

from liteagent.convert import make_default_convert

# Build the converter for Anthropic (no image hoisting needed)
convert_anthropic = make_default_convert("anthropic/claude-sonnet-4-6")

# Build one for OpenAI (will hoist tool-result images)
convert_openai = make_default_convert("gpt-5.2")

# Simulate a conversation with enriched messages
messages = [
    {"role": "user", "content": "Hi", "timestamp": 12345},
    {
        "role": "assistant",
        "content": "Hello!",
        "tool_calls": None,
        "thinking_blocks": [{"type": "thinking", "thinking": "greeting"}],
        "reasoning_content": "simple greeting",
        "provider_specific_fields": {"thought_signatures": ["sig123"]},
        "usage": {"prompt_tokens": 10, "completion_tokens": 5},
        "stop_reason": "stop",
        "timestamp": 12346,
    },
    {
        "role": "tool",
        "tool_call_id": "c0",
        "name": "chart",
        "content": [
            {"type": "text", "text": "Here is the chart."},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
        ],
        "is_error": False,
        "details": {"extra": "ui-only"},
        "timestamp": 12347,
    },
]

print("=== Anthropic converter (images stay in tool message) ===")
for m in convert_anthropic(messages):
    print(m)

print()
print("=== OpenAI converter (images hoisted to user message) ===")
for m in convert_openai(messages):
    print(m)

MODELS = [
    "anthropic/claude-sonnet-4-6",
    "anthropic/claude-opus-4-6",
    "gemini/gemini-3-pro-preview",
    "gemini/gemini-3-flash-preview",
    "gpt-5.2",
    "gpt-5.3-codex",
    "gpt-5.4",
]

for model in MODELS:
    agent = Agent(
        model=model,
        system_prompt="Use the echo tool. Be concise.",
        tools=[echo_tool],
    )

    try:
        await agent.prompt("Echo 'test'")
        roles = [m.get("role") for m in agent.messages]
        has_tool = "tool" in roles
        assistants = [m for m in agent.messages if m.get("role") == "assistant"]
        last_content = (assistants[-1].get("content") or "")[:50] if assistants else "?"
        print(f"  ✓ {model:<42} tool_used={has_tool} | {last_content}")
    except Exception as e:
        print(f"  ✗ {model:<42} ERROR: {e}")

# Pattern 1: CLI — print text deltas as they arrive

agent = Agent(
    model=MODEL,
    system_prompt="Be concise.",
)


def cli_handler(event):
    if event["type"] == "message_update" and event.get("delta_type") == "text_delta":
        text = event["delta"].get("content", "")
        print(text, end="", flush=True)
    elif event["type"] == "agent_end":
        print()  # newline at the end


agent.subscribe(cli_handler)
print("Agent: ", end="")
await agent.prompt("What is the meaning of life, in one sentence?")

import base64
import io


def make_bar_chart_b64():
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun"]
    values = [120, 135, 128, 142, 580, 131]
    fig, ax = plt.subplots(figsize=(8, 4))
    ax.bar(months, values, color=["#3498db" if v < 300 else "#e74c3c" for v in values])
    ax.set_title("Monthly API Errors")
    for i, v in enumerate(values):
        ax.text(i, v + 15, str(v), ha="center", fontweight="bold")
    plt.tight_layout()
    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=72)
    plt.close(fig)
    return base64.b64encode(buf.getvalue()).decode()


chart_b64 = make_bar_chart_b64()


async def get_chart_exec(tool_call_id, params, signal=None, on_update=None):
    return ToolResult(
        content=[
            {"type": "text", "text": "Here is the monthly error chart."},
            {
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{chart_b64}"},
            },
        ],
    )


chart_tool = Tool(
    name="get_error_chart",
    description="Get the monthly error chart. Returns text + chart image.",
    parameters={"type": "object", "properties": {}},
    execute=get_chart_exec,
)

MULTIMODAL_MODELS = [
    "anthropic/claude-sonnet-4-6",
    "anthropic/claude-opus-4-6",
    "gpt-5.2",
    "gpt-5.3-codex",
    "gpt-5.4",
]

for model in MULTIMODAL_MODELS:
    agent = Agent(
        model=model,
        tools=[chart_tool],
        system_prompt=(
            "Use get_error_chart when asked. After seeing the chart, "
            "answer the user's question about it. Be concise."
        ),
    )
    try:
        await agent.prompt(
            "Get the error chart, then tell me: which month has the highest "
            "error count? Reply with just the month name."
        )
        assistants = [m for m in agent.messages if m.get("role") == "assistant"]
        last = assistants[-1] if assistants else {}
        if last.get("error_message"):
            raise RuntimeError(last["error_message"])
        answer = (last.get("content") or "").lower()
        saw_image = "may" in answer
        status = "PASSED" if saw_image else "FAILED"
        print(f"  {status} {model:<42} | {answer[:60]}")
    except Exception as e:
        print(f"  FAILED {model:<42} | ERROR: {e}")