Skip to main content

Overview

This example demonstrates using Monty to analyze team expense data across multiple users. Instead of making 50+ sequential tool calls, the LLM writes a loop that processes each team member’s expenses in code.
This example is adapted from Anthropic’s Programmatic Tool Calling cookbook.

Why This Example Matters

With traditional tool calling, analyzing expenses for a team would require:
  1. Call get_team_members() → returns 5 members
  2. For each member:
    • Call get_expenses(user_id, quarter, category) → returns 10-15 expense items
    • Call get_custom_budget(user_id) → returns budget or null
  3. Process results in the LLM
This floods the context window with hundreds of expense items. With Monty:
  • The loop runs in the sandbox
  • Only the final summary returns to the host
  • Token usage drops dramatically

The Task

Analyze Q3 travel expenses for the Engineering team and identify who exceeded their budget (standard $5,000 or custom).

Code Structure

Type Definitions

type_definitions = '''
from typing import Any

async def get_team_members(department: str) -> dict[str, Any]:
    """Get list of team members for a department.
    Args:
        department: The department name (e.g., "Engineering").
    Returns:
        Dictionary with list of team members.
    """
    ...

async def get_expenses(user_id: int, quarter: str, category: str) -> dict[str, Any]:
    """Get expense line items for a user.
    Args:
        user_id: The user's ID.
        quarter: The quarter (e.g., "Q3").
        category: The expense category (e.g., "travel").
    Returns:
        Dictionary with expense items.
    """
    ...

async def get_custom_budget(user_id: int) -> dict[str, Any] | None:
    """Get custom budget for a user if they have one.
    Args:
        user_id: The user's ID.
    Returns:
        Custom budget info or None if no custom budget.
    """
    ...
'''

The Sandbox Code

code = """
# Get Engineering team members
team_data = await get_team_members(department="Engineering")
team_members = team_data.get("members", [])

# Standard budget
STANDARD_BUDGET = 5000

# Process each team member
total_members = len(team_members)
over_budget_list = []

for member in team_members:
    user_id = member.get("id")
    name = member.get("name")

    # Get Q3 travel expenses for this user
    expenses_data = await get_expenses(user_id=user_id, quarter="Q3", category="travel")
    expense_items = expenses_data.get("expenses", [])

    # Sum up total expenses
    total_spent = sum(item.get("amount", 0) for item in expense_items)

    # Check if they exceeded standard budget
    if total_spent > STANDARD_BUDGET:
        # Check for custom budget
        custom_budget_data = await get_custom_budget(user_id=user_id)

        if custom_budget_data is not None:
            budget = custom_budget_data.get("budget", STANDARD_BUDGET)
        else:
            budget = STANDARD_BUDGET

        # Check if they exceeded their actual budget (standard or custom)
        if total_spent > budget:
            amount_over = total_spent - budget
            over_budget_list.append({
                "name": name,
                "total_spent": total_spent,
                "budget": budget,
                "amount_over": amount_over
            })

# Return the analysis
{
    "total_team_members_analyzed": total_members,
    "count_exceeded_budget": len(over_budget_list),
    "over_budget_details": over_budget_list
}
"""

Execution

import pydantic_monty

m = pydantic_monty.Monty(
    code,
    inputs=['prompt'],
    script_name='expense.py',
    type_check=True,
    type_check_stubs=type_definitions,
)

async def main():
    output = await pydantic_monty.run_monty_async(
        m,
        inputs={'prompt': 'testing'},
        external_functions={
            'get_team_members': data.get_team_members,
            'get_expenses': data.get_expenses,
            'get_custom_budget': data.get_custom_budget,
        },
    )
    print(output)

Example Data

Team Members

team_members = [
    {'id': 1, 'name': 'Alice Chen'},
    {'id': 2, 'name': 'Bob Smith'},
    {'id': 3, 'name': 'Carol Jones'},
    {'id': 4, 'name': 'David Kim'},
    {'id': 5, 'name': 'Eve Wilson'},
]

Expense Data

Each user has 8-15 expense line items with details like:
{
    'date': '2024-07-15',
    'amount': 450.00,
    'description': 'Flight to NYC'
}

Custom Budgets

custom_budgets = {
    2: {'amount': 7000.00, 'reason': 'International travel required'},
}

Expected Output

{
    "total_team_members_analyzed": 5,
    "count_exceeded_budget": 2,
    "over_budget_details": [
        {
            "name": "Carol Jones",
            "total_spent": 6740.0,
            "budget": 5000,
            "amount_over": 1740.0
        },
        {
            "name": "Eve Wilson",
            "total_spent": 5680.0,
            "budget": 5000,
            "amount_over": 680.0
        }
    ]
}
Bob Smith spent over the standard budget but has a custom budget of $7,000, so he’s not flagged as over budget.

Key Benefits

1

Loops in Code

The for member in team_members loop is natural in Python, but would require complex orchestration with tool calls.
2

Conditional Logic

The code checks if expenses exceed the standard budget, then conditionally fetches custom budgets. This would require multiple LLM round-trips with tool calling.
3

In-Sandbox Computation

Summing expense amounts happens in the sandbox. The LLM doesn’t need to do mental math or see every expense item.
4

Reduced Token Usage

Only the final summary leaves the sandbox. With tool calling, every expense item (50+ items) would flood the context.

Running the Example

uv run python examples/expense_analysis/main.py

Async Patterns

All external functions are async, and Monty handles them seamlessly:
# Multiple async calls in sequence
team_data = await get_team_members(department="Engineering")
expenses_data = await get_expenses(user_id=user_id, quarter="Q3", category="travel")

# Conditional async calls
if total_spent > STANDARD_BUDGET:
    custom_budget_data = await get_custom_budget(user_id=user_id)
For parallel execution, use asyncio.gather() to fire multiple calls at once instead of awaiting each one sequentially.

Next Steps

  • Explore the full source in examples/expense_analysis/
  • Try Web Scraper for browser automation
  • See SQL Playground for file mounting and SQL queries

Build docs developers (and LLMs) love