crewAI/docs/how-to/bring-your-own-agent.mdx

---
title: Bring your own agent
description: Learn how to bring your own agents that work within a Crew.
icon: robots
---

Interoperability is a core concept in CrewAI. This guide will show you how to bring your own agents that work within a Crew.


## Adapter Guide for Bringing your own agents (Langgraph Agents, OpenAI Agents, etc...)
We require 3 adapters to turn any agent from different frameworks to work within crew.

1. BaseAgentAdapter
2. BaseToolAdapter
3. BaseConverter


## BaseAgentAdapter
This abstract class defines the common interface and functionality that all
agent adapters must implement. It extends BaseAgent to maintain compatibility
with the CrewAI framework while adding adapter-specific requirements.

Required Methods:

1. `def configure_tools`
2. `def configure_structured_output`

## Creating your own Adapter
To integrate an agent from a different framework (e.g., LangGraph, Autogen, OpenAI Assistants) into CrewAI, you need to create a custom adapter by inheriting from `BaseAgentAdapter`. This adapter acts as a compatibility layer, translating between the CrewAI interfaces and the specific requirements of your external agent.

Here's how you implement your custom adapter:

1.  **Inherit from `BaseAgentAdapter`**:
    ```python
    from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
    from crewai.tools import BaseTool
    from typing import List, Optional, Any, Dict

    class MyCustomAgentAdapter(BaseAgentAdapter):
        # ... implementation details ...
    ```

2.  **Implement `__init__`**:
    The constructor should call the parent class constructor `super().__init__(**kwargs)` and perform any initialization specific to your external agent. You can use the optional `agent_config` dictionary passed during CrewAI's `Agent` initialization to configure your adapter and the underlying agent.

    ```python
    def __init__(self, agent_config: Optional[Dict[str, Any]] = None, **kwargs: Any):
        super().__init__(agent_config=agent_config, **kwargs)
        # Initialize your external agent here, possibly using agent_config
        # Example: self.external_agent = initialize_my_agent(agent_config)
        print(f"Initializing MyCustomAgentAdapter with config: {agent_config}")
    ```

3.  **Implement `configure_tools`**:
    This abstract method is crucial. It receives a list of CrewAI `BaseTool` instances. Your implementation must convert or adapt these tools into the format expected by your external agent framework. This might involve wrapping them, extracting specific attributes, or registering them with the external agent instance.

    ```python
    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
        if tools:
            adapted_tools = []
            for tool in tools:
                # Adapt CrewAI BaseTool to the format your agent expects
                # Example: adapted_tool = adapt_to_my_framework(tool)
                # adapted_tools.append(adapted_tool)
                pass # Replace with your actual adaptation logic

            # Configure the external agent with the adapted tools
            # Example: self.external_agent.set_tools(adapted_tools)
            print(f"Configuring tools for MyCustomAgentAdapter: {adapted_tools}") # Placeholder
        else:
            # Handle the case where no tools are provided
            # Example: self.external_agent.set_tools([])
            print("No tools provided for MyCustomAgentAdapter.")
    ```

4.  **Implement `configure_structured_output`**:
    This method is called when the CrewAI `Agent` is configured with structured output requirements (e.g., `output_json` or `output_pydantic`). Your adapter needs to ensure the external agent is set up to comply with these requirements. This might involve setting specific parameters on the external agent or ensuring its underlying model supports the requested format. If the external agent doesn't support structured output in a way compatible with CrewAI's expectations, you might need to handle the conversion or raise an appropriate error.

    ```python
    def configure_structured_output(self, structured_output: Any) -> None:
        # Configure your external agent to produce output in the specified format
        # Example: self.external_agent.set_output_format(structured_output)
        self.adapted_structured_output = True # Signal that structured output is handled
        print(f"Configuring structured output for MyCustomAgentAdapter: {structured_output}")
    ```

By implementing these methods, your `MyCustomAgentAdapter` will allow your custom agent implementation to function correctly within a CrewAI crew, interacting with tasks and tools seamlessly. Remember to replace the example comments and print statements with your actual adaptation logic specific to the external agent framework you are integrating.

## BaseToolAdapter implementation
The `BaseToolAdapter` class is responsible for converting CrewAI's native `BaseTool` objects into a format that your specific external agent framework can understand and utilize. Different agent frameworks (like LangGraph, OpenAI Assistants, etc.) have their own unique ways of defining and handling tools, and the `BaseToolAdapter` acts as the translator.

Here's how you implement your custom tool adapter:

1.  **Inherit from `BaseToolAdapter`**:
    ```python
    from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
    from crewai.tools import BaseTool
    from typing import List, Any

    class MyCustomToolAdapter(BaseToolAdapter):
        # ... implementation details ...
    ```

