MCP Cab Tools

MCP驾驶室搜索和预订服务器

一个完整的模型上下文协议（MCP）服务器，用于搜索、预订和管理出租车，具有实时位置自动完成、安全支付处理和预订管理功能。

特性

🔍 定位服务

• 实时位置搜索：由Google Places Autocomplete API提供支持
• 智能定位分辨率：具有精确坐标的自动地理编码
• 交互式消歧：多个位置匹配时的下拉选择
• 模糊匹配：路由匹配的智能回退

🚕 预定管理

• 出租车搜索：查找任意两个位置之间的可用出租车
• 保持系统：15分钟临时预订暂停
• 乘客详细信息：安全地捕获乘客信息
• 行程规划：支持单程、往返、小时租赁和机场接送

💳 支付系统

• 模拟支付网关：使用Streamlit前端进行真实的支付模拟
• 卡片验证：Luhn算法验证、到期检查、CVV验证
• 付款会话：具有过期功能的安全会话管理
• 测试卡：Visa、Mastercard、Amex和Discover的多种测试卡

📋 完成预订流程

• 驾驶员分配：从驱动程序池自动分配
• 预订确认：最终确认，包括完整的行程详情
• 状态跟踪：实时预订状态更新
• 跨流程数据共享：用于多进程架构的基于文件的存储

建筑

Google Places API集成

服务器使用两个Google Places API端点：

1. 放置自动完成API：以用户类型返回位置建议

- 端点： https://maps.googleapis.com/maps/api/place/autocomplete/json - 返回：地点建议列表 place_id, name，以及 formatted_address

1. 地点详细信息API：获取完整的位置详细信息，包括坐标

- 端点： https://maps.googleapis.com/maps/api/place/details/json - 返回：带有精确纬度/经度的完整地点信息

完成预订流程

1. Search Phase
   User Input → Autocomplete API → Location Disambiguation
        ↓
   Search Cabs → Return Available Options

2. Booking Phase
   Select Cab → Create Hold (15 min expiry)
        ↓
   Add Passenger Details → Generate Payment Link

3. Payment Phase
   Open Payment URL → Enter Card Details → Validate
        ↓
   Process Payment → Update Session Status

4. Confirmation Phase
   Verify Payment → Assign Driver → Confirm Booking
        ↓
   Return Booking Details with Driver Info

架构组件

┌─────────────────────────────────────────────────────────┐
│                    MCP Server (FastMCP)                 │
│  - Location Services (Google Places API)               │
│  - Booking Management (Hold, Passenger, Status)        │
│  - Payment Tools (Create Order, Verify, Confirm)       │
└─────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│              File-based Storage (.storage/)             │
│  - booking_holds.json                                   │
│  - payment_sessions.json                                │
│  - passenger_data.json                                  │
└─────────────────────────────────────────────────────────┘
                            ↓
┌─────────────┬─────────────────────────┬─────────────────┐
│  FastAPI    │   Streamlit Frontend    │  Mock Services  │
│  Backend    │   (Payment UI)          │  (DB, Drivers)  │
│  Port: 8000 │   Port: 8501            │                 │
└─────────────┴─────────────────────────┴─────────────────┘

设置

先决条件

• Python 3.11+
• 启用了Places API的Google云项目
• Google Places API密钥（定位服务需要）

安装

1. 克隆存储库：

cd mcp-cabs-search-book

1. 使用安装依赖项 uv:

uv sync

或者使用pip：

pip install -e .

1. 配置环境：

创建一个 .env 项目根目录中的文件：

GOOGLE_PLACES_API_KEY=your_actual_api_key_here

重要安全注意事项：

• 永远不要承诺你的 .env 文件到版本控制
• 这 .env 文件已在 .gitignore
• 永远不要公开共享您的API密钥
• 将您的API密钥仅限于Google云控制台中所需的服务

获取Google Places API密钥

1. 首选 Google 云控制台
2. 创建新项目或选择现有项目
3. 启用以下API：

- 地点API（新增） - 或者：放置API，地理编码API（如果使用遗留）

1. 首选 凭证 → 创建凭据 → API密钥
2. 将API密钥复制到 .env 文件
3. （可选）为了安全起见，将API密钥限制为仅放置API

运行完整系统

该系统需要三个单独的过程才能实现完整功能：

1.启动MCP服务器（终端1）

python main.py
# or
python src/mcp-cab-server/server.py

2.启动支付后端（终端2）

uv run uvicorn src.mcp-cab-server.payment_backend:app --reload --port 8000

3.启动支付前端（终端3）

uv run streamlit run src/mcp-cab-server/payment_frontend.py --server.port 8501

注： 对于基本的驾驶室搜索功能，只需要MCP服务器（步骤1）。完整的预订流程需要支付组件（步骤2-3）。

用法

完整预订流程示例

第一步：寻找出租车

Search_cabs(
    pickup="Delhi Airport",
    drop="Connaught Place",
    trip_type="one way",
    departure_date="2024-02-20"
)

