A CrewAI crew for real estate market analysis

A crewai real estate market analysis example needs to do more than call an LLM — it needs to fetch live data, run calculations, and produce output a human can act on. This tutorial walks through building a three-agent crew that scrapes active listings, computes investment metrics, and emits a structured markdown report. You’ll end up with runnable code you can extend for your own market or asset class.

Step 1: Define the scope and data sources

Before writing agents, decide what “market analysis” means for your use case. For this crew we target single-family rentals in a specific MSA. The data pipeline looks like:

1. Listings source — Redfin/Realtor.com via their public APIs or HTML scraping (we’ll use a lightweight scraper against Redfin’s search endpoint).
2. Economic context — FRED series for metro-level unemployment, median income, and building permits (via fredapi).
3. Property records — County assessor APIs where available; fallback to ATTOM or Estated if you have keys.

For the tutorial we’ll stick to Redfin listings + FRED. Both are free and require only an API key for FRED.

Create a virtual environment and install dependencies:

python -m venv .venv && source .venv/bin/activate
pip install crewai==0.67.0 requests beautifulsoup4 pandas fredapi python-dotenv pydantic

Set up .env:

FRED_API_KEY=your_fred_key_here
OPENAI_API_KEY=your_openai_key_here
# Optional: if you route through n4n.ai, set OPENAI_BASE_URL=https://api.n4n.ai/v1

Step 2: Build the data layer

Agents should not contain scraping logic. Extract it into a reusable module so you can unit-test and swap sources.

# data_sources/redfin.py
import requests
from bs4 import BeautifulSoup
import pandas as pd
from typing import List, Dict
import time
import random

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
}

def fetch_listings(metro_code: str, max_pages: int = 3) -> List[Dict]:
    """
    Scrape Redfin search results for a metro area.
    metro_code examples: '12420' (Austin), '35620' (NYC), '19100' (Dallas)
    """
    listings = []
    base_url = f"https://www.redfin.com/stingray/api/gis-csv?al=1&market={metro_code}&ord=redfin-recommended-asc&page_number={{}}&region_id={metro_code}&region_type=6&sold_within_days=180&status=9&uipt=1,2,3,4,5,6,7&v=8"

    for page in range(1, max_pages + 1):
        url = base_url.format(page)
        resp = requests.get(url, headers=HEADERS, timeout=15)
        if resp.status_code != 200:
            break
        # Redfin returns CSV with a junk first line
        lines = resp.text.strip().split('\n')
        if len(lines) < 2:
            break
        df = pd.read_csv(pd.io.common.StringIO('\n'.join(lines[1:])))
        listings.extend(df.to_dict('records'))
        time.sleep(random.uniform(1.5, 3.0))  # be polite
    return listings


def normalize_listing(raw: Dict) -> Dict:
    """Map Redfin columns to our internal schema."""
    return {
        "address": raw.get("ADDRESS", ""),
        "city": raw.get("CITY", ""),
        "state": raw.get("STATE OR PROVINCE", ""),
        "zip_code": raw.get("ZIP OR POSTAL CODE", ""),
        "price": _parse_money(raw.get("PRICE", "")),
        "beds": _parse_int(raw.get("BEDS", "")),
        "baths": _parse_float(raw.get("BATHS", "")),
        "sqft": _parse_int(raw.get("SQUARE FEET", "")),
        "lot_size": _parse_lot(raw.get("LOT SIZE", "")),
        "year_built": _parse_int(raw.get("YEAR BUILT", "")),
        "days_on_market": _parse_int(raw.get("DAYS ON MARKET", "")),
        "property_type": raw.get("PROPERTY TYPE", ""),
        "latitude": raw.get("LATITUDE"),
        "longitude": raw.get("LONGITUDE"),
    }


def _parse_money(val: str) -> float:
    if not val or pd.isna(val):
        return 0.0
    return float(str(val).replace("$", "").replace(",", ""))


def _parse_int(val: str) -> int:
    try:
        return int(float(str(val).replace(",", "")))
    except (ValueError, TypeError):
        return 0


def _parse_float(val: str) -> float:
    try:
        return float(str(val).replace(",", ""))
    except (ValueError, TypeError):
        return 0.0


def _parse_lot(val: str) -> float:
    """Convert lot size strings to acres."""
    if not val or pd.isna(val):
        return 0.0
    val = str(val).lower().strip()
    if "acre" in val:
        return float(val.replace("acre", "").replace("s", "").strip())
    # assume sqft
    try:
        return float(val.replace(",", "")) / 43560
    except ValueError:
        return 0.0