2.  **Implement `configure_tools`**:
    This is the core abstract method you must implement. It receives a list of CrewAI `BaseTool` instances provided to the agent. Your task is to iterate through this list, adapt each `BaseTool` into the format expected by your external framework, and store the converted tools in the `self.converted_tools` list (which is initialized in the base class constructor).

    ```python
    def configure_tools(self, tools: List[BaseTool]) -> None:
        """Configure and convert CrewAI tools for the specific implementation."""
        self.converted_tools = [] # Reset in case it's called multiple times
        for tool in tools:
            # Sanitize the tool name if required by the target framework
            sanitized_name = self.sanitize_tool_name(tool.name)

            # --- Your Conversion Logic Goes Here ---
            # Example: Convert BaseTool to a dictionary format for LangGraph
            # converted_tool = {
            #     "name": sanitized_name,
            #     "description": tool.description,
            #     "parameters": tool.args_schema.schema() if tool.args_schema else {},
            #     # Add any other framework-specific fields
            # }

            # Example: Convert BaseTool to an OpenAI function definition
            # converted_tool = {
            #     "type": "function",
            #     "function": {
            #         "name": sanitized_name,
            #         "description": tool.description,
            #         "parameters": tool.args_schema.schema() if tool.args_schema else {"type": "object", "properties": {}},
            #     }
            # }

            # --- Replace above examples with your actual adaptation ---
            converted_tool = self.adapt_tool_to_my_framework(tool, sanitized_name) # Placeholder

            self.converted_tools.append(converted_tool)
            print(f"Adapted tool '{tool.name}' to '{sanitized_name}' for MyCustomToolAdapter") # Placeholder

        print(f"MyCustomToolAdapter finished configuring tools: {len(self.converted_tools)} adapted.") # Placeholder

    # --- Helper method for adaptation (Example) ---
    def adapt_tool_to_my_framework(self, tool: BaseTool, sanitized_name: str) -> Any:
        # Replace this with the actual logic to convert a CrewAI BaseTool
        # to the format needed by your specific external agent framework.
        # This will vary greatly depending on the target framework.
        adapted_representation = {
            "framework_specific_name": sanitized_name,
            "framework_specific_description": tool.description,
            "inputs": tool.args_schema.schema() if tool.args_schema else None,
            "implementation_reference": tool.run # Or however the framework needs to call it
        }
        # Also ensure the tool works both sync and async
        async def async_tool_wrapper(*args, **kwargs):
            output = tool.run(*args, **kwargs)
            if inspect.isawaitable(output):
                return await output
            else:
                return output

        adapted_tool = MyFrameworkTool(
            name=sanitized_name,
            description=tool.description,
            inputs=tool.args_schema.schema() if tool.args_schema else None,
            implementation_reference=async_tool_wrapper
        )
        
        return adapted_representation

    ```

3.  **Using the Adapter**:
    Typically, you would instantiate your `MyCustomToolAdapter` within your `MyCustomAgentAdapter`'s `configure_tools` method and use it to process the tools before configuring your external agent.

    ```python
    # Inside MyCustomAgentAdapter.configure_tools
    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
        if tools:
            tool_adapter = MyCustomToolAdapter() # Instantiate your tool adapter
            tool_adapter.configure_tools(tools)  # Convert the tools
            adapted_tools = tool_adapter.tools() # Get the converted tools

            # Now configure your external agent with the adapted_tools
            # Example: self.external_agent.set_tools(adapted_tools)
            print(f"Configuring external agent with adapted tools: {adapted_tools}") # Placeholder
        else:
            # Handle no tools case
            print("No tools provided for MyCustomAgentAdapter.")
    ```

By creating a `BaseToolAdapter`, you decouple the tool conversion logic from the agent adaptation, making the integration cleaner and more modular. Remember to replace the placeholder examples with the actual conversion logic required by your specific external agent framework.

## BaseConverter
The `BaseConverterAdapter` plays a crucial role when a CrewAI `Task` requires an agent to return its final output in a specific structured format, such as JSON or a Pydantic model. It bridges the gap between CrewAI's structured output requirements and the capabilities of your external agent.

Its primary responsibilities are:
1.  **Configuring the Agent for Structured Output:** Based on the `Task`'s requirements (`output_json` or `output_pydantic`), it instructs the associated `BaseAgentAdapter` (and indirectly, the external agent) on what format is expected.
2.  **Enhancing the System Prompt:** It modifies the agent's system prompt to include clear instructions on *how* to generate the output in the required structure.
3.  **Post-processing the Result:** It takes the raw output from the agent and attempts to parse, validate, and format it according to the required structure, ultimately returning a string representation (e.g., a JSON string).

