Agents & LLMs

import asyncio

from timbal import Agent, Workflow

summarizer = Agent(
    name="summarizer",
    model="anthropic/claude-sonnet-4-6",
    max_tokens=1024,
)

workflow = Workflow(name="summarize").step(
    summarizer,
    prompt="Summarize the quarterly report in three bullet points.",
)


async def main():
    result = await workflow().collect()
    print(result.output.collect_text())


asyncio.run(main())

workflow = (
    Workflow(name="support")
    .step(support_agent, prompt="Handle this ticket")  # fixed prompt
)

# Or pass prompt at call time:
result = await workflow(prompt="Customer cannot log in").collect()

import asyncio

from timbal import Agent, Workflow
from timbal.state import get_run_context

classifier = Agent(
    name="classifier",
    model="openai/gpt-5-mini",
    system_prompt="Classify the message as 'technical' or 'billing'. One word only.",
    max_tokens=64,
)

technical_agent = Agent(
    name="technical_agent",
    model="anthropic/claude-sonnet-4-6",
    system_prompt="You are a technical support specialist.",
    max_tokens=1024,
)

billing_agent = Agent(
    name="billing_agent",
    model="openai/gpt-5-mini",
    system_prompt="You are a billing support specialist.",
    max_tokens=1024,
)

workflow = (
    Workflow(name="support_router")
    .step(classifier)
    .step(
        technical_agent,
        when=lambda: "technical"
        in get_run_context().step_span("classifier").output.collect_text().lower(),
    )
    .step(
        billing_agent,
        when=lambda: "billing"
        in get_run_context().step_span("classifier").output.collect_text().lower(),
    )
)


async def main():
    result = await workflow(prompt="I can't access my account after the upgrade").collect()
    print(result.output.collect_text())


asyncio.run(main())

workflow = (
    Workflow(name="doc_pipeline")
    .step(fetch_content, url="https://example.com/report")   # plain function
    .step(extract_metadata, html=lambda: get_run_context().step_span("fetch_content").output)
    .step(summarizer, prompt=lambda: f"Summarize: {get_run_context().step_span('extract_metadata').output['text']}")
    .step(format_markdown, summary=lambda: get_run_context().step_span("summarizer").output.collect_text())
)

analysis = (
    Workflow(name="analysis")
    .step(fetch_sales)
    .step(summarizer, prompt=lambda: f"Summarize: {get_run_context().step_span('fetch_sales').output}")
)

delivery = (
    Workflow(name="delivery")
    .step(analysis)
    .step(send_email, body=lambda: get_run_context().step_span("analysis").output.collect_text())
)