Integrating CrewAI with LangChain tools

CrewAI agents become far more capable when they can reach into the LangChain ecosystem. The crewai langchain tools integration lets you reuse battle-tested utilities — web search, SQL execution, API wrappers — without rewriting them. This guide walks through the mechanics of adapting a LangChain BaseTool so CrewAI recognizes it, then wires it into a working multi-agent flow you can run today.

Prerequisites

• Python 3.10+
• An OpenAI API key (or any LLM provider CrewAI supports)
• Familiarity with basic CrewAI concepts: agents, tasks, crews

Install the required packages:

pip install crewai langchain langchain-community langchain-openai

If you prefer a different LLM backend, swap langchain-openai for the appropriate provider package.

Step 1: Create a minimal LangChain tool

LangChain tools implement BaseTool with a synchronous _run method (and optionally _arun for async). Start with something self-contained — a calculator — so you can verify the integration without external dependencies.

# tools/calculator.py
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import math
import re


class CalculatorInput(BaseModel):
    expression: str = Field(description="A mathematical expression to evaluate, e.g. '2 * (3 + 4)'")


class CalculatorTool(BaseTool):
    name: str = "calculator"
    args_schema: type[BaseModel] = CalculatorInput

    def _run(self, expression: str) -> str:
        # Safe eval: only allow digits, operators, parentheses, dots, and math.* functions
        allowed = re.compile(r"^[0-9+\-*/().\s]+$|^math\.[a-zA-Z_]+\(.*\)$")
        if not allowed.match(expression.replace(" ", "")):
            return "Error: expression contains disallowed characters"

        try:
            # Provide math module functions in a restricted namespace
            namespace = {k: getattr(math, k) for k in dir(math) if not k.startswith("_")}
            result = eval(expression, {"__builtins__": {}}, namespace)
            return str(result)
        except Exception as e:
            return f"Error: {e}"

    async def _arun(self, expression: str) -> str:
        return self._run(expression)

Verify: Run python -c "from tools.calculator import CalculatorTool; print(CalculatorTool()._run('2 * (3 + 4)'))" — you should see 14.

Step 2: Wrap the LangChain tool for CrewAI

CrewAI expects tools to be callables with a name, description, and a __call__ (or run) method that accepts a single string argument. The simplest adapter is a thin class that delegates to the LangChain tool’s _run.

# tools/crewai_adapter.py
from typing import Any
from langchain.tools import BaseTool


class CrewAIToolWrapper:
    """
    Wrap a LangChain BaseTool so CrewAI can invoke it.
    CrewAI passes a single string; we forward it to the tool's _run.
    """

    def __init__(self, langchain_tool: BaseTool):
        self._tool = langchain_tool
        self.name = langchain_tool.name
        self.description = langchain_tool.description

    def __call__(self, tool_input: str) -> str:
        # CrewAI may pass JSON or plain text; handle both
        try:
            import json
            parsed = json.loads(tool_input)
            # If the tool expects a single arg named 'expression', extract it
            if isinstance(parsed, dict) and "expression" in parsed:
                return self._tool._run(parsed["expression"])
        except Exception:
            pass
        # Fallback: treat the whole string as the expression
        return self._tool._run(tool_input)

    # CrewAI also checks for a `run` method in some versions
    def run(self, tool_input: str) -> str:
        return self.__call__(tool_input)

Why this shape? CrewAI’s Agent class inspects tool.name and tool.description for prompt construction, then calls tool(input_string). The wrapper preserves metadata and normalizes the calling convention.

Step 3: Build a CrewAI agent that uses the wrapped tool

Now instantiate the LangChain tool, wrap it, and hand it to an agent. We’ll create a “Math Specialist” that solves word problems by extracting expressions and calling the calculator.

# agents/math_agent.py
from crewai import Agent
from tools.calculator import CalculatorTool
from tools.crewai_adapter import CrewAIToolWrapper