Here's how you implement your custom converter adapter:

1.  **Inherit from `BaseConverterAdapter`**:
    ```python
    from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
    # Assuming you have your MyCustomAgentAdapter defined
    # from .my_custom_agent_adapter import MyCustomAgentAdapter
    from crewai.task import Task
    from typing import Any

    class MyCustomConverterAdapter(BaseConverterAdapter):
        # Store the expected output type (e.g., 'json', 'pydantic', 'text')
        _output_type: str = 'text' 
        _output_schema: Any = None # Store JSON schema or Pydantic model

        # ... implementation details ...
    ```

2.  **Implement `__init__`**:
    The constructor must accept the corresponding `agent_adapter` instance it will work with.

    ```python
    def __init__(self, agent_adapter: Any): # Use your specific AgentAdapter type hint
        self.agent_adapter = agent_adapter
        print(f"Initializing MyCustomConverterAdapter for agent adapter: {type(agent_adapter).__name__}")
    ```

3.  **Implement `configure_structured_output`**:
    This method receives the CrewAI `Task` object. You need to check the task's `output_json` and `output_pydantic` attributes to determine the required output structure. Store this information (e.g., in `_output_type` and `_output_schema`) and potentially call configuration methods on your `self.agent_adapter` if the external agent needs specific setup for structured output (which might have been partially handled in the agent adapter's `configure_structured_output` already).

    ```python
    def configure_structured_output(self, task: Task) -> None:
        """Configure the expected structured output based on the task."""
        if task.output_pydantic:
            self._output_type = 'pydantic'
            self._output_schema = task.output_pydantic
            print(f"Converter: Configured for Pydantic output: {self._output_schema.__name__}")
        elif task.output_json:
            self._output_type = 'json'
            self._output_schema = task.output_json
            print(f"Converter: Configured for JSON output with schema: {self._output_schema}")
        else:
            self._output_type = 'text'
            self._output_schema = None
            print("Converter: Configured for standard text output.")

        # Optionally, inform the agent adapter if needed
        # self.agent_adapter.set_output_mode(self._output_type, self._output_schema)
    ```

4.  **Implement `enhance_system_prompt`**:
    This method takes the agent's base system prompt string and should append instructions tailored to the currently configured `_output_type` and `_output_schema`. The goal is to guide the LLM powering the agent to produce output in the correct format.

    ```python
    def enhance_system_prompt(self, base_prompt: str) -> str:
        """Enhance the system prompt with structured output instructions."""
        if self._output_type == 'text':
            return base_prompt # No enhancement needed for plain text

        instructions = "\n\nYour final answer MUST be formatted as "
        if self._output_type == 'json':
            schema_str = json.dumps(self._output_schema, indent=2)
            instructions += f"a JSON object conforming to the following schema:\n```json\n{schema_str}\n```"
        elif self._output_type == 'pydantic':
            schema_str = json.dumps(self._output_schema.model_json_schema(), indent=2)
            instructions += f"a JSON object conforming to the Pydantic model '{self._output_schema.__name__}' with the following schema:\n```json\n{schema_str}\n```"

        instructions += "\nEnsure your entire response is ONLY the valid JSON object, without any introductory text, explanations, or concluding remarks."
        
        print(f"Converter: Enhancing prompt for {self._output_type} output.")
        return base_prompt + instructions
    ```
    *Note: The exact prompt engineering might need tuning based on the agent/LLM being used.* 

5.  **Implement `post_process_result`**:
    This method receives the raw string output from the agent. If structured output was requested (`json` or `pydantic`), you should attempt to parse the string into the expected format. Handle potential parsing errors (e.g., log them, attempt simple fixes, or raise an exception). Crucially, the method must **always return a string**, even if the intermediate format was a dictionary or Pydantic object (e.g., by serializing it back to a JSON string).

    ```python
    import json
    from pydantic import ValidationError

    def post_process_result(self, result: str) -> str:
        """Post-process the agent's result to ensure it matches the expected format."""
        print(f"Converter: Post-processing result for {self._output_type} output.")
        if self._output_type == 'json':
            try:
                # Attempt to parse and re-serialize to ensure validity and consistent format
                parsed_json = json.loads(result)
                # Optional: Validate against self._output_schema if it's a JSON schema dictionary
                # from jsonschema import validate
                # validate(instance=parsed_json, schema=self._output_schema)
                return json.dumps(parsed_json)
            except json.JSONDecodeError as e:
                print(f"Error: Failed to parse JSON output: {e}\nRaw output:\n{result}")
                # Handle error: return raw, raise exception, or try to fix
                return result # Example: return raw output on failure
            # except Exception as e: # Catch validation errors if using jsonschema
            #     print(f"Error: JSON output failed schema validation: {e}\nRaw output:\n{result}")
            #     return result
        elif self._output_type == 'pydantic':
            try:
                # Attempt to parse into the Pydantic model
                model_instance = self._output_schema.model_validate_json(result)
                # Return the model serialized back to JSON
                return model_instance.model_dump_json()
            except ValidationError as e:
                print(f"Error: Failed to validate Pydantic output: {e}\nRaw output:\n{result}")
                # Handle error
                return result # Example: return raw output on failure
            except json.JSONDecodeError as e:
                 print(f"Error: Failed to parse JSON for Pydantic model: {e}\nRaw output:\n{result}")
                 return result
        else: # 'text'
            return result # No processing needed for plain text
    ```

