Tool Chaining

Anvil’s ToolChain lets you combine multiple tools into sequential workflows where the output of one tool becomes the input of the next.

Basic Chaining

Use the pipe() method to chain tools:

from anvil import Anvil

anvil = Anvil()

# Create individual tools
search = anvil.use_tool(
    name="web_search",
    intent="Search the web and return results"
)

summarize = anvil.use_tool(
    name="summarize",
    intent="Summarize text into key points"
)

translate = anvil.use_tool(
    name="translate",
    intent="Translate text to Spanish"
)

# Chain them together
chain = search.pipe(summarize).pipe(translate)

# Execute the chain
result = chain.run(query="Latest AI developments")

How Chaining Works

┌─────────┐     ┌─────────────┐     ┌───────────┐
│ search  │────▶│  summarize  │────▶│ translate │
│         │     │             │     │           │
│ query   │     │ text=result │     │ text=result│
└─────────┘     └─────────────┘     └───────────┘

1. First tool executes with your initial arguments
2. Output is passed to the next tool
3. If output is a dict, it’s unpacked as **kwargs
4. If output is not a dict, it’s passed as data=output
5. Process continues until the last tool

Creating Chains

Method 1: Fluent API

chain = tool1.pipe(tool2).pipe(tool3)

Method 2: ToolChain Constructor

from anvil import ToolChain

chain = ToolChain([tool1, tool2, tool3])

Method 3: Chain Extension

# Start with one chain
chain1 = tool1.pipe(tool2)

# Extend it
chain2 = chain1.pipe(tool3)

Data Flow

Dict Output (Unpacked)

If a tool returns a dict, it’s unpacked as keyword arguments:

# Tool 1 returns: {"text": "Hello", "language": "en"}
# Tool 2 receives: run(text="Hello", language="en")

Non-Dict Output (Wrapped)

If a tool returns a non-dict value, it’s passed as data:

# Tool 1 returns: "Hello World"
# Tool 2 receives: run(data="Hello World")

Execution Methods

run(**kwargs)

Execute the chain, raising on error:

result = chain.run(query="test")

run_safe(**kwargs)

Execute without raising, returns ToolResult:

result = chain.run_safe(query="test")

if result.success:
    print(result.data)
else:
    print(f"Error: {result.error}")

run_async(**kwargs)

Async execution:

result = await chain.run_async(query="test")

run_safe_async(**kwargs)

Async execution without raising:

result = await chain.run_safe_async(query="test")

Real-World Examples

Research and Report

# Research pipeline
research = anvil.use_tool(name="search", intent="Search for information")
analyze = anvil.use_tool(name="analyze", intent="Analyze and extract insights")
format_report = anvil.use_tool(name="format", intent="Format as markdown report")

pipeline = research.pipe(analyze).pipe(format_report)
report = pipeline.run(topic="AI in healthcare")

Data Processing

# ETL pipeline
fetch = anvil.use_tool(name="fetch", intent="Fetch data from API")
transform = anvil.use_tool(name="transform", intent="Clean and transform data")
load = anvil.use_tool(name="load", intent="Load data into database")

etl = fetch.pipe(transform).pipe(load)
etl.run(endpoint="https://api.example.com/data")

Content Creation

# Content pipeline
outline = anvil.use_tool(name="outline", intent="Create content outline")
write = anvil.use_tool(name="write", intent="Write full content")
edit = anvil.use_tool(name="edit", intent="Edit for grammar and style")
seo = anvil.use_tool(name="seo", intent="Optimize for SEO")

content_pipeline = outline.pipe(write).pipe(edit).pipe(seo)
article = content_pipeline.run(topic="Getting started with Python")

Chain Properties

# Get number of tools in chain
print(len(chain))  # 3

# String representation
print(chain)  # ToolChain(search -> summarize -> translate)

Error Handling

Errors in any tool stop the chain:

result = chain.run_safe(query="test")

if not result.success:
    # Error includes which tool failed
    print(f"Chain failed: {result.error}")

With self-healing enabled, Anvil attempts to fix failing tools:

anvil = Anvil(self_healing=True)

# If summarize fails, Anvil tries to fix it
# before the error propagates
chain = search.pipe(summarize)

Logging Chain Execution

Chain events are logged:

anvil = Anvil(log_file="anvil.log")

# These events are logged:
# - chain_started
# - tool_executed (for each tool)
# - chain_completed or chain_failed

chain.run(query="test")

# Query chain events
events = anvil.logger.get_history(event_type="chain_completed")

Best Practices

1. Design Clear Interfaces

Each tool should have clear input/output contracts:

# Good: Clear what goes in and out
fetch_tool = anvil.use_tool(
    name="fetch_users",
    intent="Fetch users from API, returns list of user dicts"
)

# Good: Matches expected input from previous tool
process_tool = anvil.use_tool(
    name="process_users",
    intent="Process list of user dicts, returns summary dict"
)

2. Handle Missing Data

Tools should handle missing optional inputs:

filter_tool = anvil.use_tool(
    name="filter_data",
    intent="Filter data by criteria. If no criteria provided, return all data."
)

3. Keep Chains Focused

Don’t create overly long chains:

# Good: Focused chain
data_pipeline = fetch.pipe(clean).pipe(analyze)

# Less ideal: Too many steps
mega_chain = t1.pipe(t2).pipe(t3).pipe(t4).pipe(t5).pipe(t6).pipe(t7)

4. Use Safe Execution for Production

result = chain.run_safe(query="test")

if not result.success:
    # Log error
    logger.error(f"Chain failed: {result.error}")
    # Fallback behavior
    return default_response

Next Steps

• Credential Management
• Sandbox & Security
• Logging & Observability