CrewAI 도구 (Tools)

CrewAI에서 에이전트에게 "손과 발"을 달아 주는 게 도구예요. 에이전트는 언어 모델이라 그 자체로는 외부 세계에 아무것도 할 수 없는데요, 도구 덕분에 웹 검색을 하고, 파일을 읽고, 데이터를 분석하고, 심지어 다른 에이전트에게 일을 넘기는 협업까지 할 수 있게 돼요.

이 도구는 에이전트가 호출할 수 있는 함수(callable function) 모음이에요. CrewAI는 도구 외에도 에이전트의 능력을 확장하는 다른 수단들을 함께 제공하는데요, 원격 도구 서버인 MCP, 플랫폼 연동인 Apps, 도메인 전문성인 Skills, 검색해서 얻은 사실인 Knowledge가 그거예요. 도구가 이들과 어울려 쓰인다는 점만 일단 알아 두면 충분해요.

도구의 핵심 특징

도구가 어떤 성격을 갖는지 짚어 볼게요.

• 유틸리티(Utility): 웹 검색, 데이터 분석, 콘텐츠 생성, 에이전트 간 협업 같은 작업을 하기 위해 만들어져요.
• 통합(Integration): 에이전트의 작업 흐름에 자연스럽게 끼워져서 능력을 높여 줘요.
• 커스터마이징(Customizability): 이미 있는 도구를 쓰거나, 필요에 맞게 직접 만들 수도 있어요.
• 오류 처리(Error Handling): 견고한 오류 처리 메커니즘을 기본으로 갖춰서 매끄럽게 동작해요.
• 캐싱(Caching): 똑똑한 캐싱으로 성능을 최적화하고 중복 작업을 줄여 줘요.
• 비동기 지원(Asynchronous Support): 동기/비동기 도구를 모두 처리해서 블로킹 없이 동작할 수 있어요.
• 타입이 있는 출력(Typed Outputs): 선택적으로 Pydantic 모델을 쓰면 에이전트에게 명확한 JSON 필드를 알려 줄 수 있어요. 다만 파이썬에서 직접 호출하면 여전히 도구의 평범한 반환값을 받아요.

CrewAI 도구 사용하기

에이전트에 도구를 달려면 먼저 추가 도구 패키지를 설치해야 해요.

pip install 'crewai[tools]'

설치하고 나면 이렇게 검색·파일 도구를 만들어 에이전트에 연결해서 쓰면 돼요. 아래는 공식 문서의 예시인데, 리서처는 검색 도구, 작가는 파일 도구를 받아서 하나의 크루로 함께 일하는 구조예요.

import os
from crewai import Agent, Task, Crew
# Importing crewAI tools
from crewai_tools import (
    DirectoryReadTool,
    FileReadTool,
    SerperDevTool,
    WebsiteSearchTool
)

# Set up API keys
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key
os.environ["OPENAI_API_KEY"] = "Your Key"

# Instantiate tools
docs_tool = DirectoryReadTool(directory='./blog-posts')
file_tool = FileReadTool()
search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

# Create agents
researcher = Agent(
    role='Market Research Analyst',
    goal='Provide up-to-date market analysis of the AI industry',
    backstory='An expert analyst with a keen eye for market trends.',
    tools=[search_tool, web_rag_tool],
    verbose=True
)

writer = Agent(
    role='Content Writer',
    goal='Craft engaging blog posts about the AI industry',
    backstory='A skilled writer with a passion for technology.',
    tools=[docs_tool, file_tool],
    verbose=True
)

# Define tasks
research = Task(
    description='Research the latest trends in the AI industry and provide a summary.',
    expected_output='A summary of the top 3 trending developments in the AI industry with a unique perspective on their significance.',
    agent=researcher
)

write = Task(
    description="Write an engaging blog post about the AI industry, based on the research analyst's summary. Draw inspiration from the latest blog posts in the directory.",
    expected_output='A 4-paragraph blog post formatted in markdown with engaging, informative, and accessible content, avoiding complex jargon.',
    agent=writer,
    output_file='blog-posts/new_post.md'  # The final blog post will be saved here
)

# Assemble a crew with planning enabled
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    verbose=True,
    planning=True,  # Enable planning feature
)

# Execute tasks
crew.kickoff()

코드에서 주목할 점은 두 가지예요. 첫째, 에이전트마다 tools=[...]로 자기에게 필요한 도구만 골라 준다는 것. 둘째, 크루를 만들 때 planning=True를 켜면 작업을 시작하기 전에 계획 단계를 거친다는 거예요. 리서처의 검색 결과가 작가의 글쓰기 입력으로 흘러가는 흐름을 상상해 보면 도구가 에이전트 사이의 다리 역할을 한다는 게 바로 와닿을 거예요.