By implementing these methods, your `MyCustomConverterAdapter` ensures that structured output requests from CrewAI tasks are correctly handled by your integrated external agent, improving the reliability and usability of your custom agent within the CrewAI framework.

## Out of the Box Adapters

We provide out of the box adapters for the following frameworks:
1. LangGraph
2. OpenAI Agents

## Kicking off a crew with adapted agents:

```python
import json
import os
from typing import List

from crewai_tools import SerperDevTool
from src.crewai import Agent, Crew, Task
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

from crewai.agents.agent_adapters.langgraph.langgraph_adapter import (
    LangGraphAgentAdapter,
)
from crewai.agents.agent_adapters.openai_agents.openai_adapter import OpenAIAgentAdapter

# CrewAI Agent
code_helper_agent = Agent(
    role="Code Helper",
    goal="Help users solve coding problems effectively and provide clear explanations.",
    backstory="You are an experienced programmer with deep knowledge across multiple programming languages and frameworks. You specialize in solving complex coding challenges and explaining solutions clearly.",
    allow_delegation=False,
    verbose=True,
)
# OpenAI Agent Adapter
link_finder_agent = OpenAIAgentAdapter(
    role="Link Finder",
    goal="Find the most relevant and high-quality resources for coding tasks.",
    backstory="You are a research specialist with a talent for finding the most helpful resources. You're skilled at using search tools to discover documentation, tutorials, and examples that directly address the user's coding needs.",
    tools=[SerperDevTool()],
    allow_delegation=False,
    verbose=True,
)

# LangGraph Agent Adapter
reporter_agent = LangGraphAgentAdapter(
    role="Reporter",
    goal="Report the results of the tasks.",
    backstory="You are a reporter who reports the results of the other tasks",
    llm=ChatOpenAI(model="gpt-4o"),
    allow_delegation=True,
    verbose=True,
)


class Code(BaseModel):
    code: str


task = Task(
    description="Give an answer to the coding question: {task}",
    expected_output="A thorough answer to the coding question: {task}",
    agent=code_helper_agent,
    output_json=Code,
)
task2 = Task(
    description="Find links to resources that can help with coding tasks. Use the serper tool to find resources that can help.",
    expected_output="A list of links to resources that can help with coding tasks",
    agent=link_finder_agent,
)


class Report(BaseModel):
    code: str
    links: List[str]


task3 = Task(
    description="Report the results of the tasks.",
    expected_output="A report of the results of the tasks. this is the code produced and then the links to the resources that can help with the coding task.",
    agent=reporter_agent,
    output_json=Report,
)
# Use in CrewAI
crew = Crew(
    agents=[code_helper_agent, link_finder_agent, reporter_agent],
    tasks=[task, task2, task3],
    verbose=True,
)

result = crew.kickoff(
    inputs={"task": "How do you implement an abstract class in python?"}
)

# Print raw result first
print("Raw result:", result)

# Handle result based on its type
if hasattr(result, "json_dict") and result.json_dict:
    json_result = result.json_dict
    print("\nStructured JSON result:")
    print(f"{json.dumps(json_result, indent=2)}")

    # Access fields safely
    if isinstance(json_result, dict):
        if "code" in json_result:
            print("\nCode:")
            print(
                json_result["code"][:200] + "..."
                if len(json_result["code"]) > 200
                else json_result["code"]
            )

        if "links" in json_result:
            print("\nLinks:")
            for link in json_result["links"][:5]:  # Print first 5 links
                print(f"- {link}")
            if len(json_result["links"]) > 5:
                print(f"...and {len(json_result['links']) - 5} more links")
elif hasattr(result, "pydantic") and result.pydantic:
    print("\nPydantic model result:")
    print(result.pydantic.model_dump_json(indent=2))
else:
    # Fallback to raw output
    print("\nNo structured result available, using raw output:")
    print(result.raw[:500] + "..." if len(result.raw) > 500 else result.raw)

```