How to Build a Multi-Agent Property Listing Generator with Claude API

import os
import json
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

client = Anthropic()
MODEL = "claude-sonnet-4-6"


class Agent:
    """
    Base agent class. Each specialized agent inherits from this.
    Handles the message loop, tool dispatch, and history tracking.
    """

    def __init__(self, name: str, system_prompt: str, tools: list):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools
        self.messages = []

    def run(self, user_message: str) -> str:
        """Send a message to this agent and get a final text response."""
        self.messages.append({"role": "user", "content": user_message})

        while True:
            response = client.messages.create(
                model=MODEL,
                max_tokens=2048,
                system=self.system_prompt,
                tools=self.tools,
                messages=self.messages,
            )

            # Append assistant response to history
            self.messages.append({"role": "assistant", "content": response.content})

            # If Claude is done and returning text, extract and return it
            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return ""

            # If Claude wants to use a tool, handle it
            if response.stop_reason == "tool_use":
                tool_results = self._handle_tool_calls(response.content)
                self.messages.append({"role": "user", "content": tool_results})
            else:
                # Unexpected stop reason — bail out gracefully
                break

        return ""

    def _handle_tool_calls(self, content_blocks: list) -> list:
        """Process all tool_use blocks and return tool_result blocks."""
        results = []
        for block in content_blocks:
            if block.type == "tool_use":
                tool_output = self._dispatch_tool(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(tool_output),
                })
        return results

    def _dispatch_tool(self, tool_name: str, tool_input: dict) -> dict:
        """Override this in subclasses to handle specific tools."""
        return {"error": f"Tool '{tool_name}' not implemented in {self.name}"}

import os
import json
from agent_base import Agent, client, MODEL
from dotenv import load_dotenv

load_dotenv()


# ── Simulated MLS data store ──────────────────────────────────────────────────
MLS_DATABASE = {
    "MLS-10482": {
        "address": "4821 Gulf Shore Blvd N, Naples, FL 34103",
        "bedrooms": 3,
        "bathrooms": 2.5,
        "sqft": 2340,
        "year_built": 2018,
        "lot_size_acres": 0.24,
        "features": ["pool", "two-car garage", "chef's kitchen", "impact windows",
                     "smart home system", "covered lanai", "hardwood floors"],
        "hoa_monthly": 450,
        "taxes_annual": 8200,
        "days_on_market": 0,
        "neighborhood": "Park Shore",
        "proximity": "0.3 miles to beach, 1.2 miles to Venetian Village",
    }
}

COMP_DATABASE = {
    "Park Shore": [
        {"address": "4710 Gulf Shore Blvd N", "sqft": 2280, "sold_price": 1_850_000,
         "days_on_market": 18},
        {"address": "4930 Gulf Shore Blvd N", "sqft": 2410, "sold_price": 1_975_000,
         "days_on_market": 12},
        {"address": "4605 Gulf Shore Blvd N", "sqft": 2300, "sold_price": 1_820_000,
         "days_on_market": 24},
    ]
}

PHOTO_TAGS_DATABASE = {
    "MLS-10482": [
        {"filename": "front_exterior.jpg", "detected_features": ["curb appeal", "paver driveway",
                                                                   "lush landscaping", "impact windows"]},
        {"filename": "kitchen.jpg", "detected_features": ["quartz countertops", "stainless appliances",
                                                           "island seating", "pendant lighting"]},
        {"filename": "pool_lanai.jpg", "detected_features": ["heated pool", "covered lanai",
                                                              "outdoor kitchen", "privacy hedges"]},
        {"filename": "master_suite.jpg", "detected_features": ["tray ceiling", "walk-in closet",
                                                                "spa bath", "natural light"]},
    ]
}


# ── Tool handler functions ────────────────────────────────────────────────────

def fetch_mls_data(mls_id: str) -> dict:
    """Return raw MLS record for a given listing ID."""
    return MLS_DATABASE.get(mls_id, {"error": f"No listing found for {mls_id}"})


def fetch_comparable_sales(neighborhood: str) -> dict:
    """Return recent comparable sales for a neighborhood."""
    comps = COMP_DATABASE.get(neighborhood, [])
    if not comps:
        return {"error": f"No comps found for {neighborhood}"}
    avg_price_per_sqft = sum(c["sold_price"] / c["sqft"] for c in comps) / len(comps)
    return {
        "neighborhood": neighborhood,
        "comps": comps,
        "average_price_per_sqft": round(avg_price_per_sqft, 2),
    }


def fetch_photo_tags(mls_id: str) -> dict:
    """Return AI-detected photo feature tags for a listing."""
    tags = PHOTO_TAGS_DATABASE.get(mls_id, [])
    if not tags:
        return {"error": f"No photo tags found for {mls_id}"}
    all_features = []
    for photo in tags:
        all_features.extend(photo["detected_features"])
    return {"mls_id": mls_id, "photos": tags, "all_detected_features": list(set(all_features))}


