How to Build a Real Estate AI Agent Using LangChain and ZenRows

pip3 install langchain langchain-openai langchain-core langgraph langchain-zenrows python-dotenv aiohttp

ZENROWS_API_KEY=your_zenrows_api_key
OPENAI_API_KEY=your_openai_api_key

# agent Settings
MAX_LISTINGS_PER_QUERY=5
MODEL_NAME=gpt-4o-mini
TEMPERATURE=0

import os
import json
import urllib.parse
import time
import csv
import aiohttp
from typing import Optional
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_zenrows import ZenRowsUniversalScraper
from dotenv import load_dotenv

# ...
# load environment variables from .env file
load_dotenv()

# ...
# api keys and pipeline configuration from .env
ZENROWS_API_KEY = os.getenv("ZENROWS_API_KEY")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
TEMPERATURE = float(os.getenv("TEMPERATURE", "0"))
MAX_RESULTS = int(os.getenv("MAX_LISTINGS_PER_QUERY", "5"))

# set csv columns for property saving/export
CSV_FIELDS = ['title', 'price', 'location', 'property_type', 'beds', 'baths', 'sqft', 'description', 'agent', 'url']

# ...
@tool
async def search_properties(
    location: str, 
    property_type: Optional[str] = None, 
    max_price: Optional[str] = None,
    bedrooms: Optional[int] = None, 
    max_results: int = MAX_RESULTS
) -> str:
    """
    search for real estate properties using google search. returns urls to property listing pages.

    use this first to find property listing pages, then use process_property_listing on each url.

    args:
        location: city and state (e.g. 'Miami, FL')
        property_type: type like 'house', 'condo', 'apartment' (optional)
        max_price: max price like '800000' or '500K' (optional)
        bedrooms: number of bedrooms (optional)
        max_results: how many results to return (default 5)
    """
    try:
        # compose the query from all user-supplied filters
        parts = [f"properties for sale {location}"]
        if property_type:    # optionally append property type (if specified)
            parts.append(property_type)
        if bedrooms:
            parts.append(f"{bedrooms} bedroom")
        if max_price:
            parts.append(f"under ${max_price}")

        # join parts and print for debug
        query = " ".join(parts)
        print(f"\n[search] searching for: {query}")
        
        # url encode for use in SERP API; queries Google for real estate listings
        url = f"https://serp.api.zenrows.com/v1/targets/google/search/{urllib.parse.quote(query)}"

        # open an aiohttp session and make request to ZenRows SERP endpoint with api key
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params={"apikey": ZENROWS_API_KEY}, timeout=10) as response:
                response.raise_for_status()      # will raise exception on non-200
                data = await response.json()

        # result extraction: filter each result for a valid link, take at most max_results
        results = [
            {'title': item.get('title', ''), 'url': link, 'snippet': item.get('snippet', '')}
            for item in data.get('organic_results', [])[:max_results] 
            if (link := item.get('link', ''))
        ]

        # handle situation when no URLs are found in organic results
        if not results:
            print(f"[search] no results found")
            return json.dumps({"error": "no results found"})

        # print out all found results for debug/logging
        print(f"[search] found {len(results)} property listing pages")
        for i, r in enumerate(results, 1):
            print(f"  {i}. {r['title'][:60]}...")

        # return JSON string containing results and the count
        return json.dumps({"results": results, "count": len(results)})

    except Exception as e:
        # on any error (network, parse, rate limit etc.), print and return error message as JSON
        print(f"[search] error: {str(e)}")
        return json.dumps({"error": str(e)})

# ...
@tool
async def process_property_listing(url: str) -> str:
    """
    scrape a property listing url and extract all properties found on that page.

    this handles both scraping and extraction automatically. call this on each url 
    from search_properties results. returns extracted property data as json.

    args:
        url: the property listing page url to scrape
    """
    try:
        # timer for monitoring total scrape/extraction time
        start_time = time.time()
        
        # invoke ZenRows scraper using dynamic rendering and proxies for robust fetch
        scraper = ZenRowsUniversalScraper(zenrows_api_key=ZENROWS_API_KEY)
        markdown = await scraper.ainvoke({
            "url": url, 
            "js_render": "true", 
            "premium_proxy": "true", 
            "response_type": "markdown",
            "wait": "500"
        })
        
        # ensure enough page content scraped to proceed; else skip as insufficient
        if not markdown or len(markdown) < 200:
            return json.dumps({"status": "failed", "url": url, "error": "insufficient content"})
        
        # only send at most 8000 characters to LLM extractor
        content = markdown[:8000] if len(markdown) > 8000 else markdown

