Use Arcade tools with CrewAI
CrewAI is an agentic framework optimized for building task-oriented multi- systems. This guide explains how to integrate Arcade into your CrewAI applications.
Outcomes
You will build a CrewAI that uses Arcade to help with Gmail and Slack.
You will Learn
- How to retrieve Arcade and convert them to CrewAI format
- How to build a CrewAI with Arcade
- How to implement “just in time” (JIT) authorization using Arcade’s client
Prerequisites
The agent architecture you will build in this guide
CrewAI provides a Crew class that implements a multi- system. It provides an interface for you to define the agents, tasks, and memory. In this guide, you will manually keep track of the agent’s history and state, and use the kickoff method to invoke the agent in an agentic loop.
Integrate Arcade tools into a CrewAI agent
Create a new project
Create a new directory for your and initialize a new virtual environment:
mkdir crewai-arcade-example
cd crewai-arcade-example
uv init
uv venvBash/Zsh (macOS/Linux)
source .venv/bin/activateInstall the necessary packages:
uv add 'crewai[tools]' arcadepyCreate a new file called .env and add the following environment variables:
# Arcade API key
ARCADE_API_KEY=YOUR_ARCADE_API_KEY
# Arcade user ID (this is the email address you used to login to Arcade)
ARCADE_USER_ID={arcade_user_id}
# OpenAI API key
OPENAI_API_KEY=YOUR_OPENAI_API_KEYImport the necessary packages
Create a new file called main.py and add the following code:
from typing import Any
from arcadepy import Arcade
from arcadepy.types import ToolDefinition
from crewai.tools import BaseTool
from crewai import Agent
from crewai.events.event_listener import EventListener
from pydantic import BaseModel, Field, create_model
from dotenv import load_dotenv
import os
This includes many imports, here’s a breakdown:
- Arcade imports:
Arcade: The , used to interact with the .ToolDefinition: The definition type, used to define the input and output of a tool.
- CrewAI imports:
BaseTool: The base class, used to create custom CrewAI tools.Agent: The CrewAI class, used to create an agent.EventListener: The event listener class, used to suppress CrewAI’s rich panel output.
- Other imports:
pydanticimports: Used for data validation and model creation when converting Arcade to LangChain tools.typing.Any: A type hint for the any type.load_dotenv: Loads the environment variables from the.envfile.os: The operating system module, used to interact with the operating system.
Configure the agent
The rest of the code uses these variables to customize the and manage the . Feel free to configure them to your liking. Here, the EventListener class is used to suppress CrewAI’s rich panel output, which is useful for debugging but verbose for an interactive session like the one you’re building.
# Load environment variables from the .env file
load_dotenv()
# The Arcade User ID identifies who is authorizing each service.
ARCADE_USER_ID = os.getenv("ARCADE_USER_ID")
# This determines which MCP server is providing the tools, you can customize this to make a Notion agent. All tools from the MCP servers defined in the array will be used.
MCP_SERVERS = ["Slack"]
# This determines individual tools. Useful to pick specific tools when you don't need all of them.
TOOLS = ["Gmail_ListEmails", "Gmail_SendEmail", "Gmail_WhoAmI"]
# This determines the maximum number of tool definitions Arcade will return per MCP server
TOOL_LIMIT = 30
# This determines which LLM model will be used inside the agent
MODEL = "openai/gpt-5-mini"
# The individual objective that guides the agent's decision-making
AGENT_GOAL = "Help the user with all their requests"
# Provides context and personality to the agent, enriching interactions
AGENT_BACKSTORY = "You are a helpful assistant that can assist with Gmail and Slack."
# This defines the Agent's role. A short description of its function and expertise
AGENT_NAME = "Communication Manager"
# Suppress CrewAI's rich panel output
EventListener().formatter.verbose = FalseWrite a utility function to transform Arcade tool definitions into Pydantic models
In this utility function, you transform an Arcade definition into a Pydantic model. Later, you will transform these models to construct tools in the format expected by CrewAI. The _build_args_model function extracts the tools’ parameters, name, and description, and maps them to a Pydantic model.
TYPE_MAP: dict[str, type] = {
"string": str,
"number": float,
"integer": int,
"boolean": bool,
"array": list,
"json": dict,
}
def _python_type(val_type: str) -> type:
t = TYPE_MAP.get(val_type)
if t is None:
raise ValueError(f"Unsupported Arcade value type: {val_type}")
return t
def _build_args_model(tool_def: ToolDefinition) -> type[BaseModel]:
fields: dict[str, Any] = {}
for param in tool_def.input.parameters or []:
param_type = _python_type(param.value_schema.val_type)
if param_type is list and param.value_schema.inner_val_type:
inner = _python_type(param.value_schema.inner_val_type)
param_type = list[inner] # type: ignore[valid-type]
default = ... if param.required else None
fields[param.name] = (
param_type,
Field(default=default, description=param.description or ""),
)
return create_model(f"{tool_def.name}Input", **fields)Write a custom class that extends the CrewAI BaseTool class
Here, you define the ArcadeTool class that extends the CrewAI BaseTool class to add the following capability:
- Authorize the tool with the with the
_auth_toolhelper function - Execute the tool with the with the
_runmethod
This class captures the authorization flow outside of the agent’s ,
which is a good practice for security and context engineering. By handling
everything in the ArcadeTool class, you remove the risk of the LLM replacing
the authorization URL or leaking it, and you keep the context free from any
authorization-related traces, which reduces the risk of hallucinations, and
reduces context bloat.
class ArcadeTool(BaseTool):
"""A CrewAI tool backed by an Arcade tool definition."""
name: str
description: str
args_schema: type[BaseModel]
# Internal fields (not exposed to the agent)
arcade_tool_name: str = ""
user_id: str = ""
_client: Arcade | None = None
def _auth_tool(self):
auth = self._client.tools.authorize(
tool_name=self.arcade_tool_name,
user_id=self.user_id,
)
if auth.status != "completed":
print(f"Authorization required. Visit: {auth.url}")
self._client.auth.wait_for_completion(auth)
def _run(self, **kwargs: Any) -> str:
if self._client is None:
self._client = Arcade()
self._auth_tool()
print(f"Calling {self.arcade_tool_name}...")
result = self._client.tools.execute(
tool_name=self.arcade_tool_name,
input=kwargs,
user_id=self.user_id,
)
if not result.success:
return f"Tool error: {result.output.error.message}"
print(f"Call to {self.arcade_tool_name} successful, the agent will now process the result...")
return result.output.valueRetrieve Arcade tools and transform them into CrewAI tools
Here you get the Arcade tools you want the agent to utilize, and transform them into CrewAI tools. The first step is to initialize the , and get the you want to work with.
Here’s a breakdown of what it does for clarity:
- retrieve tools from all configured servers (defined in the
MCP_SERVERSvariable) - retrieve individual (defined in the
TOOLSvariable) - transform the Arcade to CrewAI tools with the
ArcadeToolclass you defined earlier
def get_arcade_tools(
client: Arcade,
*,
tools: list[str] | None = None,
mcp_servers: list[str] | None = None,
user_id: str = "",
) -> list[ArcadeTool]:
if not tools and not mcp_servers:
raise ValueError("Provide at least one tool name or toolkit name")
definitions: list[ToolDefinition] = []
if tools:
for name in tools:
definitions.append(client.tools.get(name=name))
if mcp_servers:
for tk in mcp_servers:
page = client.tools.list(toolkit=tk)
definitions.extend(page.items)
result: list[ArcadeTool] = []
for defn in definitions:
sanitized_name = defn.qualified_name.replace(".", "_")
t = ArcadeTool(
client=client,
name=sanitized_name,
description=defn.description,
args_schema=_build_args_model(defn),
arcade_tool_name=defn.qualified_name,
user_id=user_id,
)
result.append(t)
return resultCreate the main function
The main function is where you:
- Get the Arcade tools from the configured servers
- Create an with the Arcade
- Initialize the conversation
- Run the loop
def main():
client = Arcade()
arcade_tools = get_arcade_tools(
client,
tools=TOOLS,
mcp_servers=MCP_SERVERS,
user_id=ARCADE_USER_ID,
)
agent = Agent(
role=AGENT_NAME,
goal=AGENT_GOAL,
backstory=AGENT_BACKSTORY,
tools=arcade_tools,
)
history = []
print("Agent ready. Type 'exit' to quit.\n")
while True:
user_input = input("> ")
if user_input.strip().lower() in ("exit", "quit"):
break
history.append({"role": "user", "content": user_input})
result = agent.kickoff(history)
history.append({"role": "assistant", "content": result.raw})
print(f"\n{result.raw}\n")
if __name__ == "__main__":
main()Run the agent
uv run main.pyYou should see the responding to your prompts like any model, as well as handling any calls and authorization requests. Here are some example prompts you can try:
- “Send me an email with a random haiku about OpenAI ”
- “Summarize my latest 3 emails”
Tips for selecting tools
- Relevance: Pick only the you need. Avoid using all tools at once.
- Avoid conflicts: Be mindful of duplicate or overlapping functionality.
Next steps
Now that you have integrated Arcade tools into your CrewAI team, you can:
- Experiment with different toolkits, such as “Math” or “Search.”
- Customize the ’s prompts for specific tasks.
- Customize the authorization and execution flow to meet your application’s requirements.
Example code
main.py (full file)
from typing import Any
from arcadepy import Arcade
from arcadepy.types import ToolDefinition
from crewai.tools import BaseTool
from crewai import Agent
from crewai.events.event_listener import EventListener
from pydantic import BaseModel, Field, create_model
from dotenv import load_dotenv
import os
# Load environment variables from the .env file
load_dotenv()
# The Arcade User ID identifies who is authorizing each service.
ARCADE_USER_ID = os.getenv("ARCADE_USER_ID")
# This determines which MCP server is providing the tools, you can customize this to make a Notion agent. All tools from the MCP servers defined in the array will be used.
MCP_SERVERS = ["Slack"]
# This determines individual tools. Useful to pick specific tools when you don't need all of them.
TOOLS = ["Gmail_ListEmails", "Gmail_SendEmail", "Gmail_WhoAmI"]
# This determines the maximum number of tool definitions Arcade will return per MCP server
TOOL_LIMIT = 30
# This determines which LLM model will be used inside the agent
MODEL = "openai/gpt-5-mini"
# The individual objective that guides the agent's decision-making
AGENT_GOAL = "Help the user with all their requests"
# Provides context and personality to the agent, enriching interactions
AGENT_BACKSTORY = "You are a helpful assistant that can assist with Gmail and Slack."
# This defines the Agent's role. A short description of its function and expertise
AGENT_NAME = "Communication Manager"
# Suppress CrewAI's rich panel output
EventListener().formatter.verbose = False
TYPE_MAP: dict[str, type] = {
"string": str,
"number": float,
"integer": int,
"boolean": bool,
"array": list,
"json": dict,
}
def _python_type(val_type: str) -> type:
t = TYPE_MAP.get(val_type)
if t is None:
raise ValueError(f"Unsupported Arcade value type: {val_type}")
return t
def _build_args_model(tool_def: ToolDefinition) -> type[BaseModel]:
fields: dict[str, Any] = {}
for param in tool_def.input.parameters or []:
param_type = _python_type(param.value_schema.val_type)
if param_type is list and param.value_schema.inner_val_type:
inner = _python_type(param.value_schema.inner_val_type)
param_type = list[inner] # type: ignore[valid-type]
default = ... if param.required else None
fields[param.name] = (
param_type,
Field(default=default, description=param.description or ""),
)
return create_model(f"{tool_def.name}Input", **fields)
class ArcadeTool(BaseTool):
"""A CrewAI tool backed by an Arcade tool definition."""
name: str
description: str
args_schema: type[BaseModel]
# Internal fields (not exposed to the agent)
arcade_tool_name: str = ""
user_id: str = ""
_client: Arcade | None = None
def _auth_tool(self):
auth = self._client.tools.authorize(
tool_name=self.arcade_tool_name,
user_id=self.user_id,
)
if auth.status != "completed":
print(f"Authorization required. Visit: {auth.url}")
self._client.auth.wait_for_completion(auth)
def _run(self, **kwargs: Any) -> str:
if self._client is None:
self._client = Arcade()
self._auth_tool()
print(f"Calling {self.arcade_tool_name}...")
result = self._client.tools.execute(
tool_name=self.arcade_tool_name,
input=kwargs,
user_id=self.user_id,
)
if not result.success:
return f"Tool error: {result.output.error.message}"
print(f"Call to {self.arcade_tool_name} successful, the agent will now process the result...")
return result.output.value
def get_arcade_tools(
client: Arcade,
*,
tools: list[str] | None = None,
mcp_servers: list[str] | None = None,
user_id: str = "",
) -> list[ArcadeTool]:
if not tools and not mcp_servers:
raise ValueError("Provide at least one tool name or toolkit name")
definitions: list[ToolDefinition] = []
if tools:
for name in tools:
definitions.append(client.tools.get(name=name))
if mcp_servers:
for tk in mcp_servers:
page = client.tools.list(toolkit=tk)
definitions.extend(page.items)
result: list[ArcadeTool] = []
for defn in definitions:
sanitized_name = defn.qualified_name.replace(".", "_")
t = ArcadeTool(
client=client,
name=sanitized_name,
description=defn.description,
args_schema=_build_args_model(defn),
arcade_tool_name=defn.qualified_name,
user_id=user_id,
)
result.append(t)
return result
def main():
client = Arcade()
arcade_tools = get_arcade_tools(
client,
tools=TOOLS,
mcp_servers=MCP_SERVERS,
user_id=ARCADE_USER_ID,
)
agent = Agent(
role=AGENT_NAME,
goal=AGENT_GOAL,
backstory=AGENT_BACKSTORY,
tools=arcade_tools,
)
history = []
print("Agent ready. Type 'exit' to quit.\n")
while True:
user_input = input("> ")
if user_input.strip().lower() in ("exit", "quit"):
break
history.append({"role": "user", "content": user_input})
result = agent.kickoff(history)
history.append({"role": "assistant", "content": result.raw})
print(f"\n{result.raw}\n")
if __name__ == "__main__":
main()