def create_math_agent(llm) -> Agent:
    calc_tool = CalculatorTool()
    wrapped = CrewAIToolWrapper(calc_tool)

    return Agent(
        role="Math Specialist",
        goal="Solve mathematical word problems accurately by extracting expressions and computing results",
        backstory=(
            "You are a precise mathematician who breaks down word problems into "
            "clean arithmetic expressions. You never guess — you calculate."
        ),
        tools=[wrapped],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

Step 4: Define a task and assemble the crew

A single-agent crew is enough to prove the integration. The task asks the agent to solve a problem that requires the calculator.

# main.py
import os
from crewai import Task, Crew, Process
from langchain_openai import ChatOpenAI
from agents.math_agent import create_math_agent


def main():
    # Configure your LLM — swap base_url/model for other providers
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0,
        api_key=os.getenv("OPENAI_API_KEY"),
    )

    math_agent = create_math_agent(llm)

    task = Task(
        description=(
            "A train leaves Chicago at 60 mph. Another leaves St. Louis at 80 mph. "
            "The cities are 300 miles apart. How many hours until they meet? "
            "Use the calculator tool to compute the exact answer."
        ),
        expected_output="A single number: the hours until the trains meet, with one decimal place.",
        agent=math_agent,
    )

    crew = Crew(
        agents=[math_agent],
        tasks=[task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    print("\n=== FINAL RESULT ===")
    print(result)


if __name__ == "__main__":
    main()

Verify: Export your API key and run python main.py. You should see the agent reason about relative speed (60 + 80 = 140 mph), call the calculator with 300 / 140, and return 2.1 (or 2.142857... depending on rounding).

Step 5: Add a second agent to demonstrate tool sharing

CrewAI shines with multiple agents. Let’s add a “Problem Decomposer” that breaks complex scenarios into sub-problems, then hands each to the Math Specialist. Both agents share the same wrapped calculator tool.

# agents/decomposer_agent.py
from crewai import Agent
from tools.calculator import CalculatorTool
from tools.crewai_adapter import CrewAIToolWrapper


def create_decomposer_agent(llm) -> Agent:
    calc_tool = CalculatorTool()
    wrapped = CrewAIToolWrapper(calc_tool)

    return Agent(
        role="Problem Decomposer",
        goal="Break multi-step word problems into independent sub-problems the Math Specialist can solve",
        backstory=(
            "You excel at reading a complex scenario, identifying the distinct calculations needed, "
            "and writing clear sub-questions. You do not solve them yourself — you delegate."
        ),
        tools=[wrapped],  # Decomposer can also use the calculator for quick checks
        llm=llm,
        verbose=True,
        allow_delegation=True,  # Critical: allows this agent to hand off to others
    )

Update main.py to use both agents and a hierarchical process:

# main.py (updated)
import os
from crewai import Task, Crew, Process
from langchain_openai import ChatOpenAI
from agents.math_agent import create_math_agent
from agents.decomposer_agent import create_decomposer_agent


def main():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0,
        api_key=os.getenv("OPENAI_API_KEY"),
    )

    math_agent = create_math_agent(llm)
    decomposer = create_decomposer_agent(llm)

    # Task 1: Decomposer breaks the problem
    decompose_task = Task(
        description=(
            "A farmer has 3 fields. Field A: 120 acres at $4,500/acre. "
            "Field B: 85 acres at $5,200/acre. Field C: 200 acres at $3,800/acre. "
            "The farmer sells Field B and buys 50 more acres at $4,100/acre. "
            "What is the total value of the farmer's land after these transactions? "
            "Break this into sub-problems and delegate each to the Math Specialist."
        ),
        expected_output="A list of sub-problems with their computed answers, and the final total value.",
        agent=decomposer,
    )

    # Task 2: Math Specialist solves each sub-problem (delegated automatically)
    solve_task = Task(
        description="Solve the sub-problem assigned to you using the calculator tool.",
        expected_output="The numeric result of the assigned calculation.",
        agent=math_agent,
    )

    crew = Crew(
        agents=[decomposer, math_agent],
        tasks=[decompose_task, solve_task],
        process=Process.hierarchical,  # Decomposer manages delegation
        manager_llm=llm,
        verbose=True,
    )

    result = crew.kickoff()
    print("\n=== FINAL RESULT ===")
    print(result)


if __name__ == "__main__":
    main()

Run it again. The Decomposer should identify three calculations (value of A, value of C, value of new 50 acres), delegate each, and sum the results.

Step 6: Use a real LangChain community tool

The calculator was a teaching example. Now swap in a production tool — DuckDuckGoSearchRun — to give agents web access. The same wrapper works unchanged.

# tools/search_tool.py
from langchain_community.tools import DuckDuckGoSearchRun
from tools.crewai_adapter import CrewAIToolWrapper


def make_search_tool() -> CrewAIToolWrapper:
    search = DuckDuckGoSearchRun()
    return CrewAIToolWrapper(search)

Add it to an agent:

# agents/research_agent.py
from crewai import Agent
from tools.search_tool import make_search_tool


def create_research_agent(llm) -> Agent:
    return Agent(
        role="Research Analyst",
        goal="Find current, factual information from the web to answer questions",
        backstory="You are a meticulous researcher who verifies claims against multiple sources.",
        tools=[make_search_tool()],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

Verification: Create a task asking for “the current population of Tokyo” and run the crew. The agent will invoke the search tool, parse snippets, and return a cited answer.

Step 7: Handle structured inputs for complex tools

Some LangChain tools expect multiple named arguments (e.g., SQLDatabaseTool needs query and optionally fetch). The wrapper’s JSON parsing handles this, but you can make it explicit by updating the adapter:

# tools/crewai_adapter.py (enhanced)
    def __call__(self, tool_input: str) -> str:
        import json
        try:
            parsed = json.loads(tool_input)
            if isinstance(parsed, dict):
                # Pass kwargs directly to _run if it accepts them
                import inspect
                sig = inspect.signature(self._tool._run)
                if len(sig.parameters) > 1 or any(p.kind == p.KEYWORD_ONLY for p in sig.parameters.values()):
                    return self._tool._run(**parsed)
        except Exception:
            pass
        return self._tool._run(tool_input)

Now agents can emit {"query": "SELECT * FROM users", "fetch": "all"} and the wrapper forwards correctly.

Common pitfalls and fixes

Testing the integration in isolation

Before running a full crew, unit-test the wrapper:

# test_integration.py
import pytest
from tools.calculator import CalculatorTool
from tools.crewai_adapter import CrewAIToolWrapper


def test_wrapper_passes_through_result():
    tool = CalculatorTool()
    wrapped = CrewAIToolWrapper(tool)
    assert wrapped("2 + 2") == "4"
    assert wrapped.run("10 / 2") == "5.0"


def test_wrapper_handles_json_input():
    tool = CalculatorTool()
    wrapped = CrewAIToolWrapper(tool)
    assert wrapped('{"expression": "3 ** 4"}') == "81"


def test_wrapper_preserves_metadata():
    tool = CalculatorTool()
    wrapped = CrewAIToolWrapper(tool)
    assert wrapped.name == "calculator"
    assert "mathematical" in wrapped.description.lower()

Run with pytest test_integration.py -v.

Where to go next

• Custom LangChain tools: Wrap internal APIs, databases, or proprietary services as BaseTool subclasses, then drop them into CrewAI with the same adapter.
• Async execution: Implement _arun on your LangChain tools and use CrewAI’s async_execution=True on tasks for I/O-bound tools (search, HTTP).
• Tool routing: If you run many tools, consider a router agent that selects the right toolset per task — reduces prompt bloat and hallucination.
• Observability: Log every tool call (input, output, latency) to a centralized store. This is essential for debugging multi-agent loops.

The crewai langchain tools integration is fundamentally a calling-convention bridge. Once you understand the three touchpoints — name, description, and __call__(str) -> str — you can adapt any LangChain tool in minutes. The pattern scales: same wrapper, same agent interface, whether the tool evaluates arithmetic or queries a production data warehouse.