# Programmatic Tool Calling (Experimental)

Emulate Anthropic's Programmatic Tool Calling pattern in LangChain.

Ref:

- https://platform.claude.com/docs/en/agents-and-tools/tool-use/programmatic-tool-calling

**Problem**: Traditional tool calling requires API round-trips for each tool call.
With many tool calls, this adds latency and token overhead.

**Solution**: Let the agent generate Python code that calls multiple tools,
execute the code in a sandbox, and return only the final result.

**Benefits**:

- Reduced latency (batch multiple tool calls)
- Token efficiency (aggregate results before returning to model)
- Complex workflows (loops, conditionals over tool results)

Uses [sandbox-runtime](https://github.com/anthropic-experimental/sandbox-runtime) for secure execution.
No global install required - runs via `npx`.

## Setup



```python
import json
import re
import shutil
import tempfile
from pathlib import Path
from typing import Any

import rich
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import BaseTool, tool
from langgraph.graph.state import CompiledStateGraph

load_dotenv()

# Check for srt - prefer global install, fallback to npx
SRT_CMD: list[str]
if shutil.which("srt"):
    SRT_CMD = ["srt"]
    rich.print("[green]srt found (global)[/green]")
elif shutil.which("npx"):
    SRT_CMD = ["npx", "@anthropic-ai/sandbox-runtime"]
    rich.print("[green]srt available via npx[/green]")
else:
    raise RuntimeError("srt not available. Install Node.js for npx support.")
```


    srt available via npx



## Define Tool Registry

These tools can be called programmatically from generated code via `tool_call()`.



```python
@tool
def query_sales(region: str) -> dict[str, Any]:
    """Query sales data for a region.

    Returns: {"region": str, "revenue": int, "units": int, "top_product": str}
    Error: {"error": str} if region not found.
    """
    data = {
        "west": {"revenue": 150000, "units": 1200, "top_product": "Widget A"},
        "east": {"revenue": 220000, "units": 1800, "top_product": "Widget B"},
        "central": {"revenue": 180000, "units": 1500, "top_product": "Widget A"},
        "north": {"revenue": 95000, "units": 800, "top_product": "Widget C"},
        "south": {"revenue": 130000, "units": 1100, "top_product": "Widget B"},
    }
    region_lower = region.lower()
    if region_lower in data:
        return {"region": region, **data[region_lower]}
    return {"error": f"Unknown region: {region}"}


@tool
def get_weather(city: str) -> dict[str, Any]:
    """Get current weather for a city.

    Returns: {"city": str, "temp": int, "condition": str, "humidity": int}
    Error: {"error": str} if city not found.
    """
    data = {
        "tokyo": {"temp": 22, "condition": "Sunny", "humidity": 45},
        "new york": {"temp": 18, "condition": "Cloudy", "humidity": 60},
        "london": {"temp": 15, "condition": "Rainy", "humidity": 80},
        "paris": {"temp": 20, "condition": "Partly Cloudy", "humidity": 55},
        "sydney": {"temp": 25, "condition": "Sunny", "humidity": 40},
    }
    city_lower = city.lower()
    if city_lower in data:
        return {"city": city, **data[city_lower]}
    return {"error": f"Unknown city: {city}"}


TOOL_REGISTRY: dict[str, BaseTool] = {t.name: t for t in [query_sales, get_weather]}

rich.print("Available tools:", list(TOOL_REGISTRY.keys()))
```


    Available tools:
    ['query_sales', 'get_weather']



## Tool Search

Allows agent to discover available tools by pattern.



```python
@tool
def tool_search(pattern: str) -> list[dict[str, Any]]:
    """Search for available tools by regex pattern.

    Args:
        pattern: Regex pattern to match against tool names and descriptions.

    Returns: list of tool schemas with properties, required, description, etc.
    Empty list if no tools found.
    """
    rich.print(f"[bold cyan]tool_search({pattern!r})[/bold cyan]")
    matches = []
    for name, t in TOOL_REGISTRY.items():
        if re.search(pattern, name, re.IGNORECASE) or re.search(pattern, t.description, re.IGNORECASE):
            if t.args_schema and hasattr(t.args_schema, "model_json_schema"):
                schema = t.args_schema.model_json_schema()
                schema["name"] = name  # Add tool name
                matches.append(schema)
    rich.print(f"[dim]{json.dumps(matches, indent=2)}[/dim]")
    return matches
```

## Sandboxed Code Execution Tool

Execute Python code in a sandbox with `tool_call()` to invoke registered tools.
Uses JSON-RPC 2.0 over stdin/stdout to communicate with the host:

- **Request** (has id): `{"jsonrpc": "2.0", "method": "tool_name", "params": {...}, "id": 1}`
- **Response**: `{"jsonrpc": "2.0", "result": {...}, "id": 1}`
- **Notification** (no id): `{"jsonrpc": "2.0", "method": "print", "params": {"text": "..."}}`



```python
import asyncio


def create_sandboxed_execute_code_tool(registry: dict[str, BaseTool]) -> BaseTool:
    """Create a sandboxed code execution tool using srt with JSON-RPC 2.0."""

    async def execute_code_impl(code: str) -> str:
        """Execute async Python code in a sandbox.

        Available functions:
        - await tool_call(name, **kwargs): Call a registered tool, returns dict
        - print(): Output results (only printed text is returned)

        Communication uses JSON-RPC 2.0 protocol over stdout/stdin.

        Example:
            sales = await tool_call("query_sales", region="west")
            print(f"Revenue: {sales['revenue']}")
        """
        rich.print("[bold cyan]Executing sandboxed code:[/bold cyan]")
        rich.print(code)
        rich.print("[bold cyan]---[/bold cyan]")

        # Indent user code for async main()
        indented_code = "\n".join("    " + line for line in code.split("\n"))

        # JSON-RPC 2.0 wrapper
        wrapper = f'''
import asyncio
import json
import sys

_request_id = 0

async def tool_call(name, **kwargs):
    """Call a tool on the host via JSON-RPC 2.0."""
    global _request_id
    _request_id += 1
    request = {{
        "jsonrpc": "2.0",
        "method": name,
        "params": kwargs,
        "id": _request_id
    }}
    sys.stdout.write(json.dumps(request) + "\\n")
    sys.stdout.flush()
    response = json.loads(sys.stdin.readline().strip())
    if "error" in response:
        raise Exception(f"Tool error: {{response['error']['message']}}")
    return response["result"]

_original_print = print
def print(*args, **kwargs):
    """Print via JSON-RPC 2.0 notification."""
    import io
    buf = io.StringIO()
    kwargs["file"] = buf
    _original_print(*args, **kwargs)
    text = buf.getvalue()
    notification = {{
        "jsonrpc": "2.0",
        "method": "print",
        "params": {{"text": text}}
    }}
    sys.stdout.write(json.dumps(notification) + "\\n")
    sys.stdout.flush()

async def main():
{indented_code}

asyncio.run(main())
'''

        with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
            f.write(wrapper)
            script_path = f.name

        try:
            proc = await asyncio.create_subprocess_exec(
                *SRT_CMD,
                "python",
                "-u",
                script_path,
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            assert proc.stdin is not None
            assert proc.stdout is not None
            assert proc.stderr is not None
            proc_stdin = proc.stdin
            proc_stdout = proc.stdout
            proc_stderr = proc.stderr

            output_lines: list[str] = []

            while True:
                try:
                    line_bytes = await asyncio.wait_for(proc_stdout.readline(), timeout=30)
                except TimeoutError:
                    break

                if not line_bytes:
                    break

                line = line_bytes.decode().strip()
                if not line:
                    continue

                # Parse JSON-RPC message
                try:
                    msg = json.loads(line)
                except json.JSONDecodeError:
                    continue

                # Check if notification (no id) or request (has id)
                if "id" not in msg:
                    # Notification: handle print
                    if msg.get("method") == "print":
                        output_lines.append(msg["params"]["text"])
                else:
                    # Request: tool call
                    request_id = msg["id"]
                    tool_name = msg["method"]
                    tool_kwargs = msg.get("params", {})

                    if tool_name in registry:
                        tool_obj = registry[tool_name]
                        # Use ainvoke for async tools (like MCP tools)
                        if hasattr(tool_obj, "coroutine") and tool_obj.coroutine is not None:
                            raw_result = await tool_obj.ainvoke(tool_kwargs)
                        else:
                            raw_result = tool_obj.invoke(tool_kwargs)

                        # MCP tools may return JSON strings - parse if needed
                        if isinstance(raw_result, str):
                            try:
                                call_result = json.loads(raw_result)
                            except json.JSONDecodeError:
                                call_result = raw_result
                        else:
                            call_result = raw_result

                        rich.print(f"[dim]  tool_call({tool_name!r}, {tool_kwargs}) -> {call_result}[/dim]")

                        response = {
                            "jsonrpc": "2.0",
                            "result": call_result,
                            "id": request_id,
                        }
                    else:
                        response = {
                            "jsonrpc": "2.0",
                            "error": {"code": -32601, "message": f"Tool not found: {tool_name}"},
                            "id": request_id,
                        }

                    proc_stdin.write((json.dumps(response) + "\n").encode())
                    await proc_stdin.drain()

            stderr_bytes = await proc_stderr.read()
            stderr = stderr_bytes.decode()
            await proc.wait()

            output = "".join(output_lines)
            if stderr:
                output += f"\nStderr: {stderr}"
            if proc.returncode != 0:
                output += f"\nExit code: {proc.returncode}"

            return output.strip() if output.strip() else "Code executed successfully (no output)"

        except TimeoutError:
            proc.kill()
            return "Error: Execution timed out (30s limit)"
        except Exception as e:
            return f"Error: {type(e).__name__}: {e}"
        finally:
            Path(script_path).unlink(missing_ok=True)

    # Create StructuredTool with async coroutine
    from langchain_core.tools import StructuredTool

    return StructuredTool.from_function(
        coroutine=execute_code_impl,
        name="execute_code",
        description="""Execute async Python code in a sandbox using JSON-RPC 2.0.

Available functions:
- await tool_call(name, **kwargs): Call a registered tool, returns dict
- print(): Output results (only printed text is returned)

Example:
    sales = await tool_call("query_sales", region="west")
    print(f"Revenue: {sales['revenue']}")""",
    )


execute_code_tool = create_sandboxed_execute_code_tool(TOOL_REGISTRY)
rich.print("Sandboxed execute_code tool created (JSON-RPC 2.0)")
```


    Sandboxed execute_code tool created (JSON-RPC 2.0)



## Create Agent



```python
model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

SYSTEM_PROMPT = """You are a data analyst assistant.

You have access to:
- tool_search: Search for available tools by regex pattern
- execute_code: Run Python code in a secure sandbox with tool_call(name, **kwargs)

Workflow:
1. Use tool_search to find relevant tools
2. Use execute_code with tool_call() to invoke them in batch
3. Aggregate results and print the final summary"""

agent: CompiledStateGraph[Any] = create_agent(
    model=model,
    tools=[tool_search, execute_code_tool],
    system_prompt=SYSTEM_PROMPT,
)

rich.print("Agent created")
```


    Agent created



## Test: Multi-Region Analysis



```python
query = """For each region, get sales data and check weather in its capital city.
Capital cities: West=Sydney, East=New York, Central=London, North=Tokyo, South=Paris.

Calculate revenue per unit for each region, then recommend the best region to visit
considering both business performance (high revenue per unit) and weather (sunny preferred)."""

result = await agent.ainvoke({"messages": [{"role": "user", "content": query}]})

rich.print("\n[bold green]Final response:[/bold green]")
rich.print(result["messages"][-1].content)
```


    tool_search('sales|weather')




    [
      {
        "description": "Query sales data for a region.\n\nReturns: {\"region\": str, \"revenue\": int, \"units\": int, 
    \"top_product\": str}\nError: {\"error\": str} if region not found.",
        "properties": {
          "region": {
            "title": "Region",
            "type": "string"
          }
        },
        "required": [
          "region"
        ],
        "title": "query_sales",
        "type": "object",
        "name": "query_sales"
      },
      {
        "description": "Get current weather for a city.\n\nReturns: {\"city\": str, \"temp\": int, \"condition\": str, 
    \"humidity\": int}\nError: {\"error\": str} if city not found.",
        "properties": {
          "city": {
            "title": "City",
            "type": "string"
          }
        },
        "required": [
          "city"
        ],
        "title": "get_weather",
        "type": "object",
        "name": "get_weather"
      }
    ]




    Executing sandboxed code:




    
    # Define regions and their capitals
    regions_capitals = {
        "West": "Sydney",
        "East": "New York",
        "Central": "London",
        "North": "Tokyo",
        "South": "Paris"
    }
    
    # Collect sales data and weather for each region
    results = []
    
    for region, capital in regions_capitals.items():
        # Get sales data
        sales_data = await tool_call("query_sales", region=region)
        
        # Get weather data
        weather_data = await tool_call("get_weather", city=capital)
        
        # Calculate revenue per unit
        if 'revenue' in sales_data and 'units' in sales_data and sales_data['units'] > 0:
            revenue_per_unit = sales_data['revenue'] / sales_data['units']
        else:
            revenue_per_unit = 0
        
        results.append({
            "region": region,
            "capital": capital,
            "revenue": sales_data.get('revenue', 0),
            "units": sales_data.get('units', 0),
            "revenue_per_unit": revenue_per_unit,
            "top_product": sales_data.get('top_product', 'N/A'),
            "temperature": weather_data.get('temp', 'N/A'),
            "weather_condition": weather_data.get('condition', 'N/A'),
            "humidity": weather_data.get('humidity', 'N/A')
        })
    
    # Display results
    print("=" * 80)
    print("REGIONAL ANALYSIS: SALES DATA & WEATHER CONDITIONS")
    print("=" * 80)
    print()
    
    for r in results:
        print(f"Region: {r['region']} (Capital: {r['capital']})")
        print(f"  Sales Performance:")
        print(f"    - Revenue: ${r['revenue']:,}")
        print(f"    - Units Sold: {r['units']:,}")
        print(f"    - Revenue per Unit: ${r['revenue_per_unit']:.2f}")
        print(f"    - Top Product: {r['top_product']}")
        print(f"  Weather Conditions:")
        print(f"    - Temperature: {r['temperature']}°F")
        print(f"    - Condition: {r['weather_condition']}")
        print(f"    - Humidity: {r['humidity']}%")
        print()
    
    # Analyze and recommend
    print("=" * 80)
    print("RECOMMENDATION ANALYSIS")
    print("=" * 80)
    print()
    
    # Score each region
    scored_regions = []
    for r in results:
        # Business score (normalized revenue per unit)
        business_score = r['revenue_per_unit']
        
        # Weather score (sunny is best)
        weather_condition = r['weather_condition'].lower()
        if 'sunny' in weather_condition or 'clear' in weather_condition:
            weather_score = 10
        elif 'partly' in weather_condition or 'fair' in weather_condition:
            weather_score = 7
        elif 'cloudy' in weather_condition or 'overcast' in weather_condition:
            weather_score = 5
        elif 'rain' in weather_condition or 'storm' in weather_condition:
            weather_score = 2
        else:
            weather_score = 5  # neutral
        
        scored_regions.append({
            "region": r['region'],
            "capital": r['capital'],
            "business_score": business_score,
            "weather_score": weather_score,
            "weather_condition": r['weather_condition'],
            "revenue_per_unit": r['revenue_per_unit']
        })
    
    # Sort by business score first (most important)
    scored_regions.sort(key=lambda x: (x['business_score'], x['weather_score']), reverse=True)
    
    print("Rankings by Revenue per Unit:")
    for i, r in enumerate(scored_regions, 1):
        print(f"{i}. {r['region']}: ${r['revenue_per_unit']:.2f} per unit | Weather: {r['weather_condition']}")
    
    print()
    print("-" * 80)
    print()
    
    # Find best region with good weather
    best_overall = None
    for r in scored_regions:
        if r['weather_score'] >= 7:  # Good weather (sunny or partly sunny)
            best_overall = r
            break
    
    if best_overall:
        print(f"🌟 RECOMMENDED REGION: {best_overall['region']} ({best_overall['capital']})")
        print()
        print(f"Reasons:")
        print(f"  ✓ Strong business performance: ${best_overall['revenue_per_unit']:.2f} revenue per unit")
        print(f"  ✓ Favorable weather conditions: {best_overall['weather_condition']}")
        print()
        print(f"This region offers the best combination of high revenue per unit")
        print(f"and pleasant weather for your visit.")
    else:
        # If no region has good weather, just pick the best business performer
        best_business = scored_regions[0]
        print(f"🌟 RECOMMENDED REGION: {best_business['region']} ({best_business['capital']})")
        print()
        print(f"Reasons:")
        print(f"  ✓ Best business performance: ${best_business['revenue_per_unit']:.2f} revenue per unit")
        print(f"  ⚠ Weather: {best_business['weather_condition']} (not ideal, but business is strong)")
        print()
        print(f"Note: While weather conditions are not optimal, this region has")
        print(f"the strongest business performance and deserves priority attention.")
    




    ---




      tool_call('query_sales', {'region': 'West'}) -> {'region': 'West', 'revenue': 150000, 'units': 1200, 
    'top_product': 'Widget A'}




      tool_call('get_weather', {'city': 'Sydney'}) -> {'city': 'Sydney', 'temp': 25, 'condition': 'Sunny', 'humidity': 
    40}




      tool_call('query_sales', {'region': 'East'}) -> {'region': 'East', 'revenue': 220000, 'units': 1800, 
    'top_product': 'Widget B'}




      tool_call('get_weather', {'city': 'New York'}) -> {'city': 'New York', 'temp': 18, 'condition': 'Cloudy', 
    'humidity': 60}




      tool_call('query_sales', {'region': 'Central'}) -> {'region': 'Central', 'revenue': 180000, 'units': 1500, 
    'top_product': 'Widget A'}




      tool_call('get_weather', {'city': 'London'}) -> {'city': 'London', 'temp': 15, 'condition': 'Rainy', 'humidity': 
    80}




      tool_call('query_sales', {'region': 'North'}) -> {'region': 'North', 'revenue': 95000, 'units': 800, 
    'top_product': 'Widget C'}




      tool_call('get_weather', {'city': 'Tokyo'}) -> {'city': 'Tokyo', 'temp': 22, 'condition': 'Sunny', 'humidity': 
    45}




      tool_call('query_sales', {'region': 'South'}) -> {'region': 'South', 'revenue': 130000, 'units': 1100, 
    'top_product': 'Widget B'}




      tool_call('get_weather', {'city': 'Paris'}) -> {'city': 'Paris', 'temp': 20, 'condition': 'Partly Cloudy', 
    'humidity': 55}




    
    Final response:




    ## Summary
    
    Based on my analysis of sales data and weather conditions across all five regions, here are the key findings:
    
    ### **Recommendation: Visit the West Region (Sydney) 🌟**
    
    **Why West/Sydney is the best choice:**
    
    1. **Top Business Performance**: Highest revenue per unit at **$125.00** (generating $150,000 from 1,200 units 
    sold)
    2. **Excellent Weather**: Sunny conditions with comfortable temperature and low humidity (40%)
    3. **Best Overall Value**: The only region that combines #1 business performance with sunny weather
    
    ### **Other Notable Findings:**
    
    - **East (New York)**: Second-best revenue per unit ($122.22) but has cloudy weather
    - **Central (London)**: Good sales ($120/unit) but rainy conditions make it less appealing
    - **North (Tokyo)**: Has sunny weather but lower business performance ($118.75/unit)
    - **South (Paris)**: Lowest revenue per unit ($118.18) with only partly cloudy conditions
    
    The West region stands out as the clear winner, offering both the strongest business metrics and the most pleasant 
    weather conditions for your visit. You'll be able to focus on the highest-performing market while enjoying 
    comfortable outdoor conditions.