# ...
        # ... 
        # configure LLM to use strict json_object output for extraction step
        llm = ChatOpenAI(
            model=MODEL_NAME, 
            temperature=TEMPERATURE, 
            api_key=OPENAI_API_KEY,
            model_kwargs={"response_format": {"type": "json_object"}}
        )
        
        # prepare very specific extraction prompt to guide LLM to output consistent structured JSON
        prompt = f"""extract all property listings from this real estate page.

url: {url}

content:
{content}

return json with this exact structure:
{{
  "properties": [
    {{
      "title": "property title",
      "price": "price string",
      "location": "address",
      "property_type": "house/apartment/condo/townhouse",
      "beds": "bedrooms",
      "baths": "bathrooms",
      "sqft": "square feet",
      "description": "brief description",
      "agent": "agent name",
      "url": "{url}"
    }}
  ]
}}

extract every property listing on the page. if field missing use null. if no properties return {{"properties": []}}"""
        
        # do extraction step with LLM (with required structure enforced by model_kwargs)
        response = await llm.ainvoke(prompt)
        
        # parse response. fail if LLM does not produce valid JSON content
        result = json.loads(response.content if isinstance(response.content, str) else str(response.content))
        properties = result.get('properties', [])
        
        # ensure every property object has a reference url set
        for prop in properties:
            if not prop.get('url'):
                prop['url'] = url
        
        # monitor/how long the extraction (including LLM) took
        total_time = time.time() - start_time
        
        # report properties, url, count, timing and basic status
        return json.dumps({
            "status": "success" if properties else "no_properties",
            "url": url, 
            "properties": properties,
            "count": len(properties),
            "total_time": round(total_time, 1)
        })
        
    except Exception as e:
        # if anything fails, return error string with a short message for diagnosis
        return json.dumps({"status": "error", "url": url, "error": str(e)[:100]})

