随着智能机器人与 AI 系统的兴起，房地产行业正经历重大变革，手动工作正被全面替代。房地产企业常常面临数据分散、分析耗时、难以规模化等挑战。如果有一个能够推理、适应，并自主提供完整房产服务的系统，会怎样？
本指南你将学到：
- 现代 AI 智能体框架如何结合网络爬取基础设施来解决上述问题。
- 如何使用 CrewAI 与 Bright Data 的 MCP 服务器 创建现代化的房地产智能体。
让我们开始吧！
什么是 CrewAI？
CrewAI 是一个用于编排协作式 AI 智能体的开源框架。使用 CrewAI，你可以明确规定智能体能做什么、目标是什么，以及可使用哪些工具。这使得我们能够将智能体组织成团队（Crew），在房地产场景中执行复杂的多步骤工作流。
CrewAI 由以下核心组件组成：
- Agent（智能体）：一个由大模型驱动的“工作者”，具有明确的角色、具体目标与可选背景设定。房地产领域上下文会影响模型表现。
- Task（任务）： 指定给单个智能体的、范围清晰且有单一产出的工作，其输出可作为质量控制的基准。
- Tool（工具）： 智能体可调用的私有函数，用于领域特定功能，例如获取房产数据、进行市场分析，或调用 Bright Data 的 MCP 端点进行抓取。
- Crew（团队）： 为完成一个房地产目标而协作的一组智能体，每个智能体执行其对应任务。
- Process（流程）：执行计划，可按顺序、并行或分层方式进行，控制任务顺序、分配、委派与重复。
这类似于一个房地产团队：房产研究员负责数据提取，市场分析师提供洞察，客户经理负责沟通，挂牌专家负责营销。
想了解 CrewAI 如何与 Bright Data 等工具集成，请查看这篇指南。
什么是 MCP？
MCP 是一个开放的 JSON-RPC 2.0 标准，让 AI 智能体可以通过单一、结构化接口调用外部工具与数据源。可以把它看作房地产数据的通用连接器。
Bright Data 的 MCP 服务器将该标准落地实施，把智能体直接接入 Bright Data 的抓取技术栈，使房地产数据采集比传统方式更为简单：
- 反爬虫绕过： 请求通过 Web Unlocker 与覆盖 195 个国家、1.5 亿+ 轮换住宅 IP 池
- 动态站点支持： 专用的爬虫浏览器可渲染 JavaScript，智能体可看到完整加载的房源列表
- 结构化结果： 许多工具直接返回干净的 JSON，免去自定义解析器
该服务器发布了 50+ 开箱即用的工具，从通用 URL 抓取到房地产专用爬取器一应俱全，你的 CrewAI 智能体只需一次调用即可获取房产详情、市场数据或挂牌信息。
我们要构建什么：房地产智能体
我们将构建一个 CrewAI 房地产智能体，从 Zillow 页面研究房产，并以结构化 JSON 输出其详细信息。
你也可以通过更改链接与部分代码，将其用于其他房源。
准备条件：
在编写代码前，请确保完成以下准备：
- Python 3.11 —— 建议用于稳定性。
- Node.js + npm —— 运行 Bright Data MCP 服务器所需；从官网下载。
- Python 虚拟环境 —— 用于隔离依赖；参见
venv文档。 - Bright Data 账号 —— 注册并创建 API Token（提供免费试用额度）。
- Nebius API Key —— 在 Nebius AI Studio 创建（点击 + Get API Key）。可免费使用，无需账单信息。
步骤 1：环境配置
在终端运行以下命令，搭建项目环境并安装依赖：
mkdir real-estate-ai-system && cd real-estate-ai-system
python -m venv venv
# macOS/Linux: source venv/bin/activate
# Windows: venv\\Scripts\\activate
pip install "crewai-tools[mcp]" crewai mcp python-dotenv pandas新建 real_estate_agents.py 文件，并加入以下导入：
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters
from crewai.llm import LLM
import os
import json
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()步骤 2：配置 Bright Data MCP 服务器
在项目根目录创建 .env 文件，填入你的凭据：
BRIGHT_DATA_API_TOKEN="your_api_token_here"
WEB_UNLOCKER_ZONE="your_web_unlocker_zone"
BROWSER_ZONE="your_browser_zone"
NEBIUS_API_KEY="your_nebius_api_key"你需要：
- API Token：在 Bright Data 控制台生成新的 API Token
- Web Unlocker 区域：为房地产站点创建一个新的 Web Unlocker 区域
- Browser API 区域：为 JS 重站点创建一个新的 Browser API 区域
- Nebius API Key：已在准备条件中创建
将以下配置加入 real_estate_agents.py：
llm = LLM(
model="nebius/Qwen/Qwen3-235B-A22B",
api_key=os.getenv("NEBIUS_API_KEY")
)
server_params = StdioServerParameters(
command="npx",
args=["@brightdata/mcp"],
env={
"API_TOKEN": os.getenv("BRIGHT_DATA_API_TOKEN"),
"WEB_UNLOCKER_ZONE": os.getenv("WEB_UNLOCKER_ZONE"),
"BROWSER_ZONE": os.getenv("BROWSER_ZONE"),
},
)这会以子进程方式启动 *npx @brightdata/mcp*，并通过 MCP 标准暴露 50+ 工具，用于房地产数据提取。
步骤 3：定义智能体与任务
这里定义智能体的人设与其具体工作。在实现 CrewAI 时，务必优先进行任务设计：建议将约 80% 的精力用于任务设计，20% 用于智能体定义。将以下智能体与任务定义加入 real_estate_agents.py：
def build_scraper_agent(mcp_tools):
return Agent(
role="Senior Real Estate Data Extractor",
goal=(
"Return a JSON object with snake_case keys containing: address, price, "
"bedrooms, bathrooms, square_feet, lot_size, year_built, property_type, "
"listing_agent, days_on_market, mls_number, description, image_urls, "
"and neighborhood for the target property listing page. Ensure strict schema validation."
),
backstory=(
"Veteran real estate data engineer with years of experience extracting "
"property information from Zillow, Realtor.com, and Redfin. Skilled in "
"Bright Data MCP, proxy rotation, CAPTCHA avoidance, and strict "
"JSON-schema validation for real estate data."
),
tools=mcp_tools,
llm=llm,
max_iter=3,
verbose=True,
)
def build_scraping_task(agent):
return Task(
description=(
"Extract property data from <https://www.zillow.com/homedetails/123-Main-St-City-State-12345/123456_zpid/> "
"and return it as structured JSON."
),
expected_output="""{
"address": "123 Main Street, City, State 12345",
"price": "$450,000",
"bedrooms": 3,
"bathrooms": 2,
"square_feet": 1850,
"lot_size": "0.25 acres",
"year_built": 1995,
"property_type": "Single Family Home",
"listing_agent": "John Doe, ABC Realty",
"days_on_market": 45,
"mls_number": "MLS123456",
"description": "Beautiful home with updated kitchen...",
"image_urls": ["<https://example.com/image1.jpg>", "<https://example.com/image2.jpg>"],
"neighborhood": "Downtown Historic District"
}""",
agent=agent,
)各参数说明如下：
- role —— 简短职位名，CrewAI 会在每个系统提示中注入 @role 参数。
- goal —— 指南针式目标；CrewAI 在每次循环后对照目标判断是否停止。
- backstory —— 领域背景知识，有助于引导智能体风格并减少幻觉。
- tools —— 注入 BaseTool 列表（例如 MCP 的 search_engine、scrape_as_markdown）。
- llm —— CrewAI 在每次“思考→计划→行动→回答”循环中使用的模型定义。
- max_iter —— 智能体内部循环的硬上限（v0.30+ 默认为 20）。
- verbose —— 将每个提示、思考、工具调用输出到控制台（用于调试）。
- description —— 每次调用都会注入的行动式指令。
- expected_output —— 有效答案的正式契约（严格 JSON，无尾随逗号）。
- agent —— 将此任务绑定到特定 Agent 实例，以供 Crew.kickoff() 使用。
步骤 4：组装 Crew 并执行
本部分将智能体与任务组装成一个 Crew 并运行工作流。将以下组装与执行脚本加入 real_estate_agents.py：
def scrape_property_data():
"""Assembles and runs the scraping crew."""
with MCPServerAdapter(server_params) as mcp_tools:
scraper_agent = build_scraper_agent(mcp_tools)
scraping_task = build_scraping_task(scraper_agent)
crew = Crew(
agents=[scraper_agent],
tasks=[scraping_task],
process=Process.sequential,
verbose=True
)
return crew.kickoff()
if __name__ == "__main__":
try:
result = scrape_property_data()
print("\\n[SUCCESS] Scraping completed!")
print("Extracted property data:")
print(result)
except Exception as e:
print(f"\\n[ERROR] Scraping failed: {str(e)}")步骤 5：运行爬取程序
在终端执行以下命令运行脚本：
python real_estate_agents.py你将看到控制台中智能体的思考过程，展示其如何规划并执行任务。
最终输出会是一个干净的 JSON 对象：
{
"address": "123 Main Street, City, State 12345",
"price": "$450,000",
"bedrooms": 3,
"bathrooms": 2,
"square_feet": 1850,
"lot_size": "0.25 acres",
"year_built": 1995,
"property_type": "Single Family Home",
"listing_agent": "John Doe, ABC Realty",
"days_on_market": 45,
"mls_number": "MLS123456",
"description": "Beautiful home with updated kitchen...",
"image_urls": ["<https://example.com/image1.jpg>", "<https://example.com/image2.jpg>"],
"neighborhood": "Downtown Historic District"
}进阶实现模式
虽然基础示例展示了核心理念，但真实生产应用需要更多考量：
市场分析与线索生成
构建市场情报需要智能体分析趋势、识别机会并生成合格线索。这些智能体协作提供全面的市场洞察并锁定潜在客户。
将以下市场分析智能体加入 real_estate_agents.py：
def build_market_analysis_agent(mcp_tools):
return Agent(
role="Real Estate Market Analyst",
goal=(
"Analyze market trends, price movements, and investment opportunities. "
"Provide actionable insights for buyers, sellers, and investors based "
"on comprehensive market data and comparable property analysis."
),
backstory=(
"Senior market analyst with expertise in real estate economics, "
"property valuation, and investment analysis. Specializes in identifying "
"market trends, pricing anomalies, and investment opportunities using "
"statistical analysis and machine learning techniques."
),
tools=mcp_tools,
llm=llm,
max_iter=4,
verbose=True,
)
def build_lead_generation_agent(mcp_tools):
return Agent(
role="Real Estate Lead Generation Specialist",
goal=(
"Identify potential buyers and sellers based on market activity, "
"property searches, and behavioral patterns. Generate qualified "
"leads with contact information and engagement strategies."
),
backstory=(
"Lead generation expert with deep knowledge of real estate marketing, "
"customer behavior analysis, and digital prospecting. Experienced in "
"identifying high-value prospects and developing targeted outreach "
"campaigns for real estate professionals."
),
tools=mcp_tools,
llm=llm,
max_iter=3,
verbose=True,
)
def analyze_market_and_generate_leads(area_zip_code, price_range):
"""Perform market analysis and generate leads for a specific area."""
with MCPServerAdapter(server_params) as mcp_tools:
market_analyst = build_market_analysis_agent(mcp_tools)
lead_generator = build_lead_generation_agent(mcp_tools)
market_task = Task(
description=(
f"Analyze the real estate market for ZIP code {area_zip_code} "
f"within price range {price_range}. Research recent sales, "
"current listings, price trends, and market conditions. "
"Identify opportunities and provide investment recommendations."
),
expected_output="""{
"market_overview": {
"avg_price": "$000,000",
"median_price": "$000,000",
"price_trend": "increasing/decreasing/stable",
"days_on_market_avg": 00,
"inventory_levels": "high/medium/low"
},
"recent_sales": [],
"active_listings": 000,
"price_per_sqft_trend": "$000",
"investment_opportunities": [],
"market_forecast": "market_prediction",
"recommendations": []
}""",
agent=market_analyst,
)
lead_task = Task(
description=(
f"Generate qualified leads for {area_zip_code} area. "
"Identify potential sellers with properties likely to be listed, "
"buyers actively searching in the area, and investors looking "
"for opportunities. Include contact strategies and timing recommendations."
),
expected_output="""{
"potential_sellers": [],
"active_buyers": [],
"investor_prospects": [],
"lead_scoring": {
"high_priority": [],
"medium_priority": [],
"low_priority": []
},
"contact_strategies": [],
"follow_up_timeline": []
}""",
agent=lead_generator,
)
crew = Crew(
agents=[market_analyst, lead_generator],
tasks=[market_task, lead_task],
process=Process.sequential,
verbose=True
)
return crew.kickoff()客户互动与沟通
高效的客户管理需要专门的智能体来处理沟通、安排日程并维护全流程关系，确保一致且专业的服务体验。
将以下客户管理智能体加入 real_estate_agents.py：
def build_client_communication_agent(mcp_tools):
return Agent(
role="Real Estate Client Relations Manager",
goal=(
"Manage client communications, schedule appointments, send follow-ups, "
"and maintain client relationships throughout the buying/selling process. "
"Provide personalized service and timely responses to client inquiries."
),
backstory=(
"Experienced client relations specialist with expertise in real estate "
"customer service, appointment scheduling, and relationship management. "
"Skilled in understanding client needs, managing expectations, and "
"maintaining long-term relationships for referrals and repeat business."
),
tools=mcp_tools,
llm=llm,
max_iter=3,
verbose=True,
)
def build_appointment_scheduler_agent(mcp_tools):
return Agent(
role="Real Estate Appointment Coordinator",
goal=(
"Schedule property viewings, client meetings, and follow-up appointments. "
"Coordinate between buyers, sellers, and agents to optimize scheduling "
"and maximize showing efficiency."
),
backstory=(
"Professional appointment coordinator with deep understanding of real "
"estate workflows, client preferences, and scheduling optimization. "
"Expert in managing complex calendars and coordinating multiple stakeholders."
),
tools=mcp_tools,
llm=llm,
max_iter=2,
verbose=True,
)
def handle_client_communication(client_inquiry, client_profile):
"""Process client inquiries and manage communications."""
with MCPServerAdapter(server_params) as mcp_tools:
communication_agent = build_client_communication_agent(mcp_tools)
scheduler_agent = build_appointment_scheduler_agent(mcp_tools)
communication_task = Task(
description=(
f"Process client inquiry: '{client_inquiry}' from client with "
f"profile: {client_profile}. Provide personalized response, "
"address their specific needs, and recommend next steps."
),
expected_output="""{
"response_message": "personalized_client_response",
"client_needs_assessment": {
"budget_range": "$000,000 - $000,000",
"preferred_locations": [],
"property_requirements": [],
"timeline": "timeframe"
},
"recommended_properties": [],
"next_steps": [],
"follow_up_schedule": "timing_recommendations"
}""",
agent=communication_agent,
)
scheduling_task = Task(
description=(
"Based on the client communication, schedule appropriate "
"follow-up appointments, property viewings, or consultation "
"meetings. Optimize scheduling for client convenience and "
"agent efficiency."
),
expected_output="""{
"scheduled_appointments": [],
"property_viewing_schedule": [],
"follow_up_reminders": [],
"calendar_integration": "scheduling_details"
}""",
agent=scheduler_agent,
)
crew = Crew(
agents=[communication_agent, scheduler_agent],
tasks=[communication_task, scheduling_task],
process=Process.sequential,
verbose=True
)
return crew.kickoff()房源挂牌与营销自动化
营销自动化需要专门智能体来撰写高质量房源、进行搜索优化，并在多个平台分发。
将以下营销智能体加入 real_estate_agents.py：
def build_listing_manager_agent(mcp_tools):
return Agent(
role="Property Listing Marketing Manager",
goal=(
"Create compelling property listings, optimize for search engines, "
"and distribute across multiple platforms. Generate marketing materials "
"and track listing performance to maximize exposure and inquiries."
),
backstory=(
"Digital marketing specialist with expertise in real estate marketing, "
"SEO optimization, and multi-platform listing management. Experienced "
"in creating high-converting property descriptions and managing "
"marketing campaigns across MLS, Zillow, Realtor.com, and social media."
),
tools=mcp_tools,
llm=llm,
max_iter=4,
verbose=True,
)搜索与发现功能
智能化的房源搜索与推荐系统可帮助客户精准找到匹配需求与偏好的房源。这些智能体提供个性化的房源发现与推荐服务。
将以下搜索与发现智能体加入 real_estate_agents.py：
def build_search_agent(mcp_tools):
return Agent(
role="Intelligent Property Search Specialist",
goal=(
"Provide intelligent property search and recommendation services. "
"Understand client preferences, search multiple databases, and "
"deliver personalized property recommendations with detailed analysis."
),
backstory=(
"Search technology expert with deep understanding of real estate "
"databases, property matching algorithms, and client preference analysis. "
"Specializes in advanced search techniques and personalized recommendation "
"systems for optimal property discovery."
),
tools=mcp_tools,
llm=llm,
max_iter=4,
verbose=True,
)
def build_recommendation_agent(mcp_tools):
return Agent(
role="Property Recommendation Engine",
goal=(
"Analyze client behavior, preferences, and market data to generate "
"personalized property recommendations. Learn from client feedback "
"and continuously improve recommendation accuracy."
),
backstory=(
"Machine learning specialist with expertise in recommendation systems, "
"behavioral analysis, and predictive modeling for real estate. "
"Experienced in developing personalized recommendation engines that "
"learn from user interactions and market trends."
),
tools=mcp_tools,
llm=llm,
max_iter=3,
verbose=True,
)
def intelligent_property_search(search_criteria, client_preferences):
"""Perform intelligent property search with personalized recommendations."""
with MCPServerAdapter(server_params) as mcp_tools:
search_agent = build_search_agent(mcp_tools)
recommendation_agent = build_recommendation_agent(mcp_tools)
search_task = Task(
description=(
f"Search for properties matching criteria: {search_criteria}. "
f"Client preferences: {client_preferences}. Use advanced search "
"techniques across multiple platforms and databases. Prioritize "
"results based on client preferences and market conditions."
),
expected_output="""{
"search_results": [],
"total_matches": 0,
"search_filters_applied": [],
"alternative_suggestions": [],
"market_insights": {
"avg_price_in_area": "$000,000",
"market_trends": "trend_analysis",
"inventory_levels": "availability_status"
}
}""",
agent=search_agent,
)
recommendation_task = Task(
description=(
"Analyze search results and client preferences to generate "
"personalized recommendations. Rank properties by relevance, "
"identify hidden gems, and suggest alternative options that "
"might meet client needs."
),
expected_output="""{
"top_recommendations": [],
"personalization_score": "0-100",
"recommendation_reasoning": [],
"alternative_options": [],
"learning_insights": {
"preference_patterns": [],
"behavior_analysis": "client_behavior_summary"
}
}""",
agent=recommendation_agent,
)
crew = Crew(
agents=[search_agent, recommendation_agent],
tasks=[search_task, recommendation_task],
process=Process.sequential,
verbose=True
)
return crew.kickoff()部署与生产设置
最后，我们创建一个综合编排系统，协调所有专门智能体。这个主系统类将作为所有房地产业务的中枢。
将以下主编排加入 real_estate_agents.py：
class RealEstateAgentSystem:
def full_property_analysis(self, property_url, client_profile=None):
with MCPServerAdapter(server_params) as mcp_tools:
research_agent = build_property_research_agent(mcp_tools)
market_analyst = build_market_analysis_agent(mcp_tools)
listing_manager = build_listing_manager_agent(mcp_tools)
research_task = build_property_research_task(research_agent, property_url)
market_task = Task(
description="Analyze market conditions for the researched property",
expected_output="Market analysis with trends and recommendations",
agent=market_analyst,
)
marketing_task = Task(
description="Create marketing strategy based on property and market analysis",
expected_output="Complete marketing campaign plan",
agent=listing_manager,
)
crew = Crew(
agents=[research_agent, market_analyst, listing_manager],
tasks=[research_task, market_task, marketing_task],
process=Process.sequential,
verbose=True
)
return crew.kickoff()
def client_service_workflow(self, client_inquiry, client_profile):
communication_result = handle_client_communication(client_inquiry, client_profile)
if "search" in client_inquiry.lower():
search_criteria = self.extract_search_criteria(client_inquiry)
search_result = intelligent_property_search(search_criteria, client_profile)
return {
"communication": communication_result,
"search_results": search_result
}
return communication_result
def extract_search_criteria(self, inquiry):
criteria = {
"price_range": "extracted_from_inquiry",
"location": "extracted_from_inquiry",
"property_type": "extracted_from_inquiry",
"bedrooms": "extracted_from_inquiry",
"bathrooms": "extracted_from_inquiry"
}
return criteria
def main():
system = RealEstateAgentSystem()
property_url = "<https://www.zillow.com/homedetails/661-Cranbrook-Rd-London-ON-N6K-1W8/2071250954_zpid/>"
try:
print("=== Starting Comprehensive Property Analysis ===")
analysis_result = system.full_property_analysis(property_url)
print("\\n=== Analysis Complete ===")
print("Extracted property data:")
print(json.dumps(analysis_result, indent=2) if isinstance(analysis_result, dict) else str(analysis_result))
client_inquiry = "I'm looking for a 3-bedroom house under $500,000 in downtown area"
client_profile = {
"name": "John Smith",
"budget": "$450,000",
"preferred_locations": ["downtown", "midtown"],
"timeline": "3 months"
}
print("\\n=== Processing Client Inquiry ===")
service_result = system.client_service_workflow(client_inquiry, client_profile)
print("\\n=== Client Service Complete ===")
print("Client service results:")
print(json.dumps(service_result, indent=2) if isinstance(service_result, dict) else str(service_result))
print("\\n=== Analyzing Market and Generating Leads ===")
market_leads = analyze_market_and_generate_leads("90210", "$500,000-$1,000,000")
print("\\n=== Market Analysis Complete ===")
print("Market analysis and leads:")
print(json.dumps(market_leads, indent=2) if isinstance(market_leads, dict) else str(market_leads))
except Exception as e:
print(f"\\n[ERROR] System execution failed: {str(e)}")
if __name__ == "__main__":
main()成本优化
Bright Data 的 MCP 按使用计费，你发出的每个请求都会产生费用。以下建议可帮助你控制成本：
- 仅请求所需的房产字段，避免抓取整站或整库。
- 启用 CrewAI 的工具级缓存，若房源数据未变化则跳过调用，节省时间与额度。
- 默认使用 Web Unlocker 区域，只有在需要渲染 JavaScript 的复杂站点时再切换到 Browser API 区域。
- 为每个智能体设置合理的最大迭代次数，避免在问题列表页上无限循环。
遵循这些实践，你的 CrewAI 智能体将保持高性价比与高可靠性，可胜任生产级房地产工作负载。
结论
在本教程中，你学习了如何使用 Bright Data 的 MCP 服务器 构建 CrewAI 房地产智能体。
我们从理解这些技术及其用法入手，随后实操构建了一个房地产智能体：搭建 Bright Data MCP 服务器、配置 LLM、创建智能体、定义任务，并组装任务团队（Crew）。
你可以轻松将这些智能体适配到其他房地产目标。例如，若要从 Realtor.com 而非 Zillow 抓取，只需调整智能体的角色、目标与背景设定，以及任务的描述与预期输出即可。
支持支付宝等多种支付方式
Arindam Majumder
AI 内容创作者
Arindam Majumder 是一名开发者推广专员、YouTube博主和技术作家，专注于将大语言模型 (LLM)、智能体工作流及 AI 内容讲解得简单易懂，拥有超过5000名关注者。
Expertise
RAG（检索增强生成）
AI 智能体
Python
