这是一个全面的用于房地产数据管理的模型上下文协议(MCP)服务器。该服务器为房产列表、经纪人管理、市场分析、客户关系和区域情报等提供工具、资源和提示模板。
克隆仓库:
git clone https://github.com/agentic-ops/real-estate-mcp.git
cd real-estate-mcp
安装依赖:
pip install -r requirements.txt
运行服务器:
python main.py
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:客户房产匹配结果。服务器采用模块化、组件化架构,以确保可维护性和可扩展性:
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 检查器工具:
npx @modelcontextprotocol/inspector
这将启动 MCP 检查器界面,允许你:
服务器使用 服务器发送事件(SSE) 传输,使其与以下环境兼容:
http://127.0.0.1:8000/sse(用于建立 SSE 连接)http://127.0.0.1:8000/messages/(用于发送 MCP 消息)// 建立 SSE 连接
const eventSource = new EventSource('http://127.0.0.1:8000/sse');
eventSource.onmessage = function(event) {
const mcpMessage = JSON.parse(event.data);
// 处理 MCP 协议消息
};
// 发送 MCP 消息
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:搜索和筛选工具。main.py - 服务器入口点每个工具模块遵循一致的模式:
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} 的详细分析说明...
"""
服务器处理全面的房地产数据:
# 建立 SSE 连接(监听服务器消息)
curl -N http://127.0.0.1:8000/sse
# 发送 MCP 消息(在另一个终端中)
# 搜索房产
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/)# 安装测试依赖
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 中可重用的测试组件。htmlcov/index.html 中的可视化覆盖率报告。测试套件验证:
对于 CI/CD 管道,可使用:
# 基本测试运行
pytest tests/ --tb=short
# 生成用于 CI 报告的覆盖率
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_* 方法。本项目采用 MIT 许可证。
要深入了解此 MCP 服务器的架构、设计原则和实际应用,请阅读详细的博客文章:
[🔌 MCP 服务器 - 模型上下文协议实现](https://edwin.genego.io/ai/mcp - servers)
博客文章涵盖:
基于模型上下文协议(MCP)构建,实现无缝 AI 集成