어떤 도구들이 있나요

기본 제공 도구는 웹 검색(RAG), 파일·문서 처리, 브라우저 조작, 스크래핑 등으로 나뉘어요. 대표적인 것 몇 가지만 꼽아 볼게요.

• SerperDevTool: Serper API 기반 웹 검색 도구
• WebsiteSearchTool: 웹사이트 콘텐츠를 검색하는 RAG 도구
• FileReadTool / DirectoryReadTool: 파일과 디렉터리 구조를 읽는 도구
• PDFSearchTool / DOCXSearchTool / CSVSearchTool / TXTSearchTool / JSONSearchTool / XMLSearchTool: 각 형식의 문서 안을 검색하는 RAG 도구
• ScrapeWebsiteTool / ScrapeElementFromWebsiteTool: 웹사이트 전체 또는 특정 요소를 스크래핑하는 도구
• YoutubeChannelSearchTool / YoutubeVideoSearchTool: 유튜브 채널·영상 검색 도구

모든 도구는 오류 처리를 갖추고 있고 캐싱을 지원해요. 캐싱 동작은 도구의 cache_function 속성으로 더 세밀하게 제어할 수 있어요.

전체 목록과 최신 도구는 원문에서 확인하는 게 정확해요. 문서가 자주 갱신되다 보니 목록 전체를 여기 옮기기보다는 필요한 시점에 공식 문서를 보는 걸 권해요.

나만의 도구 만들기

에이전트의 필요에 맞춘 커스텀 도구를 만드는 방법은 크게 두 가지예요.

BaseTool 상속해서 만들기

BaseTool을 상속받고 _run 메서드를 구현하면 돼요. args_schema에 Pydantic 모델을 지정해서 입력 형식을 명확히 해 주고요, description은 에이전트가 이 도구를 언제 써야 할지 판단하는 데 쓰이니 정확히 적어 주는 게 중요해요.

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"

타입이 있는 출력 (Typed Tool Outputs)

도구가 구조화된 데이터를 돌려줄 때 Pydantic 출력 모델을 정의하면, 에이전트는 sku, quantity, needs_reorder처럼 신뢰할 수 있는 필드 이름을 받아요. 파이썬에서 직접 호출할 땐 여전히 도구가 실제로 돌려준 값을 받고요, 에이전트가 쓸 때는 CrewAI가 출력 모델을 기반으로 만든 JSON 문자열을 에이전트에게 보내 줘요.

from crewai.tools import BaseTool
from pydantic import BaseModel

class InventoryResult(BaseModel):
    sku: str
    quantity: int
    needs_reorder: bool

class InventoryTool(BaseTool):
    name: str = "Inventory Check"
    description: str = "Checks 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()

# Direct calls receive the raw Pydantic object.
result = tool.run(sku="SKU-123")
print(result.quantity)

에이전트에게 Markdown이나 짧은 텍스트로 보내고 싶다면 format_output_for_agent를 오버라이드하면 돼요. 이때도 tool.run(...)을 직접 호출하면 일반 파이썬 값을 그대로 받아요.

class InventoryTool(BaseTool):

    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}."

format_output_for_agent를 오버라이드하지 않으면 타입이 있는 출력은 에이전트에게 JSON으로 전달되고, 일반 문자열 결과는 기존처럼 동작해요.

tool 데코레이터로 만들기

BaseTool 클래스를 상속하는 대신 함수에 @tool 데코레이터를 붙이는 간단한 방법도 있어요. 데코레이터 인자로 도구 이름을 주고, 독스트링에 이 도구가 무엇에 유용한지 적어 주면 에이전트가 그걸 읽고 사용 여부를 판단해요.

from crewai.tools import tool
@tool("Name of my tool")
def my_tool(question: str) -> str:
    """Clear description for what this tool is useful for, your agent will need this information to use it."""
    # Function logic here
    return "Result from your custom tool"

비동기 도구 만들기

네트워크 요청이나 파일 I/O처럼 오래 걸리는 작업은 메인 스레드를 막지 않게 비동기로 구현할 수 있어요. 방법은 두 가지가 있는데요, tool 데코레이터에 비동기 함수를 붙이거나, 커스텀 클래스에서 async def _run을 구현하면 돼요.

from crewai.tools import tool

@tool("fetch_data_async")
async def fetch_data_async(query: str) -> str:
    """Asynchronously fetch data based on the query."""
    # Simulate async operation
    await asyncio.sleep(1)
    return f"Data retrieved for {query}"