步骤2：创建预订保留

hold_cab_booking(
    cab_id="DEL_IGI_CP_2",
    pickup="Indira Gandhi International Airport",
    drop="Connaught Place",
    departure_date="2024-02-20"
)
# Returns: hold_id (valid for 15 minutes)

步骤3：添加乘客详细信息

add_passenger_details(
    hold_id="HOLD_1001",
    passenger_name="John Doe",
    passenger_phone="+919876543210",
    passenger_email="john@example.com"
)

步骤4：创建付款单

create_payment_order(hold_id="HOLD_1001")
# Returns: payment_url (opens in browser)

步骤5：完成付款

• 在浏览器中打开付款URL
• 输入卡详细信息（使用提供的测试卡）
• 提交付款

步骤6：验证付款

verify_mock_payment(session_id="PAY_5001")
# Returns: payment status

步骤7：确认预订

confirm_booking(hold_id="HOLD_1001")
# Returns: booking_id, driver details, confirmation

测试卡

在支付前端使用这些卡：

注： 所有付款都是模拟交易。没有真正的钱被处理。

测试系统

快速支付系统测试

python test_payment_system.py

这验证了：

• 所有导入和依赖项
• 卡片验证（Luhn算法）
• 驱动程序池可用性
• 模块完整性

项目结构

mcp-cabs-search-book/
├── .env                              # Environment variables (not in repo)
├── .gitignore                        # Git ignore patterns
├── pyproject.toml                    # Project dependencies
├── README.md                         # This file
├── main.py                           # MCP server entry point
├── test_payment_system.py            # Payment system tests
└── src/
    ├── .storage/                     # File-based data storage (gitignored)
    │   ├── booking_holds.json        # Active booking holds
    │   ├── payment_sessions.json     # Payment session data
    │   └── passenger_data.json       # Passenger information
    └── mcp-cab-server/
        ├── server.py                 # Main MCP server with tools
        ├── payment_backend.py        # FastAPI payment backend
        ├── payment_frontend.py       # Streamlit payment UI
        ├── models/
        │   └── models.py             # All Pydantic models
        └── services/
            ├── geocoding.py          # Google Places API integration
            ├── helper.py             # Cab search & booking logic
            ├── mock_db.py            # Mock database with routes
            ├── payment.py            # Payment service layer
            ├── card_validator.py     # Card validation (Luhn, etc.)
            └── storage.py            # File-based storage utilities

API 参考

MCP工具

1. Search_cabs

搜索两个地点之间的可用出租车。

输入：

{
  "pickup": str,              # Pickup location
  "drop": str,                # Drop location  
  "trip_type": TripType,      # Trip type
  "departure_date": date      # Journey date
}

输出：

{
  "cabs": [
    {"cab_id": str, "cab_type": str, "price": int}
  ]
}

2. hold_cab_booking

在出租车上临时停留15分钟。

输入：

{
  "cab_id": str,              # From search results
  "pickup": str,              # Pickup location
  "drop": str,                # Drop location
  "departure_date": date      # Journey date
}

输出：

{
  "hold_id": str,             # Unique hold identifier
  "expires_at": str,          # ISO timestamp
  "cab_details": dict,        # Cab information
  "price": int                # Total price
}

3. add_passenger_details

将乘客信息添加到预订保留中。

输入：

{
  "hold_id": str,             # From hold_cab_booking
  "passenger_name": str,      # Full name
  "passenger_phone": str,     # Contact number
  "passenger_email": str,     # Email (optional)
  "special_requests": str     # Special requirements (optional)
}

4. create_payment_order

为预订生成付款链接。

输入：

{
  "hold_id": str              # Hold with passenger details
}

输出：

{
  "session_id": str,          # Payment session ID
  "payment_url": str,         # URL to open in browser
  "amount": float,            # Payment amount
  "expires_at": str           # Session expiry
}

5. verify_mock_payment

检查付款完成状态。

输入：

{
  "session_id": str           # From create_payment_order
}

输出：

{
  "status": str,              # "pending", "completed", "failed"
  "amount": float,
  "card_last4": str           # Last 4 digits (if completed)
}

6. confirm_booking

付款后完成预订，指定司机。

输入：

{
  "hold_id": str              # Hold with completed payment
}

输出：

{
  "booking_id": str,          # Final booking reference
  "driver": {
    "name": str,
    "phone": str,
    "vehicle_number": str,
    "vehicle_model": str,
    "rating": float
  },
  "booking_summary": dict     # Complete trip details
}

配置

环境变量

数据存储

该系统使用基于文件的JSON存储 src/.storage/:

• booking_holds.json：有效预订保持及其状态
• payment_sessions.json：付款会话跟踪
• passenger_data.json：每次等待的乘客信息

注： 这 .storage/ 目录是gitignored的，并在运行时自动创建。

API成本（近似值）

