AutoGen Integration

Anvil integrates with Microsoft AutoGen through the to_autogen() adapter.

Installation

pip install "anvil-agent[autogen]"

Basic Usage

from anvil import Anvil
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create Anvil tools
anvil = Anvil()

search = anvil.use_tool(
    name="web_search",
    intent="Search the web for information"
)

calculator = anvil.use_tool(
    name="calculator",
    intent="Perform mathematical calculations"
)

# Convert to AutoGen tools
autogen_tools = [
    search.to_autogen(),
    calculator.to_autogen(),
]

# Create AutoGen agent
model_client = OpenAIChatCompletionClient(model="gpt-4o")

assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    tools=autogen_tools,
    system_message="You are a helpful assistant with access to search and calculator tools.",
)

# Run conversation
async def main():
    response = await assistant.run(task="What is 15% of 847?")
    print(response.messages[-1].content)

import asyncio
asyncio.run(main())

How the Adapter Works

The to_autogen() method:

Creates an AutoGen FunctionTool
Wraps the Anvil tool’s run() method
Generates proper type hints for parameters
Handles async execution with CancellationToken

Generated Structure

# Anvil tool
tool = anvil.use_tool(
    name="search_docs",
    intent="Search documentation",
    inputs=[
        InputParam(name="query", param_type="str", required=True),
    ]
)

# Converts to approximately:
async def search_docs(query: str, cancellation_token: CancellationToken) -> str:
    """Search documentation"""
    result = anvil_tool.run(query=query)
    return str(result)

autogen_tool = FunctionTool(search_docs, description="Search documentation")

Multi-Agent Teams

Use Anvil tools across multiple AutoGen agents:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

# Create specialized tools
research_tool = anvil.use_tool(
    name="research",
    intent="Research a topic in depth"
).to_autogen()

write_tool = anvil.use_tool(
    name="write",
    intent="Write polished content"
).to_autogen()

# Create specialized agents
researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    tools=[research_tool],
    system_message="You research topics thoroughly.",
)

writer = AssistantAgent(
    name="writer",
    model_client=model_client,
    tools=[write_tool],
    system_message="You write clear, engaging content.",
)

# Create team
termination = TextMentionTermination("COMPLETE")
team = RoundRobinGroupChat(
    [researcher, writer],
    termination_condition=termination,
)

# Run team
async def main():
    result = await team.run(task="Write an article about AI agents")
    for msg in result.messages:
        print(f"{msg.source}: {msg.content[:100]}...")

asyncio.run(main())

Input Parameters

Define typed parameters for AutoGen’s type system:

from anvil import InputParam

api_tool = anvil.use_tool(
    name="call_api",
    intent="Make an API request",
    inputs=[
        InputParam(
            name="endpoint",
            param_type="str",
            required=True,
            description="API endpoint URL"
        ),
        InputParam(
            name="method",
            param_type="str",
            required=False,
            default="GET",
            description="HTTP method"
        ),
        InputParam(
            name="timeout",
            param_type="int",
            required=False,
            default=30,
            description="Request timeout in seconds"
        ),
    ]
)

autogen_tool = api_tool.to_autogen()

Using BaseTool Alternative

For more control, use the BaseTool variant:

from anvil.adapters.autogen import to_autogen_base_tool

# Creates a BaseTool subclass instead of FunctionTool
base_tool = to_autogen_base_tool(anvil_tool)

This provides:

Explicit type annotations
Custom run logic
Better IDE support

Async Execution

AutoGen is async-first. Anvil tools support this natively:

async def main():
    # Anvil tools work seamlessly with AutoGen's async model
    agent = AssistantAgent(
        name="agent",
        tools=[search.to_autogen()],
    )

    result = await agent.run(task="Search for AI news")
    print(result)

Cancellation Support

AutoGen supports task cancellation. Anvil adapters handle this:

from autogen_core import CancellationToken
import asyncio

async def main():
    token = CancellationToken()

    # Start task
    task = asyncio.create_task(
        agent.run(task="Long running search", cancellation_token=token)
    )

    # Cancel after 5 seconds
    await asyncio.sleep(5)
    token.cancel()

    try:
        result = await task
    except asyncio.CancelledError:
        print("Task was cancelled")

Error Handling

Errors propagate through AutoGen’s error handling:

async def main():
    try:
        result = await agent.run(task="Do something risky")
    except Exception as e:
        print(f"Agent failed: {e}")

With self-healing enabled, Anvil attempts to fix tool errors before they reach AutoGen.

Combining with Built-in Tools

Mix Anvil tools with AutoGen’s built-in capabilities:

from autogen_ext.tools import PythonCodeExecutionTool

# AutoGen built-in
code_tool = PythonCodeExecutionTool()

# Anvil tools
search_tool = anvil.use_tool(
    name="search",
    intent="Search the web"
).to_autogen()

# Use together
agent = AssistantAgent(
    name="developer",
    tools=[code_tool, search_tool],
)