from crewai.tools import BaseTool

class AsyncCustomTool(BaseTool):
    name: str = "async_custom_tool"
    description: str = "An asynchronous custom tool"

    async def _run(self, query: str = "") -> str:
        """Asynchronously run the tool"""
        # Your async implementation here
        await asyncio.sleep(1)
        return f"Processed {query} asynchronously"

비동기 도구는 일반 크루 워크플로우와 Flow 워크플로우 어디서든 그대로 동작하고, CrewAI가 동기·비동기 실행을 알아서 처리해 주니 호출 방법을 다르게 신경 쓸 필요가 없어요.

커스텀 캐싱: cache_function

도구가 결과를 언제 캐시할지 cache_function으로 세밀하게 제어할 수 있어요. 아래 예시는 곱셈 결과가 2의 배수일 때만 캐시하도록 만든 거예요.

from crewai.tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """Useful for when you need to multiply two numbers together."""
    return first_number * second_number

def cache_func(args, result):
    # In this case, we only cache the result if it's a multiple of 2
    cache = result % 2 == 0
    return cache

multiplication_tool.cache_function = cache_func

writer1 = Agent(
        role="Writer",
        goal="You write lessons of math for kids.",
        backstory="You're an expert in writing and you love to teach kids but you know nothing of math.",
        tools=[multiplication_tool],
        allow_delegation=False,
    )
    #...

도구 실패 보고하기 (Tool Failure)

도구가 예외를 던지지 않고 "성공적으로 끝났는데" 사실은 요청한 일을 못 한 경우가 있어요. 예를 들어 Slack이 HTTP 200으로 응답하면서 본문엔 {"ok": false, "error": "channel_not_found"}를 돌려준다든지, MCP 서버가 isError를 세운다든지 하는 경우예요. 호출 자체는 "동작"했으니 오류 텍스트가 평범한 결과처럼 에이전트에게 전달되고, 에이전트는 마지막 답변에서 문제를 설명하게 되고 실행은 성공으로 기록돼요.

이때 오류 문자열 대신 ToolFailure를 돌려주면 프레임워크가 일반 결과와 실패를 구분할 수 있어요.

from typing import Any

from crewai.tools import BaseTool
from crewai.tools.tool_failure import ToolFailure

class SendSlackMessage(BaseTool):
    name: str = "send_slack_message"
    description: str = "Post a message to a Slack channel."

    def _run(self, channel: str, text: str) -> Any:
        payload = slack.post(channel=channel, text=text)
        if not payload["ok"]:
            return ToolFailure(
                message=f"Slack rejected the message: {payload['error']}",
                code=payload["error"],
                retryable=payload["error"] == "rate_limited",
            )
        return payload

에이전트는 여전히 평범한 문장(ToolFailure.as_agent_message()가 메시지를 렌더링)을 읽으니 모델 동작은 바뀌지 않아요. 달라지는 건 실패가 이제 다운스트림 어디서든 보인다는 점이에요.

중요한 점: 실패 탐지는 선언적(declarative) 방식이에요. CrewAI는 문자열이 "오류처럼 보이는지"를 추측하지 않으니, 진짜로 오류를 텍스트로 돌려주는 도구가 실패로 오해받는 일은 없어요. 실패로 기록되는 경우는 ToolFailure를 반환했을 때, 도구가 예외를 던졌을 때, MCP 서버가 isError를 설정했을 때, 도구의 max_usage_count가 소진됐을 때, 또는 에이전트가 존재하지 않는 도구를 호출했을 때예요.

실패 정책 고르기

tool_failure_policy가 실패 후 무엇을 할지 결정해요.

Policy	Behavior
ignore	아무것도 기록·발생·조치하지 않아요.
warn (기본값)	실패를 기록하고 ToolFailureDetectedEvent를 발생시킨 뒤 계속 진행해요.
raise	기록하고 발생시킨 뒤 ToolExecutionFailedError로 중단해요.

from crewai import Agent, Crew, Task
from crewai.tools.tool_failure import ToolFailurePolicy

agent = Agent(
    role="Slack Messenger",
    goal="Post the report to Slack",
    backstory="...",
    tools=[SendSlackMessage()],
    tool_failure_policy=ToolFailurePolicy.WARN,
)

# Tighten a single high-stakes task without changing the agent.
task = Task(
    description="Post the final report to #engineering",
    expected_output="Confirmation the message was posted",
    agent=agent,
    tool_failure_policy=ToolFailurePolicy.RAISE,
)