# data_sources/fred.py
from fredapi import Fred
import pandas as pd
from typing import Dict
import os

fred = Fred(api_key=os.getenv("FRED_API_KEY"))

# FRED series IDs for metro-level indicators
SERIES_MAP = {
    "unemployment_rate": "LAUMT{metro_code}000000003A",  # metro unemployment
    "median_income": "MEHOINUS{metro_code}A052NCEN",      # median household income
    "building_permits": "BP{metro_code}0000000000SA",     # building permits
}

def fetch_metro_indicators(metro_code: str) -> Dict[str, float]:
    """Return latest value for each indicator."""
    out = {}
    for name, series_template in SERIES_MAP.items():
        series_id = series_template.format(metro_code=metro_code)
        try:
            series = fred.get_series(series_id)
            out[name] = round(float(series.dropna().iloc[-1]), 2)
        except Exception:
            out[name] = None
    return out

Step 3: Define the agents and tasks

CrewAI works best when each agent has a single responsibility and tools are explicit. We’ll create three agents:

1. Data Collector — fetches and normalizes listings + economic data
2. Investment Analyst — computes cap rate, cash-on-cash, GRM, rent estimates
3. Report Writer — synthesizes findings into a markdown brief

# crew/agents.py
from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, List, Dict, Any
import json
import statistics

# ---- Tools ----

class FetchListingsTool(BaseTool):
    name: str = "fetch_listings"
    
    def _run(self, metro_code: str, max_pages: int = 3) -> str:
        from data_sources.redfin import fetch_listings, normalize_listing
        raw = fetch_listings(metro_code, max_pages)
        normalized = [normalize_listing(r) for r in raw]
        return json.dumps(normalized)


class FetchEconomicDataTool(BaseTool):
    name: str = "fetch_economic_data"
    
    def _run(self, metro_code: str) -> str:
        from data_sources.fred import fetch_metro_indicators
        data = fetch_metro_indicators(metro_code)
        return json.dumps(data)


class ComputeMetricsTool(BaseTool):
    name: str = "compute_investment_metrics"
    
    def _run(self, listings_json: str, econ_json: str) -> str:
        listings = json.loads(listings_json)
        econ = json.loads(econ_json)
        
        results = []
        for prop in listings:
            if prop["price"] == 0 or prop["beds"] == 0:
                continue
            
            # Rent estimate: simple heuristic $/bed + $/sqft blend
            # In production, use a rent comp API (RentCast, Zillow Observed Rent Index)
            rent_per_bed = 1200  # placeholder, adjust per metro
            rent_per_sqft = 1.35
            estimated_rent = max(
                prop["beds"] * rent_per_bed,
                prop["sqft"] * rent_per_sqft
            )
            
            # Annual expenses (rough rule of thumb: 50% rule for SFR)
            annual_expenses = estimated_rent * 12 * 0.5
            noi = estimated_rent * 12 - annual_expenses
            
            cap_rate = noi / prop["price"] if prop["price"] else 0
            grm = prop["price"] / (estimated_rent * 12) if estimated_rent else 0
            
            # Cash-on-cash assumes 25% down, 6.5% rate, 30yr
            loan_amount = prop["price"] * 0.75
            monthly_pmt = loan_amount * (0.065/12) / (1 - (1 + 0.065/12)**(-360))
            annual_debt = monthly_pmt * 12
            pre_tax_cash_flow = noi - annual_debt
            cash_invested = prop["price"] * 0.25 + prop["price"] * 0.03  # down + closing
            coc_return = pre_tax_cash_flow / cash_invested if cash_invested else 0
            
            results.append({
                "address": prop["address"],
                "price": prop["price"],
                "beds": prop["beds"],
                "baths": prop["baths"],
                "sqft": prop["sqft"],
                "estimated_monthly_rent": round(estimated_rent, 0),
                "cap_rate": round(cap_rate * 100, 2),
                "grm": round(grm, 2),
                "cash_on_cash_return": round(coc_return * 100, 2),
                "noi": round(noi, 0),
            })
        
        # Add market summary
        if results:
            cap_rates = [r["cap_rate"] for r in results]
            grms = [r["grm"] for r in results]
            cocs = [r["cash_on_cash_return"] for r in results]
            summary = {
                "median_cap_rate": round(statistics.median(cap_rates), 2),
                "median_grm": round(statistics.median(grms), 2),
                "median_coc": round(statistics.median(cocs), 2),
                "property_count": len(results),
            }
        else:
            summary = {}
        
        return json.dumps({"properties": results, "market_summary": summary, "economic_context": econ})


