"""
Splitwise API client for expense creation.
Based on Splitwise API v3.0 patterns from Splitsage Chrome extension.
Migrated from scripts/splitwise_client.py for use in MCP server.
"""
import requests
from typing import Dict, Any, Optional
class SplitwiseError(Exception):
"""Raised when Splitwise API returns an error."""
pass
class SplitwiseClient:
"""
Splitwise API v3.0 client for creating expenses.
Uses form-encoded POST requests (not JSON) following Splitwise API v3.0 spec.
"""
API_BASE_URL = "https://secure.splitwise.com/api/v3.0"
ENDPOINT_CREATE_EXPENSE = "/create_expense"
ENDPOINT_GET_EXPENSES = "/get_expenses"
ENDPOINT_DELETE_EXPENSE = "/delete_expense/{expense_id}"
ENDPOINT_COMMENTS = "/expenses/{expense_id}/comments"
def __init__(self, config: Dict[str, str]):
"""
Initialize Splitwise client with configuration.
Args:
config: Configuration dictionary with keys:
- api_key: Splitwise API key
- group_id: Splitwise group ID
- payer_id: Primary payer's Splitwise user ID
- partner_id: Partner's Splitwise user ID
"""
self.api_key = config["api_key"]
self.group_id = config["group_id"]
self.payer_id = config["payer_id"]
self.partner_id = config["partner_id"]
self.base_url = self.API_BASE_URL
def create_expense(
self,
description: str,
amount: float,
date: str,
split_ratio: str = "50/50",
order_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Create an expense in Splitwise.
Args:
description: Expense description
amount: Total amount
date: Date in YYYY-MM-DD format
split_ratio: Split ratio - "50/50", "100/0" (100% payer), or "0/100" (100% partner)
order_id: Optional order ID (added to details field for traceability)
Returns:
Created expense dictionary from API response
Raises:
SplitwiseError: If API returns error status
"""
# Calculate split amounts based on ratio
payer_owed, partner_owed = self._calculate_split(amount, split_ratio)
# Build form-encoded data (Splitwise API v3.0 uses form encoding, not JSON)
data = {
"cost": f"{amount:.2f}",
"description": description,
"date": date,
"group_id": self.group_id,
# Payer always pays upfront (paid_share = amount)
"users__0__user_id": self.payer_id,
"users__0__paid_share": f"{amount:.2f}",
"users__0__owed_share": f"{payer_owed:.2f}",
# Partner never pays upfront
"users__1__user_id": self.partner_id,
"users__1__paid_share": "0.00",
"users__1__owed_share": f"{partner_owed:.2f}",
}
# Add order ID to details field for traceability
if order_id:
data["details"] = f"Order #{order_id}"
# Make API request with Bearer token
headers = {
"Authorization": f"Bearer {self.api_key}"
}
response = requests.post(
f"{self.base_url}{self.ENDPOINT_CREATE_EXPENSE}",
headers=headers,
data=data # Form-encoded, not JSON
)
# Check for errors
if response.status_code != 200:
raise SplitwiseError(
f"Splitwise API error {response.status_code}: {response.text}"
)
result = response.json()
# Handle different response structures
if "expenses" in result and len(result["expenses"]) > 0:
expense = result["expenses"][0]
elif "expense" in result:
expense = result["expense"]
else:
raise SplitwiseError(f"Unexpected API response structure: {result}")
return expense
def get_expenses(self, days: int = 7, limit: int = 0, dated_after: Optional[str] = None) -> list[Dict[str, Any]]:
"""
Get recent expenses from the group for duplicate detection.
Args:
days: Number of days to look back (default: 7, ignored if dated_after provided)
limit: Max expenses to return (0 = all, default: 0)
dated_after: Explicit start date in YYYY-MM-DD format (overrides days parameter)
Returns:
List of expense dictionaries
Raises:
SplitwiseError: If API returns error status
Note:
Splitwise API default limit is 20 expenses per page.
Setting limit=0 fetches ALL expenses in one call.
"""
from datetime import datetime, timedelta
# Calculate dated_after parameter
if dated_after is None:
dated_after = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
headers = {
"Authorization": f"Bearer {self.api_key}"
}
# Build URL with query parameters
# limit=0 means "fetch all" per Splitwise API docs
url = (
f"{self.base_url}{self.ENDPOINT_GET_EXPENSES}"
f"?group_id={self.group_id}"
f"&dated_after={dated_after}"
f"&limit={limit}"
)
response = requests.get(url, headers=headers)
if response.status_code != 200:
raise SplitwiseError(
f"Get expenses error {response.status_code}: {response.text}"
)
result = response.json()
expenses = result.get("expenses", [])
# Filter out soft-deleted expenses (deleted_at is not None)
# Splitwise API returns deleted expenses by default - we must filter them
active_expenses = [e for e in expenses if e.get("deleted_at") is None]
return active_expenses
def get_expenses_for_sync(self, days: int = 7, limit: int = 0, dated_after: Optional[str] = None) -> list[Dict[str, Any]]:
"""
Get expenses for cache sync, INCLUDING deleted ones.
This is used by sync_cache to properly remove soft-deleted expenses from the local cache.
For duplicate detection and display, use get_expenses() instead.
Args:
days: Number of days to look back (default: 7, ignored if dated_after provided)
limit: Max expenses to return (0 = all, default: 0)
dated_after: Explicit start date in YYYY-MM-DD format
Returns:
List of ALL expense dictionaries including soft-deleted ones
"""
from datetime import datetime, timedelta
if dated_after is None:
dated_after = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
headers = {
"Authorization": f"Bearer {self.api_key}"
}
url = (
f"{self.base_url}{self.ENDPOINT_GET_EXPENSES}"
f"?group_id={self.group_id}"
f"&dated_after={dated_after}"
f"&limit={limit}"
)
response = requests.get(url, headers=headers)
if response.status_code != 200:
raise SplitwiseError(
f"Get expenses error {response.status_code}: {response.text}"
)
result = response.json()
# Return ALL expenses including deleted for sync purposes
return result.get("expenses", [])
def delete_expense(self, expense_id: str) -> Dict[str, Any]:
"""
Delete an expense from Splitwise.
Args:
expense_id: ID of the expense to delete
Returns:
API response dictionary
Raises:
SplitwiseError: If API returns error status
"""
headers = {
"Authorization": f"Bearer {self.api_key}"
}
url = f"{self.base_url}{self.ENDPOINT_DELETE_EXPENSE.format(expense_id=expense_id)}"
response = requests.post(url, headers=headers)
if response.status_code != 200:
raise SplitwiseError(
f"Delete expense error {response.status_code}: {response.text}"
)
result = response.json()
return result
def _calculate_split(self, amount: float, split_ratio: str) -> tuple[float, float]:
"""
Calculate owed shares based on split ratio.
Ensures that the two shares always sum to the exact total
by rounding one person down and giving the other person the remainder.
Args:
amount: Total amount
split_ratio: Format "X/Y" where X is payer's percentage, Y is partner's percentage
Examples: "50/50", "75/25", "100/0", "0/100"
Returns:
Tuple of (payer_owed, partner_owed)
"""
# Parse the split ratio
try:
parts = split_ratio.split("/")
if len(parts) != 2:
raise ValueError(f"Invalid split ratio format: {split_ratio}")
payer_pct = int(parts[0])
partner_pct = int(parts[1])
if payer_pct + partner_pct != 100:
raise ValueError(f"Split ratio must sum to 100: {split_ratio}")
except ValueError as e:
raise ValueError(f"Invalid split ratio: {split_ratio}. Use format 'X/Y' where X+Y=100")
# Calculate amounts
payer_owed = round(amount * payer_pct / 100, 2)
# Give partner the remainder to ensure exact total
partner_owed = round(amount - payer_owed, 2)
return (payer_owed, partner_owed)
