docs/en/learn/create-custom-tools.mdx
This guide provides detailed instructions on creating custom tools for the CrewAI framework and how to efficiently manage and utilize these tools, incorporating the latest functionalities such as tool delegation, error handling, and dynamic tool calling. It also highlights the importance of collaboration tools, enabling agents to perform a wide range of actions.
<Tip> **Want to publish your tool for the community?** If you're building a tool that others could benefit from, check out the [Publish Custom Tools](/en/guides/tools/publish-custom-tools) guide to learn how to package and distribute your tool on PyPI. </Tip>BaseToolTo create a personalized tool, inherit from BaseTool and define the necessary attributes, including the args_schema for input validation, and the _run method.
from typing import Type
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class MyToolInput(BaseModel):
"""Input schema for MyCustomTool."""
argument: str = Field(..., description="Description of the argument.")
class MyCustomTool(BaseTool):
name: str = "Name of my tool"
description: str = "What this tool does. It's vital for effective utilization."
args_schema: Type[BaseModel] = MyToolInput
def _run(self, argument: str) -> str:
# Your tool's logic here
return "Tool's result"
tool DecoratorAlternatively, you can use the tool decorator @tool. This approach allows you to define the tool's attributes and functionality directly within a function,
offering a concise and efficient way to create specialized tools tailored to your needs.
from crewai.tools import tool
@tool("Tool Name")
def my_simple_tool(question: str) -> str:
"""Tool description for clarity."""
# Tool logic here
return "Tool output"
To optimize tool performance with caching, define custom caching strategies using the cache_function attribute.
@tool("Tool with Caching")
def cached_tool(argument: str) -> str:
"""Tool functionality description."""
return "Cacheable result"
def my_cache_strategy(arguments: dict, result: str) -> bool:
# Define custom caching logic
return True if some_condition else False
cached_tool.cache_function = my_cache_strategy
CrewAI supports async tools for non-blocking I/O operations. This is useful when your tool needs to make HTTP requests, database queries, or other I/O-bound operations.
@tool Decorator with Async FunctionsThe simplest way to create an async tool is using the @tool decorator with an async function:
import aiohttp
from crewai.tools import tool
@tool("Async Web Fetcher")
async def fetch_webpage(url: str) -> str:
"""Fetch content from a webpage asynchronously."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
BaseTool with Async SupportFor more control, subclass BaseTool and implement both _run (sync) and _arun (async) methods:
import requests
import aiohttp
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class WebFetcherInput(BaseModel):
"""Input schema for WebFetcher."""
url: str = Field(..., description="The URL to fetch")
class WebFetcherTool(BaseTool):
name: str = "Web Fetcher"
description: str = "Fetches content from a URL"
args_schema: type[BaseModel] = WebFetcherInput
def _run(self, url: str) -> str:
"""Synchronous implementation."""
return requests.get(url).text
async def _arun(self, url: str) -> str:
"""Asynchronous implementation for non-blocking I/O."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
By adhering to these guidelines and incorporating new functionalities and collaboration tools into your tool creation and management processes, you can leverage the full capabilities of the CrewAI framework, enhancing both the development experience and the efficiency of your AI agents.