커스텀 도구 만들기

커스텀 도구 만들기 (Create Custom Tools)

CrewAI에서 커스텀 도구를 만들고 관리하는 방법을 다룹니다. 도구 위임, 오류 처리, 동적 도구 호출 같은 최신 기능을 포함해, 에이전트가 다양한 동작을 수행하게 해 주는 협업 도구의 중요성도 함께 짚어 줍니다. 기본 패턴은 BaseTool 서브클래싱과 @tool 데코레이터 두 가지입니다.

출처: 공식문서

도구를 커뮤니티에 배포하고 싶다면? 다른 사람에게도 유용할 도구를 만들고 있다면 Publish Custom Tools 가이드에서 PyPI에 패키징·배포하는 법을 확인하세요.

BaseTool 서브클래싱

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"

커스터마이즈된 도구를 만들려면 BaseTool에서 상속받아 args_schema(입력 검증용)와 _run 메서드를 포함한 필요한 속성을 정의합니다.

tool 데코레이터 사용

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"

@tool 데코레이터를 쓰면 함수 안에서 도구의 속성과 기능을 직접 정의할 수 있어, 자신의 필요에 맞는 특수 도구를 간결하고 효율적으로 만들 수 있습니다.

모범 사례: 타입화된 출력 정의하기

도구가 구조화된 데이터를 반환할 때는 Pydantic 출력 모델을 정의하세요. 그러면 에이전트가 평문 텍스트에서 추측하는 대신 결과를 명확한 필드로 읽을 수 있습니다. 타입화된 출력은 ID, 상태 값, 점수, 가격, 목록처럼 필드가 안정적인 결과에 유용합니다. 짧은 산문 결과에는 평문 문자열도 문제없습니다.

직접 Python 호출은 여전히 도구가 반환하는 값을 받습니다. 에이전트가 타입화된 도구를 사용하면 CrewAI는 출력 모델에 기반한 JSON을 에이전트에게 보냅니다.

Pydantic 모델 반환하기

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

tool = InventoryTool()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

{"sku":"SKU-123","quantity":14,"needs_reorder":false}

BaseTool에 Pydantic 반환 타입 애노테이션이 있으면 CrewAI가 출력 스키마를 추론합니다.

에이전트가 InventoryTool을 호출하면 다음과 같은 JSON을 받습니다.

딕셔너리 결과에는 result_schema 사용

from crewai.tools import tool
from pydantic import BaseModel, Field

class ProductResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    name: str = Field(description="The product name.")
    in_stock: bool = Field(description="Whether the product is available.")

@tool("Product Lookup", result_schema=ProductResult)
def product_lookup(sku: str) -> dict[str, object]:
    """Look up product availability by SKU."""
    catalog = {
        "SKU-123": ("Noise-canceling headset", True),
        "SKU-456": ("USB-C dock", False),
    }
    name, in_stock = catalog.get(sku, ("Unknown product", False))
    return {
        "sku": sku,
        "name": name,
        "in_stock": in_stock,
    }

도구가 딕셔너리를 반환하면 result_schema를 명시적으로 설정합니다. BaseTool 서브클래스나 @tool 데코레이터에서 할 수 있습니다.

에이전트에게 보낼 텍스트 커스터마이즈

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class InventoryResult(BaseModel):
    sku: str = Field(description="The product SKU.")
    quantity: int = Field(description="Units available.")
    needs_reorder: bool = Field(description="Whether the item should be reordered.")

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Check current stock for a product SKU."

    def _run(self, sku: str) -> InventoryResult:
        quantity = {"SKU-123": 14, "SKU-456": 0}.get(sku, 0)
        return InventoryResult(sku=sku, quantity=quantity, needs_reorder=quantity < 5)

    def format_output_for_agent(self, raw_result: object) -> str:
        result = InventoryResult.model_validate(raw_result)
        status = "reorder needed" if result.needs_reorder else "stock is healthy"
        return f"{result.sku}: {result.quantity} units. {status}."

tool = InventoryTool()
result = tool.run(sku="SKU-123")

# Direct Python calls receive the raw Pydantic object.
print(result.quantity)

기본적으로 타입화된 도구 출력은 에이전트에게 JSON으로 전달됩니다. 에이전트가 짧은 요약을 받아야 한다면 BaseTool을 서브클래싱해 format_output_for_agent를 오버라이드하세요. 이 오버라이드는 에이전트가 보는 것만 바꿀 뿐, tool.run(...)을 직접 호출하면 여전히 일반 Python 값을 반환합니다.

도구용 캐시 함수 정의

@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

도구 성능을 캐싱으로 최적화하려면 cache_function 속성으로 커스텀 캐싱 전략을 정의합니다.

비동기 도구 만들기

CrewAI는 논블로킹 I/O 연산을 위한 비동기 도구를 지원합니다. HTTP 요청, 데이터베이스 쿼리, 그 외 I/O 바운드 연산을 해야 하는 도구에 유용합니다.

@tool 데코레이터와 async 함수

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()

async 도구를 만드는 가장 간단한 방법은 async 함수에 @tool 데코레이터를 쓰는 것입니다.

BaseTool 서브클래싱으로 async 지원

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()

더 많은 제어가 필요하면 BaseTool을 서브클래싱해 _run(동기)과 _arun(비동기) 메서드를 모두 구현합니다.

이 지침을 따르고 협업 도구와 새 기능을 도구 생성·관리 프로세스에 녹여내면, CrewAI 프레임워크의 역량을 최대한 활용해 개발 경험과 AI 에이전트의 효율을 모두 높일 수 있습니다.

더 알아보기