# ...
@tool
async def save_properties_to_csv(properties_json: str, filename: Optional[str] = None) -> str:
    """
    save property data to csv file.

    call this after collecting properties from process_property_listing calls.
    pass the combined properties as json string.

    args:
        properties_json: json string containing list of property dictionaries
        filename: csv filename (optional, auto-generated if not provided)
    """
    try:
        # ensure properties_json is parsed into python list/dict
        properties = json.loads(properties_json) if isinstance(properties_json, str) else properties_json
        
        # if no valid property objects are present, abort
        if not properties:
            return json.dumps({"error": "no properties to save"})

        # select default output filename if not specified, force .csv extension
        if not filename:
            filename = f"properties_{time.strftime('%Y%m%d_%H%M%S')}.csv"
        if not filename.endswith('.csv'):
            filename += '.csv'

        # open csv file for writing and dump header/rows
        with open(filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
            writer.writeheader()
            for prop in properties:
                writer.writerow({field: prop.get(field) for field in CSV_FIELDS})

        # report status, output filename, count of properties saved
        return json.dumps({
            "status": "success", 
            "filename": filename, 
            "properties_saved": len(properties)
        })

    except Exception as e:
        # handle all errors and return error status and string
        return json.dumps({"error": str(e)})

import asyncio
import json
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from tools import (
    search_properties, 
    process_property_listing, 
    save_properties_to_csv,
    MODEL_NAME, 
    TEMPERATURE, 
    OPENAI_API_KEY, 
    ZENROWS_API_KEY
)

# ...
def create_real_estate_agent():
    """create real estate agent with tools and memory."""
    # initialize openai chat model with loaded config
    llm = ChatOpenAI(model=MODEL_NAME, temperature=TEMPERATURE, api_key=OPENAI_API_KEY)
    # register agent tools for search, data extraction, and saving
    tools = [search_properties, process_property_listing, save_properties_to_csv]

    # system prompt outlines the workflow for the agent and summary policy
    system_prompt = """you are a real estate data aggregation agent.

workflow:
1. parse user's search criteria (location required, price/type/beds optional)
2. use search_properties to find property listing urls
3. call process_property_listing on each url to extract property data
4. collect all extracted properties from the tool results
5. combine them into a single json array
6. call save_properties_to_csv with the combined properties json
7. summarize results: total found, price ranges, csv filename

important:
- if user provides a url directly, call process_property_listing on it
- the properties are returned in each process_property_listing call - collect them
- pass all collected properties to save_properties_to_csv as json string
- track failed urls but continue processing others
- provide friendly summaries without technical errors"""

    # build the agent with tools, prompt, and memory checkpointing
    return create_agent(
        model=llm, 
        tools=tools, 
        system_prompt=system_prompt, 
        checkpointer=MemorySaver()
    )

# ...
async def execute_search(query: str, search_limit: int = 5) -> dict:
    """
    execute real estate search workflow.

    args:
        query: search query describing properties to find
        search_limit: max number of properties to search
        
    returns:
        dict with agent response
    """
    # check for required api keys in environment
    if not OPENAI_API_KEY or not ZENROWS_API_KEY:
        print("\nerror: missing api keys in .env file")
        return {"success": False, "error": "missing api keys"}

    # print basic workflow initiation info
    print("\n" + "="*80)
    print("real estate agent - starting workflow")
    print("="*80)
    print(f"query: {query}")
    print(f"search limit: {search_limit}")
    print("="*80)

    # create and configure the autonomous agent
    agent = create_real_estate_agent()
    config = {"configurable": {"thread_id": "real_estate_conversation"}}
    
    # create enhanced prompt for the agent (enforces limit)
    enhanced_query = f"{query} (find up to {search_limit} properties)"
    inputs = {"messages": [HumanMessage(content=enhanced_query)]}

    try:
        print("\n[agent] agent is working...\n")
        # invoke the agent with inputs in batch, capturing result
        result = await agent.ainvoke(inputs, config=config)
        
        # extract the agent's reply message
        if "messages" in result:
            final_message = result["messages"][-1]
            response = final_message.content if hasattr(final_message, 'content') else str(final_message)
        else:
            response = str(result)
        
        # print the agent's output in a clear format
        print("\n" + "="*80)
        print("agent response:")
        print("="*80)
        print(response)
        print("="*80 + "\n")
        
        # package response as result dictionary
        return {"success": True, "response": response}
        
    except Exception as e:
        print(f"\n[agent] error: {str(e)}\n")
        return {"success": False, "error": str(e)}

async def main():
    """example usage"""    
    result = await execute_search(
        query="find properties for sale in miami, florida under $500k",
        search_limit=10
    )

    if result['success']:
        print(f"\ndata aggregation completed successfully")
    else:
        print(f"\ndata aggregation failed: {result.get('error')}")

# run the example pipeline when executed directly
if __name__ == "__main__":
    asyncio.run(main())

python agent.py

================================================================================
real estate agent - starting workflow
================================================================================
query: find properties for sale in miami, florida under $500k
search limit: 10
================================================================================

[agent] agent is working...


[search] searching for: properties for sale Miami, FL under $500000
[search] found 10 property listing pages
  1. Homes for Sale Under $500K in Miami FL...
  2. Homes for Sale Under $500k in Miami, FL...
  3. Miami, FL Homes for Sale under $500K...
  4. Homes For Sale Under $500K in Miami Gardens, FL...
  5. 1.7K+ Homes for Sale Under 500K in Miami, Florida...
  6. New Homes in Miami & Dade County, FL Under $500K...
  7. Miami Homes For Sale...
  8. Miami/Dade County FL Houses under ......
  9. Homes for Sale Under $500K in Miami-Dade County FL...
  10. Top 6 Homes for Sale in Miami for Under $500000...

================================================================================
agent response:
================================================================================
I found a total of **22 properties** for sale in Miami, Florida, under $500,000. Here are some highlights:

### Price Ranges:
- The lowest price is **$195,000** for a condo at 871 Ne 207th Ter #102.
- The highest price is **$500,000** for several condos, including one at 229 SW 9th #218.

### Summary of Properties:
- **Total Properties Found:** 22
- **Price Range:** $195,000 - $500,000
- **CSV Filename:** [miami_properties_under_500k.csv](sandbox:/miami_properties_under_500k.csv)

If you need more details or further assistance, feel free to ask!
================================================================================


data aggregation completed successfully