#!/usr/bin/env python3
"""
Splitwise MCP Server - Main server implementation.
Provides 7 MCP tools for expense management:
- splitwise_check_duplicate: Check if expense is duplicate
- splitwise_reserve_expense: Atomically reserve expense slot
- splitwise_create_expense: Create expense and confirm reservation
- splitwise_sync_cache: Sync cache with Splitwise API
- splitwise_get_expenses: Query expenses from cache
- splitwise_delete_expense: Delete expense from Splitwise and cache
- splitwise_find_by_order_id: Find expense by order ID in notes (prevents duplicates)
"""
import asyncio
import json
import logging
import os
from pathlib import Path
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .cache import CacheManager
from .client import SplitwiseClient
from .config import ConfigurationError, load_config
from .tools import (
handle_check_duplicate,
handle_reserve_expense,
handle_create_expense,
handle_sync_cache,
handle_get_expenses,
handle_delete_expense,
handle_find_by_order_id,
)
# Configure logging
log_level = os.environ.get("LOG_LEVEL", "INFO")
logging.basicConfig(level=getattr(logging, log_level.upper(), logging.INFO))
logger = logging.getLogger(__name__)
# Server instance
server = Server("splitwise-mcp-server")
# Global state (initialized in main)
cache_manager: CacheManager | None = None
splitwise_client: SplitwiseClient | None = None
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List all available Splitwise MCP tools."""
return [
Tool(
name="splitwise_check_duplicate",
description="Check if an expense would be a duplicate based on description, amount, and date similarity",
inputSchema={
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "Expense description to check"
},
"amount": {
"type": "number",
"description": "Expense amount in dollars"
},
"date": {
"type": "string",
"description": "Expense date in YYYY-MM-DD format"
},
"group_id": {
"type": "string",
"description": "Splitwise group ID"
}
},
"required": ["description", "amount", "date", "group_id"]
}
),
Tool(
name="splitwise_reserve_expense",
description="Atomically reserve an expense slot to prevent race conditions. Returns reservation_id for use in create_expense.",
inputSchema={
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "Expense description"
},
"amount": {
"type": "number",
"description": "Expense amount in dollars"
},
"date": {
"type": "string",
"description": "Expense date in YYYY-MM-DD format"
},
"group_id": {
"type": "string",
"description": "Splitwise group ID"
},
"order_id": {
"type": "string",
"description": "Order ID for idempotent duplicate detection (e.g., Amazon order ID, Instacart order ID)"
}
},
"required": ["description", "amount", "date", "group_id", "order_id"]
}
),
Tool(
name="splitwise_create_expense",
description="Create expense in Splitwise and confirm reservation. Supports custom split ratios.",
inputSchema={
"type": "object",
"properties": {
"reservation_id": {
"type": "string",
"description": "Reservation ID from reserve_expense"
},
"description": {
"type": "string",
"description": "Expense description"
},
"amount": {
"type": "number",
"description": "Expense amount in dollars"
},
"date": {
"type": "string",
"description": "Expense date in YYYY-MM-DD format"
},
"group_id": {
"type": "string",
"description": "Splitwise group ID"
},
"payer_id": {
"type": "string",
"description": "Primary payer's Splitwise user ID"
},
"partner_id": {
"type": "string",
"description": "Partner's Splitwise user ID"
},
"order_id": {
"type": "string",
"description": "Order ID for traceability (optional)"
},
"split_ratio": {
"type": "string",
"description": "Split ratio as 'X/Y' where X is payer's percentage, Y is partner's (e.g., '50/50', '75/25', '100/0'). Default: '50/50'"
}
},
"required": ["reservation_id", "description", "amount", "date", "group_id", "payer_id", "partner_id"]
}
),
Tool(
name="splitwise_sync_cache",
description="Sync local cache with Splitwise API. Fetches recent expenses and updates cache.",
inputSchema={
"type": "object",
"properties": {
"group_id": {
"type": "string",
"description": "Splitwise group ID"
},
"since_date": {
"type": "string",
"description": "Start date for sync in YYYY-MM-DD format (optional, defaults to 7 days ago)"
},
"limit": {
"type": "integer",
"description": "Maximum number of expenses to fetch (default 100)"
}
},
"required": ["group_id"]
}
),
Tool(
name="splitwise_get_expenses",
description="Query expenses from cache (fast, no API call). Returns expenses for date range.",
inputSchema={
"type": "object",
"properties": {
"group_id": {
"type": "string",
"description": "Splitwise group ID"
},
"start_date": {
"type": "string",
"description": "Start date in YYYY-MM-DD format"
},
"end_date": {
"type": "string",
"description": "End date in YYYY-MM-DD format (optional, defaults to today)"
}
},
"required": ["group_id", "start_date"]
}
),
Tool(
name="splitwise_delete_expense",
description="Delete expense from Splitwise and remove from cache.",
inputSchema={
"type": "object",
"properties": {
"expense_id": {
"type": "string",
"description": "Splitwise expense ID to delete"
},
"group_id": {
"type": "string",
"description": "Splitwise group ID"
}
},
"required": ["expense_id", "group_id"]
}
),
Tool(
name="splitwise_find_by_order_id",
description="Find expense by Amazon order ID in notes field. Most reliable duplicate detection method.",
inputSchema={
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Amazon order ID to search for (e.g., '113-6001239-5817844')"
},
"group_id": {
"type": "string",
"description": "Splitwise group ID (optional, for future filtering)"
}
},
"required": ["order_id"]
}
),
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool calls by routing to appropriate handler."""
if cache_manager is None or splitwise_client is None:
return [TextContent(
type="text",
text=json.dumps({
"error": "Server not initialized",
"status": "error"
})
)]
try:
if name == "splitwise_check_duplicate":
result = await handle_check_duplicate(cache_manager, arguments)
elif name == "splitwise_reserve_expense":
result = await handle_reserve_expense(cache_manager, arguments)
elif name == "splitwise_create_expense":
result = await handle_create_expense(cache_manager, splitwise_client, arguments)
elif name == "splitwise_sync_cache":
result = await handle_sync_cache(cache_manager, splitwise_client, arguments)
elif name == "splitwise_get_expenses":
result = await handle_get_expenses(cache_manager, arguments)
elif name == "splitwise_delete_expense":
result = await handle_delete_expense(cache_manager, splitwise_client, arguments)
elif name == "splitwise_find_by_order_id":
result = await handle_find_by_order_id(cache_manager, arguments)
else:
result = {
"error": f"Unknown tool: {name}",
"status": "error"
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error handling {name}: {e}", exc_info=True)
return [TextContent(
type="text",
text=json.dumps({
"error": str(e),
"status": "error",
"tool": name
})
)]
async def main():
"""Run the MCP server."""
global cache_manager, splitwise_client
try:
# Load configuration using flexible loader
config = load_config()
logger.info("Configuration loaded successfully")
logger.info(f"Group ID: {config.get('group_id')}")
# Determine cache database path
# Use SPLITWISE_CACHE_PATH env var, or default to ./splitwise_cache.db
cache_db_path = os.environ.get("SPLITWISE_CACHE_PATH", "./splitwise_cache.db")
logger.info(f"Using cache database: {cache_db_path}")
# Initialize components
cache_manager = CacheManager(cache_db_path)
splitwise_client = SplitwiseClient(config)
logger.info("Splitwise MCP Server initialized")
# Run server
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
except ConfigurationError as e:
logger.error(f"Configuration error: {e}")
raise
except Exception as e:
logger.error(f"Failed to start server: {e}", exc_info=True)
raise
if __name__ == "__main__":
asyncio.run(main())