# Or set a baseline once for every agent in the crew.
crew = Crew(
    agents=[agent],
    tasks=[task],
    tool_failure_policy=ToolFailurePolicy.WARN,
)

설정 우선순위는 가장 구체적인 쪽이 이겨요: tool → task → agent → crew → warn. 각 단계의 기본값은 None이라 "바깥쪽에서 상속"한다는 뜻이고, 아무것도 설정하지 않으면 실제 기본값은 warn이 돼요.

실패 확인하기

기록된 실패는 구조화돼 있어서 다운스트림이 문자열을 파싱할 필요가 없어요.

result = crew.kickoff()

if result.has_tool_failures:
    for record in result.tool_failures:
        print(record.tool_name)          # "send_slack_message"
        print(record.failure.code)       # "channel_not_found"
        print(record.failure.reason)     # ToolFailureReason.TOOL_REPORTED
        print(record.summary())

tool_failures는 TaskOutput, CrewOutput, LiteAgentOutput에서 모두 사용할 수 있어요. 크루가 비어 있지 않은 목록과 함께 성공적으로 끝날 수도 있으니, raw를 완전한 결과로 취급하기 전에 이걸 확인해야 해요.

실패가 발생하는 즉시 반응하고 싶다면 이벤트를 구독하면 돼요.

from crewai.events import ToolFailureDetectedEvent
from crewai.events.event_bus import crewai_event_bus

@crewai_event_bus.on(ToolFailureDetectedEvent)
def on_tool_failure(source, event):
    print(f"{event.tool_name} failed: {event.failure.message} ({event.policy})")

이 이벤트는 raise 정책이 중단되기 전에 발생하므로 구독자는 항상 실패를 관찰할 수 있어요. ToolUsageFinishedEvent에도 failure 필드가 있어서, 추적 UI가 두 이벤트를 상관시키지 않고도 호출을 실패로 표시할 수 있어요.

마무리

도구는 CrewAI 에이전트의 능력을 크게 넓혀 주는 핵심이에요. 웹 검색부터 데이터 분석, 그리고 에이전트 간 협업까지 다양한 작업을 가능하게 하죠. CrewAI로 뭔가를 만들 때는 기존 도구와 커스텀 도구를 함께 활용하고, 오류 처리·캐싱·유연한 도구 인자까지 고려하면 에이전트의 성능과 능력을 훨씬 잘 끌어낼 수 있어요.

데이터스케쳐스 실무 관점

도구는 곧 에이전트가 실제 시스템에 닿는 "저수준 진입점"이라, 우리가 데이터 파이프라인이나 문서 작업을 에이전트에 맡길 때 가장 먼저 손이 가는 곳이에요. 실무에서 몇 가지를 유의하면 좋아요.

• 권한의 최소화: 에이전트에 달 도구는 가능한 한 적고 구체적으로. 읽기 전용 도구(FileReadTool, DirectoryReadTool)만 필요한 작업에 검색이나 스크래핑 도구까지 몽땅 넣으면, 모델이 원치 않는 행동을 할 여지가 커져요. 작가 에이전트에는 파일 도구만, 리서처에는 검색 도구만 주는 공식 예시가 좋은 기준이에요.
• 캐싱 비용: 기본 캐싱은 비용과 응답 시간을 줄여 주지만, 민감하거나 자주 바뀌는 데이터(예: 실시간 시세, 데이터 카탈로그 상태)를 캐시하면 낡은 결과로 작업할 수 있어요. cache_function으로 캐시 조건을 좁혀 주는 걸 검토하세요.
• 실패 정책을 명시적으로: 기본 정책이 warn이라 실패를 기록해도 흐름은 이어져요. 재무 보고나 배포처럼 실패가 치명적인 작업은 raise로 단단히 막고, 단순 알림 같은 건 warn으로 두는 식으로 작업의 성격에 맞춰 정책을 나누면 좋아요. result.has_tool_failures를 체크해서 "성공했지만 도구 실패가 있었던" 실행을 걸러내는 습관도 들이는 걸 권해요.
• 타입 출력 활용: 구조화된 결과를 돌려줄 때 Pydantic 출력 모델을 정의하면 에이전트가 필드 이름을 신뢰할 수 있고, 다운스트림 파싱도 그만큼 단순해져요. 데이터셋 스키마나 인벤토리 상태처럼 형태가 고정된 응답은 typed output으로 다루는 게 실무적으로 깔끔해요.

더 알아보기

• 원문: Tools - CrewAI
• 관련 개념: MCP (원격 도구 서버), Agent Capabilities, Skills, Knowledge
• 도구 패키지·통합: crewai-tools (GitHub), LangChain Tools