🚀 房地产MCP服务器
这是一个全面的用于房地产数据管理的模型上下文协议(MCP)服务器。该服务器为房产列表、经纪人管理、市场分析、客户关系和区域情报等提供工具、资源和提示模板。
🚀 快速开始
-
克隆仓库:
git clone https://github.com/agentic-ops/real-estate-mcp.git
cd real-estate-mcp
-
安装依赖:
pip install -r requirements.txt
-
运行服务器:
python main.py
✨ 主要特性
MCP功能
- 30 + 工具:涵盖全面的房地产运营功能。
- 10 种资源:包括 5 个静态资源和 5 个动态资源模板。
- 11 个提示模板:用户可控制的分析模板,涵盖 4 个类别。
- SSE 传输:支持与 Web 兼容的服务器发送事件(Server - Sent Events)端点。
工具类别
🏠 房产管理(7 个工具)
- 可通过多个条件搜索和筛选房产。
- 获取房产详细信息和全面洞察。
- 提供基于区域和经纪人的房产列表。
- 进行市场背景和可比分析。
👥 经纪人运营(6 个工具)
- 管理经纪人资料和专业领域。
- 提供绩效仪表盘和指标。
- 管理客户和房产组合。
- 跟踪销售情况并进行分析。
📊 市场分析(7 个工具)
- 提供市场概述和价格分析。
- 分析特定区域的市场表现。
- 进行投资机会分析。
- 开展区域对比分析。
- 跟踪交易情况。
🤝 客户管理(3 个工具)
- 管理客户资料和偏好。
- 运用房产匹配算法。
- 根据预算和标准提供推荐。
🏘️ 区域情报(9 个工具)
- 提供全面的区域报告。
- 涵盖区域设施和人口统计信息。
- 包含学校、公园、购物和医疗保健数据。
- 提供城市概述和区域比较。
⚙️ 系统管理(2 个工具)
资源
静态资源
realestate://all - properties
:完整的房产列表。
realestate://all - agents
:经纪人目录。
realestate://market - overview
:当前市场趋势。
realestate://all - areas
:区域信息。
realestate://amenities
:完整的设施数据库。
动态资源模板
realestate://properties/area/{area}
:特定区域的房产。
realestate://agent/{agent_id}/dashboard
:经纪人绩效仪表盘。
realestate://market/area/{area}
:特定区域的市场分析。
realestate://property/{property_id}/insights
:房产洞察。
realestate://client/{client_id}/matches
:客户房产匹配结果。
提示模板(共 11 个)
房产提示模板(2 个)
- 房产分析:全面的房产评估和洞察。
- 房产比较:房产的并排比较分析。
客户提示模板(3 个)
- 客户匹配:个性化的房产推荐。
- 客户咨询:结构化的咨询框架。
- 客户反馈分析:优化搜索策略。
市场提示模板(3 个)
- 市场报告:全面的区域市场分析。
- 投资分析:投资回报率和机会评估。
- 比较市场分析:多区域比较。
经纪人提示模板(3 个)
- 经纪人绩效:绩效仪表盘和分析。
- 经纪人营销策略:业务发展和营销。
- 经纪人培训发展:技能提升和培训计划。
🏗️ 架构
服务器采用模块化、组件化架构,以确保可维护性和可扩展性:
real - estate - mcp/
├── main.py # 主服务器入口点
├── utils.py # 核心数据管理工具
├── tools/ # MCP 工具(按类别组织)
│ ├── property_tools.py # 房产搜索、筛选、洞察
│ ├── agent_tools.py # 经纪人资料、绩效、仪表盘
│ ├── market_tools.py # 市场分析和趋势
│ ├── client_tools.py # 客户管理和匹配
│ ├── area_tools.py # 区域情报和设施
│ └── system_tools.py # 数据管理和系统工具
├── resources/ # MCP 资源(按领域组织)
│ ├── property_resources.py # 与房产相关的资源
│ ├── agent_resources.py # 与经纪人相关的资源
│ ├── market_resources.py # 市场分析资源
│ ├── client_resources.py # 客户管理资源
│ └── location_resources.py # 区域和设施资源
├── prompts/ # MCP 提示模板(用户可控制的模板)
│ ├── __init__.py # 中央提示模板注册
│ ├── property_prompts.py # 房产分析和比较提示模板
│ ├── client_prompts.py # 客户匹配和咨询提示模板
│ ├── market_prompts.py # 市场分析和投资提示模板
│ └── agent_prompts.py # 经纪人绩效和发展提示模板
└── data/ # 房地产数据文件
├── properties/
├── agents/
├── clients/
├── market/
├── transactions/
├── areas/
└── amenities/
📦 安装指南
-
克隆仓库:
git clone https://github.com/agentic - ops/real - estate - mcp.git
cd real - estate - mcp
-
安装依赖:
pip install -r requirements.txt
-
运行服务器:
python main.py
🔍 MCP 检查器
要检查和调试 MCP 服务器,可使用 MCP 检查器工具:
npx @modelcontextprotocol/inspector
这将启动 MCP 检查器界面,允许你:
- 实时监控 MCP 消息。
- 调试工具和资源调用。
- 检查服务器响应。
- 测试服务器功能。
🌐 服务器传输
服务器使用 服务器发送事件(SSE) 传输,使其与以下环境兼容:
- 网页浏览器和 HTTP 客户端。
- 传统 MCP 客户端。
- 自定义集成。
连接详情
- SSE 端点:
http://127.0.0.1:8000/sse
(用于建立 SSE 连接)
- 消息端点:
http://127.0.0.1:8000/messages/
(用于发送 MCP 消息)
- 传输方式:SSE(服务器发送事件)
- 协议:MCP(模型上下文协议)
网页客户端示例
const eventSource = new EventSource('http://127.0.0.1:8000/sse');
eventSource.onmessage = function(event) {
const mcpMessage = JSON.parse(event.data);
};
async function sendMCPMessage(message) {
const response = await fetch('http://127.0.0.1:8000/messages/', {
method: 'POST',
headers: {
'Content - Type': 'application/json',
},
body: JSON.stringify(message)
});
return response.json();
}
🔧 技术细节
核心组件
utils.py
- 数据管理
RealEstateDataManager
:中央数据访问类。
PropertyFilter
:搜索和筛选工具。
- JSON 数据加载和缓存。
- 交叉引用和关系映射。
main.py
- 服务器入口点
- FastMCP 服务器初始化。
- 组件注册编排。
- SSE 传输配置。
- 启动日志记录和诊断。
工具模块
每个工具模块遵循一致的模式:
def register_[category]_tools(mcp: FastMCP):
"""向 MCP 服务器注册所有 [类别] 工具"""
@mcp.tool()
def tool_function(parameters) -> str:
"""工具描述"""
return json.dumps(result, indent = 2)
资源模块
资源按领域组织,以提高可维护性:
房产资源 (property_resources.py
)
- 房产列表和搜索结果。
- 房产洞察和市场背景。
- 基于区域的房产筛选。
经纪人资源 (agent_resources.py
)
市场资源 (market_resources.py
)
客户资源 (client_resources.py
)
位置资源 (location_resources.py
)
每个模块遵循一致的模式:
def register_[domain]_resources(mcp: FastMCP):
"""向 MCP 服务器注册所有 [领域] 资源"""
@mcp.resource("realestate://resource - name")
def resource_function() -> str:
"""资源描述"""
return json.dumps(data, indent = 2)
提示模板
提示模板指导 AI 分析:
@mcp.prompt()
def analysis_prompt(param: str = "default") -> str:
"""分析提示模板描述"""
return f"""
针对 {param} 的详细分析说明...
"""
📊 数据结构
服务器处理全面的房地产数据:
- 5 种房产类型:维多利亚式住宅、现代住宅、豪华住宅、联排别墅。
- 3 名经纪人:专业房地产经纪人。
- 6 名客户:有不同偏好的买家、卖家和投资者。
- 多笔销售记录:近期交易历史。
- 5 个区域:河滨市中心、伍德克里斯特、峡谷顶、阿灵顿高地、拉塞拉。
- 各类设施:学校、公园、购物中心、医疗保健设施。
💻 使用示例
基础用法
curl -N http://127.0.0.1:8000/sse
curl -X POST http://127.0.0.1:8000/messages/ \
-H "Content-Type: application/json" \
-d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "search_properties", "arguments": {"query": "Victorian"}}}'
curl -X POST http://127.0.0.1:8000/messages/ \
-H "Content-Type: application/json" \
-d '{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "filter_properties", "arguments": {"min_price": 500000, "max_price": 1000000}}}'
curl -X POST http://127.0.0.1:8000/messages/ \
-H "Content-Type: application/json" \
-d '{"jsonrpc": "2.0", "id": 3, "method": "resources/read", "params": {"uri": "realestate://market-overview"}}'
curl -X POST http://127.0.0.1:8000/messages/ \
-H "Content-Type: application/json" \
-d '{"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "match_client_preferences", "arguments": {"client_id": "CLI001"}}}'
🧪 测试
测试结构
tests/
├── conftest.py # Pytest 配置和共享测试夹具
├── unit/ # 核心组件的单元测试
│ ├── test_utils.py # RealEstateDataManager 和 PropertyFilter 测试
│ └── test_*.py # 其他单元测试
├── integration/ # MCP 组件的集成测试
│ ├── test_property_tools.py # 房产工具集成测试
│ ├── test_all_tools.py # 所有其他工具类别的测试
│ ├── test_resources.py # 静态和模板资源测试
│ └── test_prompts.py # 提示模板测试
└── __init__.py
测试类别
单元测试 (tests/unit/
)
- 数据管理器测试:
RealEstateDataManager
的核心功能。
- 筛选测试:房产筛选逻辑和边界情况。
- 实用函数测试:辅助函数和数据验证。
集成测试 (tests/integration/
)
- 房产工具:搜索、筛选、洞察和基于区域的查询。
- 经纪人工具:资料管理、绩效仪表盘。
- 市场工具:市场分析和趋势计算。
- 客户工具:客户匹配和偏好算法。
- 区域工具:区域情报和设施数据。
- 系统工具:数据刷新和系统统计。
- 资源:静态资源和动态模板。
- 提示模板:模板生成和参数处理(4 个类别共 11 个提示模板)。
运行测试
前提条件
pip install -r requirements.txt
快速测试命令
pytest
pytest --cov=. --cov-report=html
pytest tests/unit/
pytest tests/integration/
pytest tests/integration/test_property_tools.py
使用测试运行脚本
python run_tests.py
python run_tests.py unit
python run_tests.py integration
python run_tests.py property
python run_tests.py resources
python run_tests.py all -v -c
测试特性
测试夹具和测试数据
- 隔离测试环境:每个测试使用临时数据目录。
- 模拟数据:所有测试用例使用一致的测试数据。
- 共享测试夹具:
conftest.py
中可重用的测试组件。
- 数据管理器模拟:无文件系统依赖的隔离测试。
覆盖率和报告
- 代码覆盖率:使用 pytest - cov 生成全面的覆盖率报告。
- HTML 报告:
htmlcov/index.html
中的可视化覆盖率报告。
- 未覆盖代码行:识别未覆盖的代码路径。
- 分支覆盖率:逻辑分支测试。
测试配置
- pytest.ini:集中式测试配置。
- 自动发现:按命名约定自动发现测试。
- 并行执行:支持并行测试执行。
- 过滤:过滤警告以获得清晰的测试输出。
测试数据验证
测试套件验证:
- ✅ 所有 30 + 个工具在模拟和真实数据下都能正常工作。
- ✅ 房产筛选逻辑能处理边界情况。
- ✅ 搜索功能不区分大小写且全面。
- ✅ 经纪人绩效计算准确。
- ✅ 市场分析工具能正确处理数据。
- ✅ 客户匹配算法按预期工作。
- ✅ 区域情报能正确汇总数据。
- ✅ 资源端点返回有效的 JSON。
- ✅ 提示模板能生成正确的指令。
- ✅ 处理缺失或无效数据的错误处理机制。
- ✅ 数据刷新和缓存机制。
- ✅ 系统统计和摘要。
持续集成
对于 CI/CD 管道,可使用:
pytest tests/ --tb=short
pytest tests/ --cov=. --cov-report=xml --cov-report=term-missing
pytest tests/unit/ --tb=short
pytest tests/integration/ --tb=short
编写新测试
添加新功能时:
- 单元测试:添加到
tests/unit/
以测试核心逻辑。
- 集成测试:添加到相应的
tests/integration/test_*.py
文件。
- 使用测试夹具:利用
conftest.py
中现有的测试夹具。
- 模拟外部依赖:使用
unittest.mock
进行隔离测试。
- 测试边界情况:包括边界条件和错误场景。
- 遵循命名约定:使用
test_*.py
文件、Test*
类和 test_*
方法。
🔄 SSE 传输的优势
- 与 Web 兼容:可直接集成到浏览器。
- 实时性:服务器发送事件实现实时更新。
- 符合 HTTP 标准:可与标准 HTTP 工具配合使用。
- 对防火墙友好:使用标准 HTTP 端口。
- 可扩展:支持多个并发连接。
📄 许可证
本项目采用 MIT 许可证。
🤝 贡献指南
- 分叉仓库。
- 创建功能分支。
- 按照既定模式添加组件。
- 进行全面测试。
- 提交拉取请求。
📖 扩展阅读
要深入了解此 MCP 服务器的架构、设计原则和实际应用,请阅读详细的博客文章:
[🔌 MCP 服务器 - 模型上下文协议实现](https://edwin.genego.io/ai/mcp - servers)
博客文章涵盖:
- 理解 MCP 服务器及其商业影响。
- 深入剖析架构并提供代码示例。
- 解释 MCP 工具、提示模板和资源。
- 实际使用场景和实现模式。
- 安全考虑和最佳实践。
- MCP 技术的未来影响。
基于模型上下文协议(MCP)构建,实现无缝 AI 集成