Build a Multi-Agent Chat API with CrewAI + FastAPI (in Python)
- amareshpat
- 1 day ago
- 9 min read
If you’ve been eyeing multi-agent frameworks and wondering how to put them to work behind a clean, production-friendly HTTP interface, this guide is for you. We’ll build a chat API using CrewAI (for multi-agent orchestration) and FastAPI (for the web layer). The API will route complex user requests to multiple specialized agents and tools—e.g., checking weather forecasts before making travel suggestions.
We’ll also demo a real prompt the API can answer:
“I want to visit the Golden Gate Bridge in San Francisco sometime this week but I want to go on a sunny day. Can you suggest a good day to do this?”
What is CrewAI?
CrewAI is a lightweight Python framework for designing agents, composing them into “crews,” and executing tasks with sequential or hierarchical processes. Agents have roles/goals, can use tools, share context/memory, and collaborate toward outcomes. It’s designed for real-world automation with guardrails, knowledge, and observability baked in. (CrewAI Documentation)
Key concepts:
Agents: autonomous workers with a role/goal who can use tools and collaborate. (CrewAI Documentation)
Tools: functions (often decorated) that extend agent capabilities (web search, APIs, file ops, etc.). (CrewAI Documentation)
Tasks: assignments for agents; can be single-agent or collaborative. (CrewAI Documentation)
Processes: Sequential (linear) or Hierarchical (a manager coordinates/validates work). (CrewAI Documentation)
LLMs: CrewAI uses LiteLLM under the hood and by default reads OPENAI_MODEL_NAME (defaults to gpt-4o-mini) if unset. Configure your provider via env vars. (CrewAI Documentation)
Why CrewAI for a chat API?
It’s easy to compose specialized agents (planner, researcher, tool-user) and route complex queries across them.
Tools are first-class, so calling external APIs (like weather) is clean and testable.
Processes let you start simple (sequential) and later graduate to a manager-style flow (hierarchical) when complexity grows. (CrewAI Documentation)
What we’ll build
A FastAPI app exposing POST /chat. Internally, a Crew with multiple agents will:
Interpret the user’s query.
Use tools to geocode a place and fetch daily weather.
Select a good day to visit (sunny, low precipitation probability).
Compose a helpful, conversational answer.
For weather, we’ll use Open-Meteo, a free, no-key API with both geocoding and forecast endpoints. We’ll call:
Geocoding: https://geocoding-api.open-meteo.com/v1/search?name=... to convert “San Francisco” → lat/lon. (Open Meteo)
Forecast: request daily variables (e.g., weather_code, precipitation_probability_max, temperature_2m_max, cloud_cover_mean) for the next 7 days and the user’s timezone. (Open Meteo)Open-Meteo is free for non-commercial use and needs no API key. (Open Meteo)
Heads-up on weather codes: Open-Meteo follows WMO weather interpretation codes (e.g., 0 = clear sky). We’ll treat 0–1 as “sunny / mainly clear,” and further filter by low precipitation probability. (Open Meteo)
Project layout & setup
mkdir crewai_chat_api && cd crewai_chat_api
python -m venv .venv && source .venv/bin/activate
pip install "fastapi>=0.115" "uvicorn[standard]>=0.30" "requests>=2.32" "pydantic>=2.7" "python-dotenv>=1.0" "crewai>=0.51"
Create a .env (for your LLM provider). With OpenAI as an example:
OPENAI_API_KEY=sk-...
# Optional: override default
# OPENAI_MODEL_NAME=gpt-4o-mini
The code
Save the following as app.py.
import os
import json
import requests
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dotenv import load_dotenv
# --- CrewAI imports ---
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool # Decorator for defining tools
# Load environment for LLM provider (LiteLLM via CrewAI)
load_dotenv()
# ------------------------------
# FastAPI app & request schema
# ------------------------------
app = FastAPI(title="CrewAI Chat API", version="1.0")
class ChatRequest(BaseModel):
message: str
# Optional hints to improve results (e.g., user city/timezone)
location_hint: str | None = None
timezone: str | None = "America/Los_Angeles"
# ------------------------------
# Weather & Geocoding Tools
# ------------------------------
OPEN_METEO_GEOCODE = "https://geocoding-api.open-meteo.com/v1/search"
OPEN_METEO_FORECAST = "https://api.open-meteo.com/v1/forecast"
def _normalize_city(name: str) -> str:
return name.strip()
@tool("geocode_city")
def geocode_city(name: str) -> dict:
"""Geocode a city/landmark name into latitude/longitude using Open-Meteo.
Input: City or place name (e.g., "San Francisco" or "Golden Gate Bridge").
Output: JSON with best match: {name, country, latitude, longitude}."""
try:
q = _normalize_city(name)
resp = requests.get(OPEN_METEO_GEOCODE, params={"name": q, "count": 1})
resp.raise_for_status()
data = resp.json()
if not data.get("results"):
return {"ok": False, "reason": f"No geocoding results for '{name}'"}
r = data["results"][0]
return {
"ok": True,
"name": r.get("name") or name,
"country": r.get("country"),
"latitude": r["latitude"],
"longitude": r["longitude"],
}
except Exception as e:
return {"ok": False, "reason": f"geocode error: {e}"}
def _date_range_this_week(tz: str):
# "This week" as the next 7 days from now in the provided timezone
# CrewAI runs server-side; we keep it simple and use UTC, letting Open-Meteo
# localize via timezone param.
today = datetime.now(timezone.utc).date()
end = today + timedelta(days=7)
return today.isoformat(), end.isoformat()
@tool("get_daily_forecast")
def get_daily_forecast(lat_lon_tz: str) -> dict:
"""Get a 7-day daily forecast for 'lat,lon|timezone', including weather_code,
precipitation_probability_max, temperature_2m_max, and cloud_cover_mean.
Returns a list of days with ISO date and fields."""
try:
lat_s, lon_s, tz = lat_lon_tz.split("|")
lat = float(lat_s)
lon = float(lon_s)
start, end = _date_range_this_week(tz)
params = {
"latitude": lat,
"longitude": lon,
"timezone": tz,
"start_date": start,
"end_date": end,
"daily": [
"weather_code",
"precipitation_probability_max",
"temperature_2m_max",
"cloud_cover_mean",
],
}
# requests can pass list values; Open-Meteo accepts repeated params
resp = requests.get(OPEN_METEO_FORECAST, params=params, timeout=20)
resp.raise_for_status()
f = resp.json().get("daily", {})
days = []
for i, d in enumerate(f.get("time", [])):
days.append({
"date": d,
"weather_code": f.get("weather_code", [None]*len(f.get("time", [])))[i],
"precipitation_probability_max": f.get("precipitation_probability_max", [None]*len(f.get("time", [])))[i],
"temperature_2m_max": f.get("temperature_2m_max", [None]*len(f.get("time", [])))[i],
"cloud_cover_mean": f.get("cloud_cover_mean", [None]*len(f.get("time", [])))[i],
})
return {"ok": True, "days": days}
except Exception as e:
return {"ok": False, "reason": f"forecast error: {e}"}
@tool("pick_sunny_day")
def pick_sunny_day(forecast_json: str) -> dict:
"""Select the best 'sunny' day from a forecast JSON string.
Strategy:
- Prefer weather_code 0 or 1 (clear or mainly clear).
- Low precipitation_probability_max (e.g., < 20%).
- Lower cloud_cover_mean is better.
- Warmer temperature_2m_max is a tie-breaker.
Returns top candidates with reasoning."""
try:
data = json.loads(forecast_json)
days = data.get("days", [])
candidates = []
for d in days:
code = d.get("weather_code")
p = d.get("precipitation_probability_max")
cloud = d.get("cloud_cover_mean")
tmax = d.get("temperature_2m_max")
if code in (0, 1) and (p is None or p <= 20):
score = 100
if p is not None:
score -= p # lower precipitation prob -> higher score
if cloud is not None:
score -= (cloud / 2) # penalize cloudiness moderately
if tmax is not None:
score += (tmax / 2) # slightly prefer warmer
candidates.append((score, d))
# If no “sunny” candidates, fall back to least-bad by precip + cloud
if not candidates:
for d in days:
p = d.get("precipitation_probability_max") or 50
cloud = d.get("cloud_cover_mean") or 50
tmax = d.get("temperature_2m_max") or 0
score = 100 - p - (cloud / 2) + (tmax / 2)
candidates.append((score, d))
candidates.sort(key=lambda x: x[0], reverse=True)
best = [c[1] for c in candidates[:2]]
return {"ok": True, "candidates": best}
except Exception as e:
return {"ok": False, "reason": f"pick error: {e}"}
# ------------------------------
# Agents
# ------------------------------
# You can set model via environment (OPENAI_MODEL_NAME) per CrewAI docs.
# Optionally, pass `llm={"model": "gpt-4o-mini"}` into Agent(...) to override.
weather_analyst = Agent(
role="Meteorologist",
goal=(
"Given a user request with a place and a rough date window, "
"determine the best sunny day to visit in the next 7 days using tools."
),
backstory=(
"An expert forecaster skilled at interpreting Open-Meteo daily forecasts "
"and translating them into user-friendly recommendations."
),
tools=[geocode_city, get_daily_forecast, pick_sunny_day],
allow_delegation=False,
)
trip_planner = Agent(
role="Itinerary Planner",
goal=(
"Compose a friendly, concise recommendation with the best day and a "
"brief why, including a second-choice fallback and practical tips."
),
backstory=(
"A local-savvy planner who crafts helpful, on-point suggestions based "
"on structured inputs from the meteorologist."
),
tools=[],
allow_delegation=False,
)
# Optional: manager agent for hierarchical process
coordinator = Agent(
role="Coordinator",
goal="Ensure accurate tool use, validate assumptions, and finalize results.",
backstory="A detail-oriented manager who routes work to the right agent.",
tools=[],
allow_delegation=True,
)
# ------------------------------
# Tasks
# ------------------------------
weather_task = Task(
description=(
"1) Identify the location (city or landmark) from the user's message. "
"2) Geocode to lat/lon (geocode_city). "
"3) Fetch daily forecast for the next week (get_daily_forecast) in the user's timezone. "
"4) Pick the best sunny day with reasoning (pick_sunny_day). "
"Return JSON with 'candidates' containing 1-2 day objects."
),
expected_output=(
"A compact JSON string with a 'candidates' array. Example:\n"
'{"candidates": [{"date": "2025-10-26", "weather_code": 0, '
'"precipitation_probability_max": 5, "temperature_2m_max": 20, "cloud_cover_mean": 10}]}'
),
agent=weather_analyst,
)
planner_task = Task(
description=(
"Use the weather candidates and the user's message to produce a final answer. "
"State the best day (and 2nd-best if available), why it’s recommended (sunny/low rain), "
"and provide a short, practical tip. Keep it under 8 sentences."
),
expected_output="A conversational paragraph tailored to the user's request.",
agent=trip_planner,
)
# ------------------------------
# Crew (Hierarchical for complex routing; swap to Sequential if preferred)
# ------------------------------
crew = Crew(
agents=[weather_analyst, trip_planner, coordinator],
tasks=[weather_task, planner_task],
process=Process.HIERARCHICAL, # or Process.SEQUENTIAL
manager_agent=coordinator, # needed for hierarchical process
)
# ------------------------------
# API route
# ------------------------------
@app.post("/chat")
def chat(req: ChatRequest):
try:
inputs = {
"user_message": req.message,
"timezone": req.timezone or "UTC",
# heuristic: a hint helps geocoding when user only names a landmark
"location_hint": req.location_hint or "",
}
# The meteorologist will extract the location text; we also pass a hint.
# The tool chain expects a single string for get_daily_forecast:
# "lat|lon|timezone". That's constructed inside the agent via tools.
result = crew.kickoff(inputs=inputs)
# `result` can be a structured object or string depending on Task outputs.
# We'll stringify safely.
if hasattr(result, "raw"):
answer = result.raw
else:
answer = str(result)
return {"answer": answer}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
How it works
ToolsWe declared three tools with CrewAI’s @tool decorator:
geocode_city(name) → lat/lon (Open-Meteo Geocoding). (Open Meteo)
get_daily_forecast("lat|lon|timezone") → next-7-days daily forecast with weather code and precipitation probability max. (Open Meteo)
pick_sunny_day(forecast_json) → selects top 1–2 days favoring clear (code 0) or mainly clear (code 1) and low precipitation probability. (Weather code semantics follow WMO codes). (Open Meteo)
Agents & Tasks
The Meteorologist agent runs the geocoding → forecast → sunny-day selection pipeline and returns a compact JSON.
The Itinerary Planner crafts the final user-facing answer.
The Coordinator (manager) ensures proper routing/validation using a Hierarchical process—handy as your crews and tools grow. (CrewAI Documentation)
LLM configurationCrewAI connects to your provider via LiteLLM. If you don’t set OPENAI_MODEL_NAME, it defaults to gpt-4o-mini. You can set other providers/models via env vars (see CrewAI LLM docs). (CrewAI Documentation)
Run it
uvicorn app:app --reload --port 8000
Test with curl:
curl -s -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{
"message": "I want to visit the Golden Gate bridge in San Francisco sometime this week but I want to go on a sunny day. Can you suggest a good day to do this?",
"timezone": "America/Los_Angeles"
}' | jq .
You should get a friendly paragraph recommending a specific day (plus a fallback), rationale (sunny/low rain), and a small tip (e.g., “go early to beat crowds”).
Making the weather picker smarter
Open-Meteo daily fields are rich; you can request sunshine duration, UV index, or cloud cover stats and further refine selection. Some endpoints expose precipitation probability max and mean cloud cover at the daily level; include those to rank “sunny.” (Open Meteo)
You can also:
Add an hourly check (e.g., pick mid-day hours with best odds of clear skies).
Prefer weekends if the user says “this weekend.”
Bound to a window (“Tue-Thu”) if present in the message.
Beyond weather: adding more tools
One of CrewAI’s strengths is tool extensibility. For example:
Web Search (Serper, Tavily, etc.) for venue details.
Calendar tool to check personal availability.
Maps/Distance tool for travel time.
Add them with @tool and mount on the right agent. CrewAI’s tool docs and community examples are a great reference. (CrewAI Documentation)
Sequential vs. Hierarchical
Sequential is simple: weather_task then planner_task.
Hierarchical introduces a manager that can validate intermediate outputs, re-ask tools, or add follow-up tasks. When your API grows to 4–6 agents (e.g., Router → Weather → Activities → Writer), hierarchical pays off. (CrewAI Documentation)
Switching is a one-liner:
crew = Crew(
agents=[weather_analyst, trip_planner],
tasks=[weather_task, planner_task],
process=Process.SEQUENTIAL
)
Production tips
Timezones: Always pass a user timezone to Open-Meteo when requesting daily variables. It’s required and ensures sunrise/sunset alignment. (Open Meteo)
Validation & Guardrails: Add JSON schema checks on the meteorologist output so the planner never sees malformed data.
Observability: Log each tool call and response for debugging bad picks.
Caching: Cache geocoding and daily forecasts for a few minutes to reduce latency.
Model control: Pin models via env (OPENAI_MODEL_NAME) and consider a cheaper model for tool-heavy agents. (CrewAI Documentation)
Error paths: If no “clear” days exist, return the least-bad candidates (we do), and explain the trade-off.
FAQ
Can I swap weather providers?Yes. The tool boundary makes it easy. If you prefer OpenWeather, adjust the tool and map its condition codes (e.g., 800=Clear). (OpenWeatherMap)
How does the API support arbitrary questions?Add more agents (e.g., “Router” or “Researcher”) and tools. The Coordinator agent can decide which tasks to run, turning this into a general multi-agent chat backend.
Do I have to return a paragraph?No—return structured JSON and let your frontend render. Just change planner_task.expected_output and the final response.
Wrap-up
With CrewAI, you can stand up a tool-using, multi-agent chat API in a few dozen lines:
Agents with clear roles and tools do the heavy lifting.
Processes (sequential or hierarchical) scale your complexity.
FastAPI keeps deployment and integration straightforward.
This architecture is easy to extend: add a router agent, more domain tools (RAG search, calendars, ticketing), and richer output schemas as your assistant grows.
References
CrewAI docs (agents, tools, tasks, processes, LLMs). (CrewAI Documentation)
Open-Meteo (free weather, geocoding, daily variables). (Open Meteo)
WMO/Weather codes context. (Open Meteo)
Happy building—and enjoy that sunny Golden Gate day!
Comments