CrewAI agents become genuinely useful when they can reach outside their context window. A REST API tool lets agents query live data, trigger workflows, or mutate state — turning a chat loop into an automation engine. This guide walks through building a production-ready CrewAI custom tool that wraps any REST endpoint, handles auth and retries, and returns structured data your agents can reason over.
Step 1: Define the tool interface and data contracts
Start by deciding what the tool exposes to the agent. CrewAI tools inherit from BaseTool and must implement _run (sync) or _arun (async). The agent sees the class docstring and the _run signature, so type hints and clear descriptions directly affect tool selection accuracy.
Create a Pydantic model for the tool input. This validates arguments before the HTTP call and gives the LLM a schema to follow.
# tools/rest_tool.py
from pydantic import BaseModel, Field, HttpUrl
from typing import Optional, Literal, Dict, Any
from enum import Enum
class HttpMethod(str, Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
PATCH = "PATCH"
DELETE = "DELETE"
class RestApiInput(BaseModel):
"""Arguments the agent must provide to call the REST API."""
url: HttpUrl = Field(..., description="Full endpoint URL including query params")
method: HttpMethod = Field(default=HttpMethod.GET, description="HTTP verb")
headers: Optional[Dict[str, str]] = Field(
default=None, description="Additional headers (auth added automatically)"
)
json_body: Optional[Dict[str, Any]] = Field(
default=None, description="JSON payload for POST/PUT/PATCH"
)
params: Optional[Dict[str, Any]] = Field(
default=None, description="Query string parameters"
)
timeout_seconds: int = Field(default=30, ge=1, le=120)The output model matters just as much. Agents struggle with raw JSON strings. Return a typed object with the fields the agent actually needs.
class RestApiOutput(BaseModel):
"""Structured result the agent can reference in subsequent steps."""
status_code: int
success: bool
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
response_headers: Dict[str, str] = Field(default_factory=dict)
elapsed_ms: intStep 2: Implement the tool with auth, retries, and observability
The tool should handle the concerns every HTTP client needs: authentication, timeout enforcement, retry logic for transient failures, and structured logging. Keep credentials out of the agent’s context — inject them at tool initialization.
# tools/rest_tool.py (continued)
import time
import logging
from typing import Type
from crewai.tools import BaseTool
import httpx
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
)
logger = logging.getLogger(__name__)
class RestApiTool(BaseTool):
name: str = "rest_api_client"
"Call a REST API endpoint. Provide the full URL, HTTP method, optional headers, "
"query params, and JSON body. Returns structured response with status, data, and timing. "
"Authentication headers are injected automatically from tool configuration."
)
args_schema: Type[BaseModel] = RestApiInput
def __init__(
self,
base_headers: Optional[Dict[str, str]] = None,
default_timeout: int = 30,
max_retries: int = 3,
**kwargs,
):
super().__init__(**kwargs)
self._base_headers = base_headers or {}
self._default_timeout = default_timeout
self._max_retries = max_retries
self._client: Optional[httpx.Client] = None
@property
def client(self) -> httpx.Client:
if self._client is None:
self._client = httpx.Client(
timeout=httpx.Timeout(self._default_timeout),
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
)
return self._client
def _merge_headers(self, extra: Optional[Dict[str, str]]) -> Dict[str, str]:
merged = {**self._base_headers}
if extra:
merged.update(extra)
return merged
@retry(
wait=wait_exponential_jitter(initial=0.5, max=4),
stop=stop_after_attempt(3),
retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)),
reraise=True,
)
def _request_with_retry(
self,
method: str,
url: str,
headers: Dict[str, str],
json_body: Optional[Dict],
params: Optional[Dict],
timeout: int,
) -> httpx.Response:
return self.client.request(
method=method,
url=url,
headers=headers,
json=json_body,
params=params,
timeout=timeout,
)
def _run(
self,
url: str,
method: HttpMethod = HttpMethod.GET,
headers: Optional[Dict[str, str]] = None,
json_body: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
timeout_seconds: int = 30,
) -> RestApiOutput:
start = time.perf_counter()
request_headers = self._merge_headers(headers)
logger.info("REST tool request", extra={"method": method.value, "url": url})
try:
response = self._request_with_retry(
method=method.value,
url=url,
headers=request_headers,
json_body=json_body,
params=params,
timeout=timeout_seconds,
)
elapsed_ms = int((time.perf_counter() - start) * 1000)
response_headers = dict(response.headers)
try:
data = response.json() if response.content else None
except Exception:
data = {"raw_text": response.text[:2000]}
output = RestApiOutput(
status_code=response.status_code,
success=200 <= response.status_code < 300,
data=data,
error=None if response.is_success else f"HTTP {response.status_code}: {response.text[:500]}",
response_headers=response_headers,
elapsed_ms=elapsed_ms,
)
logger.info(
"REST tool response",
extra={"status": response.status_code, "elapsed_ms": elapsed_ms, "success": output.success},
)
return output
except httpx.TimeoutException as e:
elapsed_ms = int((time.perf_counter() - start) * 1000)
logger.warning("REST tool timeout", extra={"url": url, "elapsed_ms": elapsed_ms})
return RestApiOutput(
status_code=408,
success=False,
error=f"Request timed out after {timeout_seconds}s",
elapsed_ms=elapsed_ms,
)
except httpx.HTTPStatusError as e:
elapsed_ms = int((time.perf_counter() - start) * 1000)
return RestApiOutput(
status_code=e.response.status_code,
success=False,
error=f"HTTP error: {e.response.text[:500]}",
elapsed_ms=elapsed_ms,
)
except Exception as e:
elapsed_ms = int((time.perf_counter() - start) * 1000)
logger.exception("REST tool unexpected error")
return RestApiOutput(
status_code=500,
success=False,
error=f"Unexpected error: {type(e).__name__}: {str(e)[:200]}",
elapsed_ms=elapsed_ms,
)Key design choices here:
- Tenacity retries only on network-level failures, not 4xx/5xx responses. The agent should decide how to handle application errors.
- Structured logging with
extrafields makes it trivial to query in Datadog, Loki, or CloudWatch. - Header merging lets you inject
Authorization,X-Request-ID, orUser-Agentat tool construction while allowing per-call overrides. - httpx.Client reuse avoids connection overhead and respects keep-alive limits.
Step 3: Wire the tool into a CrewAI agent
Instantiate the tool with your auth configuration, then pass it to the agent’s tools list. The agent will see the docstring and schema automatically.
# agents/research_agent.py
from crewai import Agent, Task, Crew, Process
from tools.rest_tool import RestApiTool
import os
def build_github_research_agent() -> Agent:
github_token = os.getenv("GITHUB_TOKEN")
if not github_token:
raise RuntimeError("GITHUB_TOKEN not set")
github_tool = RestApiTool(
base_headers={
"Authorization": f"Bearer {github_token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
"User-Agent": "crewai-research-agent/1.0",
},
default_timeout=20,
max_retries=3,
)
return Agent(
role="GitHub Repository Analyst",
goal="Answer questions about GitHub repositories using the REST API",
backstory=(
"You fetch live repository data, issues, PRs, and commit history "
"to answer technical questions. You always cite the API endpoints used."
),
tools=[github_tool],
verbose=True,
allow_delegation=False,
)Step 4: Design tasks that guide the agent’s tool usage
Agents need explicit guidance on when and how to use the tool. The task description should specify the endpoint patterns, expected response shapes, and how to chain calls.
# tasks/github_tasks.py
from crewai import Task
from agents.research_agent import build_github_research_agent
def create_repo_analysis_task(repo_owner: str, repo_name: str, question: str) -> Task:
agent = build_github_research_agent()
return Task(
description=(
f"Analyze the GitHub repository {repo_owner}/{repo_name} to answer: {question}\n\n"
"Available API endpoints (use the rest_api_client tool):\n"
f"- GET https://api.github.com/repos/{repo_owner}/{repo_name} — repo metadata\n"
f"- GET https://api.github.com/repos/{repo_owner}/{repo_name}/issues — open issues\n"
f"- GET https://api.github.com/repos/{repo_owner}/{repo_name}/pulls — pull requests\n"
f"- GET https://api.github.com/repos/{repo_owner}/{repo_name}/commits — recent commits\n"
f"- GET https://api.github.com/repos/{repo_owner}/{repo_name}/contents/{{path}} — file contents\n\n"
"Process:\n"
"1. Start with the repo metadata endpoint to verify access and get default branch.\n"
"2. Fetch relevant issues, PRs, or commits based on the question.\n"
"3. If the question references specific files, use the contents endpoint.\n"
"4. Synthesize findings into a concise answer with endpoint citations.\n"
"5. If rate limited (403 with 'rate limit' in response), wait and retry once."
),
expected_output=(
"A markdown report with:\n"
"- Direct answer to the question\n"
"- Key findings with API endpoint references\n"
"- Any limitations or missing data\n"
"- Raw data summary (counts, dates, statuses)"
),
agent=agent,
)Step 5: Run the crew and verify the output
Execute the crew and inspect the structured tool outputs. The RestApiOutput model gives you programmatic access to status codes, parsed JSON, and timing — useful for downstream logic or evals.
# main.py
import json
import os
from dotenv import load_dotenv
from tasks.github_tasks import create_repo_analysis_task
def main():
load_dotenv()
task = create_repo_analysis_task(
repo_owner="crewAIInc",
repo_name="crewAI",
question="What are the top 3 most recent merged PRs and what files did they modify?",
)
crew = Crew(
agents=[task.agent],
tasks=[task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print("\n=== FINAL RESULT ===")
print(result)
# For programmatic access to tool outputs, inspect task.output
# Each tool call returns a RestApiOutput instance
if hasattr(task, "output") and task.output:
print("\n=== TOOL CALL TRACE ===")
# CrewAI stores tool results in task.output.raw or similar
# depending on version; check your installed version's attributes
if __name__ == "__main__":
main()Verification checklist:
- Run
python main.py— you should see the agent invokerest_api_clientmultiple times in the verbose logs. - Check that each tool call logs
REST tool requestandREST tool responsewith status codes and latency. - The final answer should cite specific endpoints (e.g., “per GET /repos/crewAIInc/crewAI/pulls”).
- If you set
GITHUB_TOKENto an invalid value, the tool returnssuccess=falsewith a 401 error — the agent should surface this rather than hallucinate.
Step 6: Add async support for parallel tool calls
CrewAI’s sequential process runs one task at a time, but a single task may benefit from parallel API calls (e.g., fetching issues and PRs simultaneously). Implement _arun and use asyncio.gather in a wrapper.
# tools/rest_tool.py (add to RestApiTool class)
import asyncio
from typing import List
class AsyncRestApiTool(RestApiTool):
"""Async variant for parallel execution within a single task."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._async_client: Optional[httpx.AsyncClient] = None
@property
def async_client(self) -> httpx.AsyncClient:
if self._async_client is None:
self._async_client = httpx.AsyncClient(
timeout=httpx.Timeout(self._default_timeout),
limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
)
return self._async_client
async def _arun(
self,
url: str,
method: HttpMethod = HttpMethod.GET,
headers: Optional[Dict[str, str]] = None,
json_body: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
timeout_seconds: int = 30,
) -> RestApiOutput:
# Reuse the sync logic but with async client
start = time.perf_counter()
request_headers = self._merge_headers(headers)
try:
response = await self.async_client.request(
method=method.value,
url=url,
headers=request_headers,
json=json_body,
params=params,
timeout=timeout_seconds,
)
elapsed_ms = int((time.perf_counter() - start) * 1000)
data = response.json() if response.content else None
return RestApiOutput(
status_code=response.status_code,
success=response.is_success,
data=data,
error=None if response.is_success else f"HTTP {response.status_code}",
response_headers=dict(response.headers),
elapsed_ms=elapsed_ms,
)
except Exception as e:
elapsed_ms = int((time.perf_counter() - start) * 1000)
return RestApiOutput(
status_code=500,
success=False,
error=str(e)[:200],
elapsed_ms=elapsed_ms,
)
async def batch_get(self, urls: List[str], headers: Optional[Dict] = None) -> List[RestApiOutput]:
"""Fetch multiple URLs in parallel."""
tasks = [
self._arun(url=url, method=HttpMethod.GET, headers=headers)
for url in urls
]
return await asyncio.gather(*tasks)Update the task to use the async tool when parallel fetches make sense:
# tasks/github_tasks.py (updated)
def create_parallel_repo_task(repo_owner: str, repo_name: str) -> Task:
agent = build_github_research_agent()
# Replace the tool with async variant
agent.tools = [
AsyncRestApiTool(
base_headers=agent.tools[0]._base_headers,
default_timeout=agent.tools[0]._default_timeout,
)
]
return Task(
description=(
f"Fetch repository metadata, open issues, and open PRs for {repo_owner}/{repo_name} "
"in parallel using the async tool's batch_get method. Then summarize."
),
expected_output="Summary with counts and key details from all three endpoints",
agent=agent,
)Step 7: Handle pagination and large responses
Most REST APIs paginate. Build a small helper into the tool so agents don’t need to reason about Link headers or cursor parameters.
# tools/rest_tool.py (add method to RestApiTool)
def paginated_get(
self,
url: str,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
max_pages: int = 5,
page_param: str = "page",
per_page_param: str = "per_page",
per_page: int = 100,
) -> List[Dict[str, Any]]:
"""Fetch all pages up to max_pages, returning combined results."""
all_items: List[Dict[str, Any]] = []
current_params = {**params, page_param: 1, per_page_param: per_page} if params else {page_param: 1, per_page_param: per_page}
for page_num in range(1, max_pages + 1):
current_params[page_param] = page_num
result = self._run(
url=url,
method=HttpMethod.GET,
headers=headers,
params=current_params,
)
if not result.success or not result.data:
break
# Handle GitHub-style array responses and paginated object responses
items = result.data if isinstance(result.data, list) else result.data.get("items", [])
if not items:
break
all_items.extend(items)
# Check for next page via Link header (GitHub) or item count < per_page
link_header = result.response_headers.get("link", "")
if f'rel="next"' not in link_header and len(items) < per_page:
break
return all_itemsNow the agent can call a single tool method and get consolidated results:
# In task description:
"- Use the tool's paginated_get method to fetch ALL open issues (up to 5 pages, 100 per page) "
"from https://api.github.com/repos/{owner}/{repo}/issues"Step 8: Secure credential handling in production
Never bake tokens into tool instances that get serialized or logged. Use a credential provider pattern that pulls secrets at runtime.
# tools/credential_provider.py
from abc import ABC, abstractmethod
from typing import Optional
import os
class CredentialProvider(ABC):
@abstractmethod
def get_token(self, service: str) -> Optional[str]:
pass
class EnvCredentialProvider(CredentialProvider):
def get_token(self, service: str) -> Optional[str]:
return os.getenv(f"{service.upper()}_TOKEN")
class VaultCredentialProvider(CredentialProvider):
"""Example: HashiCorp Vault or AWS Secrets Manager integration."""
def __init__(self, vault_client):
self._vault = vault_client
def get_token(self, service: str) -> Optional[str]:
secret = self._vault.read_secret(f"crewai/{service}")
return secret.get("token") if secret else None
def build_authenticated_tool(
service: str,
provider: CredentialProvider,
base_url: str,
extra_headers: Optional[Dict[str, str]] = None,
) -> RestApiTool:
token = provider.get_token(service)
if not token:
raise ValueError(f"No credential found for service: {service}")
headers = {"Authorization": f"Bearer {token}"}
if extra_headers:
headers.update(extra_headers)
return RestApiTool(base_headers=headers)Usage:
# agents/production_agent.py
from tools.credential_provider import EnvCredentialProvider, build_authenticated_tool
provider = EnvCredentialProvider()
github_tool = build_authenticated_tool(
service="github",
provider=provider,
base_url="https://api.github.com",
extra_headers={"Accept": "application/vnd.github+json"},
)This keeps credentials out of agent memory, tool serialization, and log output.
Step 9: Add request/response middleware for cross-cutting concerns
Real systems need request ID propagation, distributed tracing headers, and rate-limit awareness. Add a middleware layer to the httpx client.
# tools/middleware.py
import uuid
import httpx
from typing import Callable, Awaitable
class RequestMiddleware:
def __init__(self, service_name: str = "crewai-agent"):
self.service_name = service_name
def sync_middleware(self, request: httpx.Request) -> httpx.Request:
request.headers.setdefault("X-Request-ID", str(uuid.uuid4()))
request.headers.setdefault("X-Service-Name", self.service_name)
return request
async def async_middleware(self, request: httpx.Request) -> httpx.Request:
return self.sync_middleware(request)
def create_instrumented_client(
middleware: RequestMiddleware,
**client_kwargs,
) -> httpx.Client:
transport = httpx.HTTPTransport()
client = httpx.Client(transport=transport, **client_kwargs)
# Wrap send to inject headers
original_send = client.send
def instrumented_send(request: httpx.Request, **kwargs):
request = middleware.sync_middleware(request)
return original_send(request, **kwargs)
client.send = instrumented_send
return clientWire it into the tool:
# tools/rest_tool.py (in __init__)
from tools.middleware import RequestMiddleware, create_instrumented_client
self._middleware = RequestMiddleware(service_name="github-analyst")
self._client = create_instrumented_client(
middleware=self._middleware,
timeout=httpx.Timeout(self._default_timeout),
limits=httpx.Limits(max_connections=10),
)Now every outbound request carries a traceable X-Request-ID, making debugging across services straightforward.
Step 10: Test the tool in isolation before handing to agents
Unit test the tool directly with a mock server. This catches schema mismatches, auth header issues, and error handling without burning API quota or involving LLM non-determinism.
# tests/test_rest_tool.py
import pytest
import httpx
from unittest.mock import AsyncMock, patch
from tools.rest_tool import RestApiTool, RestApiInput, RestApiOutput
@pytest.fixture
def mock_github_response():
return {
"id": 123456,
"name": "test-repo",
"full_name": "owner/test-repo",
"private": False,
"stargazers_count": 42,
}
def test_rest_tool_success(mock_github_response):
tool = RestApiTool(
base_headers={"Authorization": "Bearer test-token"},
default_timeout=10,
)
with patch.object(tool.client, "request") as mock_request:
mock_response = httpx.Response(
200,
json=mock_github_response,
headers={"Content-Type": "application/json"},
)
mock_request.return_value = mock_response
result = tool._run(
url="https://api.github.com/repos/owner/test-repo",
method="GET",
)
assert isinstance(result, RestApiOutput)
assert result.success is True
assert result.status_code == 200
assert result.data == mock_github_response
assert result.elapsed_ms > 0
# Verify auth header was injected
call_args = mock_request.call_args
assert call_args.kwargs["headers"]["Authorization"] == "Bearer test-token"
def test_rest_tool_rate_limit_handling():
tool = RestApiTool(base_headers={}, default_timeout=10)
with patch.object(tool.client, "request") as mock_request:
mock_response = httpx.Response(
403,
json={"message": "API rate limit exceeded"},
headers={"X-RateLimit-Remaining": "0"},
)
mock_request.return_value = mock_response
result = tool._run(url="https://api.github.com/repos/owner/repo")
assert result.success is False
assert result.status_code == 403
assert "rate limit" in result.error.lower()
def test_rest_tool_timeout_retry():
tool = RestApiTool(base_headers={}, default_timeout=1, max_retries=2)
with patch.object(tool.client, "request") as mock_request:
mock_request.side_effect = httpx.TimeoutException("Connection timed out")
result = tool._run(url="https://api.github.com/repos/owner/repo")
assert result.success is False
assert result.status_code == 408
assert mock_request.call_count == 3 # initial + 2 retriesRun with pytest tests/test_rest_tool.py -v. All tests should pass before you wire the tool into any agent.
You now have a REST API tool that handles auth, retries, pagination, async parallelism, observability, and secure credential management — ready for any CrewAI agent that needs live data. The same pattern applies to GraphQL endpoints, webhook callbacks, or internal service meshes. Swap the HTTP client for aiohttp or requests if your stack demands it; the interface contract stays the same.