class WriteReportTool(BaseTool):
    name: str = "write_markdown_report"
    
    def _run(self, analysis_json: str, metro_name: str) -> str:
        data = json.loads(analysis_json)
        props = data.get("properties", [])
        summary = data.get("market_summary", {})
        econ = data.get("economic_context", {})
        
        lines = [
            f"# {metro_name} Single-Family Rental Market Brief",
            f"\n*Generated by CrewAI investment crew*",
            f"\n## Market Summary",
            f"- **Properties analyzed**: {summary.get('property_count', 0)}",
            f"- **Median cap rate**: {summary.get('median_cap_rate', 'N/A')}%",
            f"- **Median GRM**: {summary.get('median_grm', 'N/A')}",
            f"- **Median cash-on-cash**: {summary.get('median_coc', 'N/A')}%",
            f"\n## Economic Context",
            f"- Unemployment rate: {econ.get('unemployment_rate', 'N/A')}%",
            f"- Median household income: ${econ.get('median_income', 'N/A'):,}" if econ.get('median_income') else "- Median household income: N/A",
            f"- Building permits (SAAR): {econ.get('building_permits', 'N/A'):,}" if econ.get('building_permits') else "- Building permits: N/A",
            f"\n## Top Opportunities (Cap Rate ≥ Median)",
        ]
        
        if props:
            median_cap = summary.get("median_cap_rate", 0)
            top_props = [p for p in props if p["cap_rate"] >= median_cap]
            top_props.sort(key=lambda x: x["cap_rate"], reverse=True)
            
            for p in top_props[:10]:
                lines.append(
                    f"- **{p['address']}** — ${p['price']:,} | "
                    f"{p['beds']}bd/{p['baths']}ba | {p['sqft']:,} sqft | "
                    f"Est. rent: ${p['estimated_monthly_rent']:,.0f}/mo | "
                    f"Cap: {p['cap_rate']}% | GRM: {p['grm']} | CoC: {p['cash_on_cash_return']}%"
                )
        
        lines.extend([
            f"\n## Methodology Notes",
            f"- Rent estimates use metro-level heuristics; replace with comp API for production.",
            f"- Expenses assume 50% rule (taxes, insurance, maintenance, vacancy, mgmt).",
            f"- Financing: 25% down, 6.5% fixed 5/8% rate, 30-year amortization.",
            f"- Data sources: Redfin listings (last 180 days), FRED economic series.",
        ])
        
        return "\n".join(lines)


# ---- Agent Definitions ----

def create_agents(llm_model: str = "gpt-4o-mini"):
    collector = Agent(
        role="Real Estate Data Collector",
        goal="Fetch and normalize listing and economic data for the target metro",
        backstory="You specialize in pulling clean, structured property data from public sources.",
        tools=[FetchListingsTool(), FetchEconomicDataTool()],
        llm=llm_model,
        verbose=True,
    )
    
    analyst = Agent(
        role="Investment Analyst",
        goal="Compute accurate investment metrics for each property and summarize market conditions",
        backstory="You turn raw property data into cap rates, cash-on-cash returns, and GRMs using standard underwriting assumptions.",
        tools=[ComputeMetricsTool()],
        llm=llm_model,
        verbose=True,
    )
    
    writer = Agent(
        role="Investment Report Writer",
        goal="Produce a concise, actionable markdown brief for decision makers",
        backstory="You synthesize analysis into a format a principal can read in 3 minutes.",
        tools=[WriteReportTool()],
        llm=llm_model,
        verbose=True,
    )
    
    return collector, analyst, writer

Step 4: Wire the crew with explicit task dependencies

Tasks must pass outputs explicitly. CrewAI’s context parameter handles this, but being explicit about JSON serialization avoids silent failures.

# crew/tasks.py
from crewai import Task
from crew.agents import create_agents

def create_tasks(metro_code: str, metro_name: str):
    collector, analyst, writer = create_agents()
    
    collect_task = Task(
        description=(
            f"Fetch active listings for metro code {metro_code} (max 3 pages) "
            f"and economic indicators from FRED. Return both as JSON."
        ),
        expected_output="JSON with keys 'listings' (array) and 'economic' (object)",
        agent=collector,
    )
    
    analyze_task = Task(
        description=(
            "Take the listings and economic data from the previous task. "
            "Compute cap rate, GRM, cash-on-cash return, and estimated rent for each property. "
            "Return JSON with 'properties' array and 'market_summary' object."
        ),
        expected_output="JSON with investment metrics per property plus market summary",
        agent=analyst,
        context=[collect_task],
    )
    
    report_task = Task(
        description=(
            f"Render the analysis JSON into a markdown investment brief for {metro_name}. "
            "Include market summary, economic context, top opportunities, and methodology notes."
        ),
        expected_output="Complete markdown report as a single string",
        agent=writer,
        context=[analyze_task],
    )
    
    return [collect_task, analyze_task, report_task]