def format_listing_output(headline: str, description: str,
                           features: list, pricing_narrative: str) -> dict:
    """Assemble final listing dictionary from agent outputs."""
    return {
        "headline": headline,
        "description": description,
        "key_features": features,
        "pricing_narrative": pricing_narrative,
    }


# ── Description Agent ─────────────────────────────────────────────────────────

DESCRIPTION_SYSTEM = """You are a luxury real estate copywriter specializing in Naples, Florida.
Your job is to write compelling property descriptions that attract high-net-worth buyers.
Use the MLS data and photo tags tools to gather information before writing.
Keep descriptions between 120 and 150 words. Lead with lifestyle, not specs."""

DESCRIPTION_TOOLS = [
    {
        "name": "fetch_mls_data",
        "description": "Retrieve full MLS property data by listing ID",
        "input_schema": {
            "type": "object",
            "properties": {
                "mls_id": {"type": "string", "description": "The MLS listing ID, e.g. MLS-10482"}
            },
            "required": ["mls_id"],
        },
    },
    {
        "name": "fetch_photo_tags",
        "description": "Retrieve AI-detected visual features from listing photos",
        "input_schema": {
            "type": "object",
            "properties": {
                "mls_id": {"type": "string", "description": "The MLS listing ID"}
            },
            "required": ["mls_id"],
        },
    },
]


class DescriptionAgent(Agent):
    def __init__(self):
        super().__init__("DescriptionAgent", DESCRIPTION_SYSTEM, DESCRIPTION_TOOLS)

    def _dispatch_tool(self, tool_name: str, tool_input: dict) -> dict:
        if tool_name == "fetch_mls_data":
            return fetch_mls_data(tool_input["mls_id"])
        if tool_name == "fetch_photo_tags":
            return fetch_photo_tags(tool_input["mls_id"])
        return super()._dispatch_tool(tool_name, tool_input)


# ── Pricing Agent ─────────────────────────────────────────────────────────────

PRICING_SYSTEM = """You are a real estate pricing analyst. Your job is to analyze comparable
sales and produce a short, confident pricing narrative (2-3 sentences) that justifies a
suggested list price. Always show your reasoning based on price-per-square-foot data."""

PRICING_TOOLS = [
    {
        "name": "fetch_mls_data",
        "description": "Retrieve full MLS property data by listing ID",
        "input_schema": {
            "type": "object",
            "properties": {
                "mls_id": {"type": "string", "description": "The MLS listing ID"}
            },
            "required": ["mls_id"],
        },
    },
    {
        "name": "fetch_comparable_sales",
        "description": "Retrieve recent comparable sales data for a neighborhood",
        "input_schema": {
            "type": "object",
            "properties": {
                "neighborhood": {"type": "string", "description": "Neighborhood name, e.g. Park Shore"}
            },
            "required": ["neighborhood"],
        },
    },
]


class PricingAgent(Agent):
    def __init__(self):
        super().__init__("PricingAgent", PRICING_SYSTEM, PRICING_TOOLS)

    def _dispatch_tool(self, tool_name: str, tool_input: dict) -> dict:
        if tool_name == "fetch_mls_data":
            return fetch_mls_data(tool_input["mls_id"])
        if tool_name == "fetch_comparable_sales":
            return fetch_comparable_sales(tool_input["neighborhood"])
        return super()._dispatch_tool(tool_name, tool_input)


# ── Photo Tagging Agent ───────────────────────────────────────────────────────

PHOTO_SYSTEM = """You are a real estate media specialist. Your job is to review AI-detected
photo tags and return a clean, prioritized list of the 6 most marketable visual features
for a listing. Return them as a JSON array of strings only — no extra commentary."""

PHOTO_TOOLS = [
    {
        "name": "fetch_photo_tags",
        "description": "Retrieve AI-detected visual features from listing photos",
        "input_schema": {
            "type": "object",
            "properties": {
                "mls_id": {"type": "string", "description": "The MLS listing ID"}
            },
            "required": ["mls_id"],
        },
    }
]


class PhotoTagAgent(Agent):
    def __init__(self):
        super().__init__("PhotoTagAgent", PHOTO_SYSTEM, PHOTO_TOOLS)

    def _dispatch_tool(self, tool_name: str, tool_input: dict) -> dict:
        if tool_name == "fetch_photo_tags":
            return fetch_photo_tags(tool_input["mls_id"])
        return super()._dispatch_tool(tool_name, tool_input)

import os
import json
from dotenv import load_dotenv

load_dotenv()

# These are the tool definitions passed to the orchestrator agent.
# The orchestrator doesn't call the specialized agents as tools —
# instead it calls them internally in Python and uses their outputs
# to populate the format_listing_output tool.

