CrewAI Integration

Anvil integrates with CrewAI through the to_crewai() adapter, allowing you to use generated tools with CrewAI agents.

Installation

pip install "anvil-agent[crewai]"

Basic Usage

from anvil import Anvil
from crewai import Agent, Task, Crew

# Create Anvil tools
anvil = Anvil()

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

analyze = anvil.use_tool(
    name="analyze_data",
    intent="Analyze data and extract insights"
)

# Convert to CrewAI tools
crew_tools = [
    search.to_crewai(),
    analyze.to_crewai(),
]

# Create CrewAI agent
researcher = Agent(
    role="Research Analyst",
    goal="Find and analyze information",
    backstory="Expert at finding and synthesizing information",
    tools=crew_tools,
    verbose=True,
)

# Create task
research_task = Task(
    description="Research the latest developments in AI agents",
    expected_output="A summary of key developments",
    agent=researcher,
)

# Create and run crew
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
)

result = crew.kickoff()
print(result)

How the Adapter Works

The to_crewai() method:

Creates a CrewAI BaseTool subclass
Sets name and description from your tool
Generates an args_schema using Pydantic
Wraps the _run() method for execution

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:
class SearchDocsInput(BaseModel):
    query: str = Field(description="query parameter")

class SearchDocsTool(BaseTool):
    name: str = "search_docs"
    description: str = "Search documentation"
    args_schema: Type[BaseModel] = SearchDocsInput

    def _run(self, query: str) -> Any:
        return anvil_tool.run(query=query)

With Multiple Agents

Use different tools for different agents:

# Create specialized tools
search_tool = anvil.use_tool(
    name="web_search",
    intent="Search the web"
).to_crewai()

write_tool = anvil.use_tool(
    name="write_content",
    intent="Write polished content"
).to_crewai()

code_tool = anvil.use_tool(
    name="analyze_code",
    intent="Analyze and explain code"
).to_crewai()

# Assign to specialized agents
researcher = Agent(
    role="Researcher",
    goal="Find information",
    tools=[search_tool],
)

writer = Agent(
    role="Writer",
    goal="Create content",
    tools=[write_tool],
)

developer = Agent(
    role="Developer",
    goal="Analyze code",
    tools=[code_tool],
)

# Build crew
crew = Crew(
    agents=[researcher, writer, developer],
    tasks=[...],
    process=Process.sequential,
)

Input Parameters

Define explicit inputs for better agent understanding:

from anvil import InputParam

email_tool = anvil.use_tool(
    name="send_email",
    intent="Send an email to a recipient",
    inputs=[
        InputParam(
            name="to",
            param_type="str",
            required=True,
            description="Recipient email address"
        ),
        InputParam(
            name="subject",
            param_type="str",
            required=True,
            description="Email subject line"
        ),
        InputParam(
            name="body",
            param_type="str",
            required=True,
            description="Email body content"
        ),
        InputParam(
            name="cc",
            param_type="list",
            required=False,
            default=[],
            description="CC recipients"
        ),
    ]
)

crew_tool = email_tool.to_crewai()

Tool Descriptions

CrewAI agents use tool descriptions to decide when to use them. Write clear intents:

# Good - specific and actionable
anvil.use_tool(
    name="github_search",
    intent="Search GitHub repositories by name, topic, or language. Returns repo names, stars, and descriptions."
)

# Less effective - vague
anvil.use_tool(
    name="search",
    intent="Search for things"
)

Error Handling

CrewAI handles tool errors gracefully. With Anvil’s self-healing, many errors are automatically fixed:

anvil = Anvil(
    self_healing=True,
    max_heal_attempts=2,
)

# If a tool fails, Anvil will attempt to fix it
# before the error reaches CrewAI
tool = anvil.use_tool(name="api_call", intent="Call external API")

Sequential vs Hierarchical

Anvil tools work with both CrewAI process types:

# Sequential process
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential,
)

# Hierarchical process
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4"),
)

Combining with Built-in Tools

Mix Anvil tools with CrewAI’s built-in tools:

from crewai_tools import SerperDevTool, WebsiteSearchTool

# CrewAI built-in tools
serper = SerperDevTool()
website = WebsiteSearchTool()

# Anvil tools
custom = anvil.use_tool(
    name="custom_analysis",
    intent="Perform custom data analysis"
).to_crewai()

# Use together
agent = Agent(
    role="Analyst",
    tools=[serper, website, custom],
)

Async Support

CrewAI supports async execution:

import asyncio

async def main():
    crew = Crew(agents=[...], tasks=[...])
    result = await crew.kickoff_async()
    print(result)

asyncio.run(main())

Anvil tools work seamlessly with async crews.