Step 5: Run the crew and verify output

# main.py
import json
import sys
from crewai import Crew, Process
from crew.tasks import create_tasks

METRO_CONFIG = {
    "12420": "Austin-Round Rock, TX",
    "19100": "Dallas-Fort Worth-Arlington, TX",
    "35620": "New York-Newark-Jersey City, NY-NJ-PA",
    "37980": "Phoenix-Mesa-Chandler, AZ",
    "31080": "Los Angeles-Long Beach-Anaheim, CA",
}

def run_crew(metro_code: str = "12420"):
    metro_name = METRO_CONFIG.get(metro_code, "Unknown Metro")
    tasks = create_tasks(metro_code, metro_name)
    
    crew = Crew(
        agents=[t.agent for t in tasks],
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff()
    return result


if __name__ == "__main__":
    metro = sys.argv[1] if len(sys.argv) > 1 else "12420"
    print(f"\n=== Running crew for {METRO_CONFIG.get(metro, metro)} ===\n")
    output = run_crew(metro)
    
    # Save report
    with open(f"market_brief_{metro}.md", "w") as f:
        f.write(str(output))
    
    print(f"\n=== Report saved to market_brief_{metro}.md ===")

Run it:

python main.py 12420  # Austin

Verification checklist:

1. File exists — market_brief_12420.md created in working directory
2. Structure present — Open the file; confirm it has # Austin-Round Rock, TX Single-Family Rental Market Brief, a Market Summary table, Economic Context, and at least 5 property rows under Top Opportunities
3. Metrics look sane — Cap rates 3–7%, GRM 10–18, CoC 2–10% for current Texas markets. If you see 50% cap rates or negative GRM, the rent heuristic or expense model is off for your metro — adjust rent_per_bed/rent_per_sqft in ComputeMetricsTool
4. No empty fields — Every property row should have address, price, beds, baths, sqft, and all four metrics populated
5. Economic context filled — Unemployment, income, and permits should show numeric values, not “N/A”

Step 6: Harden for production

The tutorial crew works end-to-end but has sharp edges. Address these before relying on it for real decisions:

Example: Adding a reasoning task

# crew/tasks.py (addition)
from crewai import Task
from crew.agents import create_agents

reasoning_tool = BaseTool(
    name="flag_anomalies",
    description="Identify properties where metrics deviate from market norms and explain why",
    func=lambda analysis_json: _flag_anomalies(analysis_json)
)

def _flag_anomalies(analysis_json: str) -> str:
    data = json.loads(analysis_json)
    props = data.get("properties", [])
    median_cap = data.get("market_summary", {}).get("median_cap_rate", 0)
    flags = []
    for p in props:
        if p["cap_rate"] > median_cap * 1.5:
            flags.append(f"HIGH CAP: {p['address']} at {p['cap_rate']}% — verify condition/rent roll")
        if p["grm"] < 8:
            flags.append(f"LOW GRM: {p['address']} at {p['grm']} — possible distress or data error")
    return json.dumps({"flags": flags})

# In create_tasks():
reasoning_task = Task(
    description="Review the investment metrics and flag any anomalies with explanations",
    expected_output="JSON with 'flags' array of strings",
    agent=analyst,  # reuse analyst or create a dedicated Reviewer agent
    context=[analyze_task],
    tools=[reasoning_tool],
)

Insert reasoning_task between analyze_task and report_task, and update WriteReportTool to consume the flags.

Step 7: Extend the pattern

This crew template maps to any asset class where you have:

1. A listings feed (LoopNet for commercial, LandWatch for land, BoatTrader for marine)
2. A valuation model (DCF for commercial, residual land value for development, comps for SFR)
3. A report format your stakeholders already read

Swap the data source tools and the metric tool; keep the collector/analyst/writer roles. The crew structure stays the same.

What to do next: Pick your target metro, replace the rent heuristic with a real comp API, and schedule the crew to run weekly via GitHub Actions or Airflow. Commit the markdown outputs to a repo — now you have a versioned market tracker you can diff over time.