• 地点自动补全每1000个请求约2.83美元
• 地点详细信息：每1000个请求约17美元
• 每次出租车搜索总计：~0.04美元（2次自动完成+2次详细信息调用）

系统端口

发展

添加模拟路线

编辑 src/mcp-cab-server/services/mock_db.py:

MOCK_CAB_DB = {
    ("pickup_location_name", "drop_location_name"): [
        {"cab_id": "ROUTE_ID", "cab_type": "sedan", "price": 500},
        {"cab_id": "ROUTE_ID_2", "cab_type": "suv", "price": 700},
    ],
}

注： 位置名称应匹配Google Places API输出（小写）。

添加模拟驱动程序

编辑 MOCK_DRIVERS 列入 mock_db.py:

MOCK_DRIVERS = [
    {
        "name": "Driver Name",
        "phone": "+91-9876543210",
        "vehicle_number": "DL-01-AB-1234",
        "vehicle_model": "Honda City",
        "rating": 4.8
    }
]

扩展服务器

1. 添加MCP工具：用以下方式装饰功能 @mcp.tool() 在 server.py
2. 添加模型：在中定义Pydantic模型 models/models.py
3. 添加服务：在中创建服务模块 services/
4. 添加API端点：在中添加FastAPI路由 payment_backend.py

代码结构

• 清洁代码：为清楚起见，删除了评论，只保留了基本的内联评论
• 类型安全：全程完整的Pydantic模型验证
• 错误处理：带有日志记录的全面错误消息
• 存储：基于线程安全的文件存储，具有自动序列化功能

故障排除

定位服务

“找不到GOOGLE_PLACES_API_KEY”

• 确保 .env 文件存在于项目根目录中
• 验证格式： GOOGLE_PLACES_API_KEY=your_key （无报价）
• 在Google云控制台中检查API密钥是否有效
• 验证Places API是否已启用以及帐单是否处于活动状态

“未找到位置”

• 检查API密钥权限和配额
• 验证Google Cloud项目是否已启用计费
• 尝试使用不太具体的搜索词
• 检查服务器日志以了解详细错误

支付系统

支付前端无法启动

• 确保安装了所有依赖项： uv sync
• 检查端口8501是否可用
• 验证支付后端是否在端口8000上运行

卡验证失败

• 使用提供的测试卡（见测试卡部分）
• 确保到期日为未来（MM/YY格式）
• CVV:3位数字（美国运通为4位）
• 检查浏览器控制台以了解详细错误

“保留已过期”

• 保留在15分钟后过期
• 创建新的驾驶室等待并快速重试
• 检查系统时间是否同步

存储问题

.storage文件上的“权限被拒绝”

• 确保项目目录的写入权限
• 检查 .storage/ 目录不是只读的
• 验证没有其他进程锁定文件

一般问题

在IDE中导入警告

• 此项目结构正常
• 代码在运行时正常工作
• PYTHONPATH由入口点正确设置

多流程协调

• 首先启动MCP服务器
• 然后启动支付后端
• 最终启动支付前端
• 所有三个必须同时运行才能实现完整功能

安全说明

什么是安全的

• ✅ 所有付款均为模拟付款
• ✅ 没有实际的金融交易发生
• ✅ 测试卡号为公共领域
• ✅ 没有永久存储敏感数据
• ✅ .env 文件被忽略

重要提醒

• 🔒 永不承诺 .env 文件到版本控制
• 🔒 永远不要公开分享你的Google API密钥
• 🔒 在Google云控制台中使用API密钥限制
• 🔒 保持 .storage/ 目录gitignored
• 🔒 这只是一个演示系统

测试与验证

卡验证功能

• Luhn算法：验证卡校验和
• 卡片类型检测：识别Visa、万事达卡、美国运通、Discover
• 到期验证：检查过期卡
• CVV长度：根据卡类型进行验证（3或4位数字）
• 持卡人姓名：基本格式验证

预订暂停系统

• 15分钟到期：自动持有到期
• 状态跟踪：举行→ passenger_已添加→ 付款→ 付款_成功→ 确认
• 清理线程：旧过期保留的后台清理
• 线程安全存储：并发访问保护

生产考量

⚠️ 这是一个示范项目。用于生产用途：

1. 用真实网关（Stripe、Razorpay等）取代模拟支付
2. 使用合适的数据库（PostgreSQL、MongoDB）
3. 添加身份验证和授权
4. 实现速率限制和API安全
5. 添加全面的日志记录和监控
6. 使用特定于环境的配置
7. 添加单元和集成测试
8. 实施正确的错误处理和重试逻辑
9. 为敏感信息添加数据加密
10. 将消息队列用于后台任务

许可证

该项目用于示范和教育目的。

贡献

请随意分叉和扩展此演示项目！

支持

对于问题：

1. 检查服务器日志以获取详细的错误消息
2. 跑 python test_payment_system.py 验证设置
3. 验证Google API密钥和帐单状态
4. 确保所有三个进程都在运行
5. 检查浏览器控制台是否存在前端错误