ORCHESTRATOR_TOOLS = [
    {
        "name": "fetch_mls_data",
        "description": "Retrieve raw MLS listing data including bedrooms, sqft, features, and location",
        "input_schema": {
            "type": "object",
            "properties": {
                "mls_id": {
                    "type": "string",
                    "description": "The MLS listing ID, e.g. MLS-10482",
                }
            },
            "required": ["mls_id"],
        },
    },
    {
        "name": "format_listing_output",
        "description": (
            "Assemble the final property listing from all agent outputs. "
            "Call this only after you have the description, pricing narrative, and features."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "headline": {
                    "type": "string",
                    "description": "A compelling single-line listing headline under 12 words",
                },
                "description": {
                    "type": "string",
                    "description": "The full property description from the description agent",
                },
                "features": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Top 6 marketable features from the photo tag agent",
                },
                "pricing_narrative": {
                    "type": "string",
                    "description": "2-3 sentence pricing justification from the pricing agent",
                },
            },
            "required": ["headline", "description", "features", "pricing_narrative"],
        },
    },
]

import os
import json
from anthropic import Anthropic
from dotenv import load_dotenv
from specialized_agents import (
    DescriptionAgent,
    PricingAgent,
    PhotoTagAgent,
    fetch_mls_data,
    format_listing_output,
)
from orchestrator_tools import ORCHESTRATOR_TOOLS

load_dotenv()

client = Anthropic()
MODEL = "claude-sonnet-4-6"

ORCHESTRATOR_SYSTEM = """You are a real estate listing orchestrator. Your job is to coordinate
three specialist agents and produce a complete, formatted property listing.

Follow this exact sequence:
1. Call fetch_mls_data to understand the property
2. Use that data to inform the description, pricing, and photo contexts
3. Call format_listing_output with the headline, description, features list, and pricing narrative
   that you have received from the specialist agents

You will be given the specialist agent outputs directly in your user messages.
Do not invent data — only use what the agents provide."""


class ListingOrchestrator:
    """
    Coordinates DescriptionAgent, PricingAgent, and PhotoTagAgent,
    then uses Claude to assemble and format the final listing.
    """

    def __init__(self):
        self.description_agent = DescriptionAgent()
        self.pricing_agent = PricingAgent()
        self.photo_agent = PhotoTagAgent()
        self.messages = []

    def generate_listing(self, mls_id: str) -> dict:
        """Full pipeline: run all agents, orchestrate, return final listing dict."""
        print(f"\n[Orchestrator] Starting listing generation for {mls_id}")

        # ── Step A: Run specialist agents in parallel (simulated sequentially here)
        print("[Orchestrator] Running DescriptionAgent...")
        description = self.description_agent.run(
            f"Write a luxury property description for listing {mls_id}. "
            f"Use the fetch_mls_data and fetch_photo_tags tools to gather details first."
        )

        print("[Orchestrator] Running PricingAgent...")
        pricing = self.pricing_agent.run(
            f"Analyze listing {mls_id} and write a 2-3 sentence pricing narrative. "
            f"Use fetch_mls_data first, then fetch_comparable_sales for the neighborhood."
        )

        print("[Orchestrator] Running PhotoTagAgent...")
        features_raw = self.photo_agent.run(
            f"Retrieve photo tags for listing {mls_id} and return the top 6 marketable "
            f"features as a JSON array of strings."
        )

        # Parse photo agent's JSON output safely
        try:
            features = json.loads(features_raw)
            if not isinstance(features, list):
                features = [features_raw]
        except json.JSONDecodeError:
            # If Claude added surrounding text, try to extract the array
            start = features_raw.find("[")
            end = features_raw.rfind("]") + 1
            if start != -1 and end > start:
                features = json.loads(features_raw[start:end])
            else:
                features = [features_raw]

        print("[Orchestrator] All specialist agents complete. Running final assembly...")

        # ── Step B: Hand everything to Claude orchestrator to assemble
        combined_context = (
            f"Specialist agent outputs for {mls_id}:\n\n"
            f"DESCRIPTION:\n{description}\n\n"
            f"PRICING NARRATIVE:\n{pricing}\n\n"
            f"TOP FEATURES (from photo agent):\n{json.dumps(features)}\n\n"
            f"Now call format_listing_output with a compelling headline you write yourself, "
            f"plus these description, features, and pricing_narrative values."
        )

        self.messages.append({"role": "user", "content": combined_context})

        final_listing = {}

        # ── Step C: Orchestrator run loop
        while True:
            response = client.messages.create(
                model=MODEL,
                max_tokens=1024,
                system=ORCHESTRATOR_SYSTEM,
                tools=ORCHESTRATOR_TOOLS,
                messages=self.messages,
            )

            self.messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                # Orchestrator finished without calling format_listing_output
                print("[Orchestrator] Warning: ended without calling format tool.")
                break

            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type