We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/MatiasL13/lunchmoney-mcp'
If you have feedback or need assistance with the MCP directory API, please join our Discord server
"""Lunch Money MCP Server"""
import os
import asyncio
import json
from typing import Any, Dict, List, Optional, Union
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel,
)
import httpx
from pydantic import BaseModel, Field
from dotenv import load_dotenv
load_dotenv()
# Initialize the MCP server
server = Server("lunchmoney-mcp")
# Configuration
API_BASE_URL = "https://dev.lunchmoney.app"
ACCESS_TOKEN = os.getenv("LUNCHMONEY_ACCESS_TOKEN")
def validate_access_token():
"""Validate that the ACCESS_TOKEN is available"""
if not ACCESS_TOKEN:
import sys
print("Error: LUNCHMONEY_ACCESS_TOKEN environment variable is required", file=sys.stderr)
print("Get your token from: https://my.lunchmoney.app/developers", file=sys.stderr)
sys.exit(1)
# HTTP client configuration - removed get_http_client() function as it caused async context manager issues
# Pydantic models for request/response validation
class UserResponse(BaseModel):
user_id: int
user_name: str
user_email: str
account_id: int
budget_name: str
api_key_label: Optional[str] = None
class CategoryResponse(BaseModel):
id: int
name: str
description: Optional[str] = None
is_income: bool
exclude_from_budget: bool
exclude_from_totals: bool
archived: bool
archived_on: Optional[str] = None
updated_at: str
created_at: str
is_group: bool
group_id: Optional[int] = None
order: Optional[int] = None
class TransactionResponse(BaseModel):
id: int
date: str
amount: str
currency: str
to_base: float
payee: str
category_id: Optional[int] = None
category_name: Optional[str] = None
category_group_id: Optional[int] = None
category_group_name: Optional[str] = None
is_income: bool
exclude_from_budget: bool
exclude_from_totals: bool
created_at: str
updated_at: str
status: str
is_pending: bool
date_created: str
group_id: Optional[int] = None
parent_id: Optional[int] = None
is_group: bool
group_description: Optional[str] = None
tags: Optional[List[Dict[str, Any]]] = None
external_id: Optional[str] = None
# MCP Tools Implementation
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List all available Lunch Money API tools"""
return [
# User tools
Tool(
name="get_user",
description="Get user information including account details and preferences",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
# Category tools
Tool(
name="get_all_categories",
description="Get all categories with optional format (nested or flattened)",
inputSchema={
"type": "object",
"properties": {
"format": {
"type": "string",
"enum": ["nested", "flattened"],
"description": "Format for category response"
}
},
"required": []
}
),
Tool(
name="get_single_category",
description="Get details of a specific category by ID",
inputSchema={
"type": "object",
"properties": {
"category_id": {
"type": "integer",
"description": "ID of the category to retrieve"
}
},
"required": ["category_id"]
}
),
Tool(
name="create_category",
description="Create a new category",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name of the category"
},
"description": {
"type": "string",
"description": "Description of the category"
},
"is_income": {
"type": "boolean",
"description": "Whether this is an income category"
},
"exclude_from_budget": {
"type": "boolean",
"description": "Whether to exclude from budget"
},
"exclude_from_totals": {
"type": "boolean",
"description": "Whether to exclude from totals"
},
"archived": {
"type": "boolean",
"description": "Whether the category is archived"
},
"group_id": {
"type": "integer",
"description": "ID of the category group"
}
},
"required": ["name"]
}
),
Tool(
name="update_category",
description="Update an existing category",
inputSchema={
"type": "object",
"properties": {
"category_id": {
"type": "integer",
"description": "ID of the category to update"
},
"name": {
"type": "string",
"description": "Name of the category"
},
"description": {
"type": "string",
"description": "Description of the category"
},
"is_income": {
"type": "boolean",
"description": "Whether this is an income category"
},
"exclude_from_budget": {
"type": "boolean",
"description": "Whether to exclude from budget"
},
"exclude_from_totals": {
"type": "boolean",
"description": "Whether to exclude from totals"
},
"archived": {
"type": "boolean",
"description": "Whether the category is archived"
},
"group_id": {
"type": "integer",
"description": "ID of the category group"
}
},
"required": ["category_id"]
}
),
Tool(
name="delete_category",
description="Delete a category (soft delete)",
inputSchema={
"type": "object",
"properties": {
"category_id": {
"type": "integer",
"description": "ID of the category to delete"
}
},
"required": ["category_id"]
}
),
# Transaction tools
Tool(
name="get_all_transactions",
description="Get all transactions with optional filters",
inputSchema={
"type": "object",
"properties": {
"tag_id": {
"type": "integer",
"description": "Filter by tag ID"
},
"recurring_id": {
"type": "integer",
"description": "Filter by recurring item ID"
},
"plaid_account_id": {
"type": "integer",
"description": "Filter by Plaid account ID"
},
"category_id": {
"type": "integer",
"description": "Filter by category ID"
},
"asset_id": {
"type": "integer",
"description": "Filter by asset ID"
},
"group_id": {
"type": "integer",
"description": "Filter by group ID"
},
"is_group": {
"type": "boolean",
"description": "Filter by group transactions"
},
"status": {
"type": "string",
"enum": ["cleared", "uncleared", "recurring", "recurring_suggested"],
"description": "Filter by transaction status"
},
"offset": {
"type": "integer",
"description": "Offset for pagination"
},
"limit": {
"type": "integer",
"description": "Limit for pagination"
},
"start_date": {
"type": "string",
"description": "Start date filter (YYYY-MM-DD)"
},
"end_date": {
"type": "string",
"description": "End date filter (YYYY-MM-DD)"
},
"debit_as_negative": {
"type": "boolean",
"description": "Return debit amounts as negative"
}
},
"required": []
}
),
Tool(
name="get_single_transaction",
description="Get details of a specific transaction",
inputSchema={
"type": "object",
"properties": {
"transaction_id": {
"type": "integer",
"description": "ID of the transaction to retrieve"
},
"debit_as_negative": {
"type": "boolean",
"description": "Return debit amounts as negative"
}
},
"required": ["transaction_id"]
}
),
Tool(
name="insert_transactions",
description="Insert one or more transactions",
inputSchema={
"type": "object",
"properties": {
"transactions": {
"type": "array",
"items": {
"type": "object",
"properties": {
"date": {
"type": "string",
"description": "Date in ISO format (YYYY-MM-DD)"
},
"amount": {
"type": "number",
"description": "Amount of the transaction"
},
"currency": {
"type": "string",
"description": "Currency code (e.g., USD, EUR)"
},
"asset_id": {
"type": "integer",
"description": "Asset/account ID"
},
"payee": {
"type": "string",
"description": "Payee name"
},
"category_id": {
"type": "integer",
"description": "Category ID"
},
"notes": {
"type": "string",
"description": "Transaction notes"
},
"status": {
"type": "string",
"enum": ["cleared", "uncleared"],
"description": "Transaction status"
},
"external_id": {
"type": "string",
"description": "External ID for the transaction"
},
"tags": {
"type": "array",
"items": {"type": "integer"},
"description": "Array of tag IDs"
},
"plaid_account_id": {
"type": "integer",
"description": "Plaid account ID"
}
},
"required": ["date", "amount", "currency", "asset_id", "payee"]
}
},
"apply_rules": {
"type": "boolean",
"description": "Whether to apply rules to the transactions"
},
"skip_duplicates": {
"type": "boolean",
"description": "Whether to skip duplicate transactions"
},
"check_for_recurring": {
"type": "boolean",
"description": "Whether to check for recurring transactions"
},
"debit_as_negative": {
"type": "boolean",
"description": "Treat debit amounts as negative"
},
"skip_balance_update": {
"type": "boolean",
"description": "Whether to skip balance update"
}
},
"required": ["transactions"]
}
),
Tool(
name="update_transaction",
description="Update a specific transaction",
inputSchema={
"type": "object",
"properties": {
"transaction_id": {
"type": "integer",
"description": "ID of the transaction to update"
},
"date": {
"type": "string",
"description": "Date in ISO format (YYYY-MM-DD)"
},
"amount": {
"type": "number",
"description": "Amount of the transaction"
},
"currency": {
"type": "string",
"description": "Currency code (e.g., USD, EUR)"
},
"payee": {
"type": "string",
"description": "Payee name"
},
"category_id": {
"type": "integer",
"description": "Category ID"
},
"notes": {
"type": "string",
"description": "Transaction notes"
},
"status": {
"type": "string",
"enum": ["cleared", "uncleared"],
"description": "Transaction status"
},
"tags": {
"type": "array",
"items": {"type": "integer"},
"description": "Array of tag IDs"
},
"plaid_account_id": {
"type": "integer",
"description": "Plaid account ID"
},
"debit_as_negative": {
"type": "boolean",
"description": "Treat debit amounts as negative"
},
"skip_balance_update": {
"type": "boolean",
"description": "Whether to skip balance update"
}
},
"required": ["transaction_id"]
}
),
# Tag tools
Tool(
name="get_all_tags",
description="Get all tags",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
# Asset tools
Tool(
name="get_all_assets",
description="Get all assets (accounts)",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="create_asset",
description="Create a new asset/account",
inputSchema={
"type": "object",
"properties": {
"type_name": {
"type": "string",
"description": "Type of asset (e.g., 'checking', 'savings', 'credit')"
},
"subtype_name": {
"type": "string",
"description": "Subtype of asset"
},
"name": {
"type": "string",
"description": "Name of the asset"
},
"display_name": {
"type": "string",
"description": "Display name for the asset"
},
"balance": {
"type": "number",
"description": "Current balance"
},
"balance_as_of": {
"type": "string",
"description": "Date of the balance (YYYY-MM-DD)"
},
"currency": {
"type": "string",
"description": "Currency code"
},
"institution_name": {
"type": "string",
"description": "Institution name"
}
},
"required": ["type_name", "name", "balance", "balance_as_of", "currency"]
}
),
Tool(
name="update_asset",
description="Update an existing asset",
inputSchema={
"type": "object",
"properties": {
"asset_id": {
"type": "integer",
"description": "ID of the asset to update"
},
"name": {
"type": "string",
"description": "Name of the asset"
},
"display_name": {
"type": "string",
"description": "Display name for the asset"
},
"balance": {
"type": "number",
"description": "Current balance"
},
"balance_as_of": {
"type": "string",
"description": "Date of the balance (YYYY-MM-DD)"
},
"currency": {
"type": "string",
"description": "Currency code"
},
"institution_name": {
"type": "string",
"description": "Institution name"
}
},
"required": ["asset_id"]
}
),
# Budget tools
Tool(
name="get_budget_summary",
description="Get budget summary for a specific period",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "Start date (YYYY-MM-DD)"
},
"end_date": {
"type": "string",
"description": "End date (YYYY-MM-DD)"
},
"currency": {
"type": "string",
"description": "Currency code"
}
},
"required": ["start_date", "end_date"]
}
),
Tool(
name="upsert_budget",
description="Create or update budget data",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "Start date (YYYY-MM-DD)"
},
"category_id": {
"type": "integer",
"description": "Category ID"
},
"amount": {
"type": "number",
"description": "Budget amount"
},
"currency": {
"type": "string",
"description": "Currency code"
}
},
"required": ["start_date", "category_id", "amount"]
}
),
# Recurring items tools
Tool(
name="get_recurring_items",
description="Get all recurring items",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
# Plaid account tools
Tool(
name="get_all_plaid_accounts",
description="Get all Plaid accounts",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="trigger_plaid_fetch",
description="Trigger a fetch for latest data from Plaid",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"description": "Start date for fetch (YYYY-MM-DD)"
},
"end_date": {
"type": "string",
"description": "End date for fetch (YYYY-MM-DD)"
},
"plaid_account_id": {
"type": "integer",
"description": "Specific Plaid account ID to fetch"
}
},
"required": []
}
),
# Crypto tools
Tool(
name="get_all_crypto",
description="Get all crypto assets",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="update_manual_crypto",
description="Update a manual crypto asset",
inputSchema={
"type": "object",
"properties": {
"crypto_id": {
"type": "integer",
"description": "ID of the crypto asset to update"
},
"name": {
"type": "string",
"description": "Name of the crypto asset"
},
"display_name": {
"type": "string",
"description": "Display name"
},
"institution_name": {
"type": "string",
"description": "Institution name"
},
"balance": {
"type": "number",
"description": "Current balance"
},
"currency": {
"type": "string",
"description": "Cryptocurrency symbol"
}
},
"required": ["crypto_id"]
}
)
]
# Tool call handlers
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls"""
async with httpx.AsyncClient(
base_url=API_BASE_URL,
headers={
"Authorization": f"Bearer {ACCESS_TOKEN}",
"Content-Type": "application/json",
},
timeout=30.0,
) as client:
try:
if name == "get_user":
response = await client.get("/v1/me")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_categories":
params = {}
if "format" in arguments:
params["format"] = arguments["format"]
response = await client.get("/v1/categories", params=params)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_single_category":
category_id = arguments["category_id"]
response = await client.get(f"/v1/categories/{category_id}")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "create_category":
response = await client.post("/v1/categories", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "update_category":
category_id = arguments.pop("category_id")
response = await client.put(f"/v1/categories/{category_id}", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "delete_category":
category_id = arguments["category_id"]
response = await client.delete(f"/v1/categories/{category_id}")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_transactions":
params = {k: v for k, v in arguments.items() if v is not None}
response = await client.get("/v1/transactions", params=params)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_single_transaction":
transaction_id = arguments["transaction_id"]
params = {}
if "debit_as_negative" in arguments:
params["debit_as_negative"] = arguments["debit_as_negative"]
response = await client.get(f"/v1/transactions/{transaction_id}", params=params)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "insert_transactions":
response = await client.post("/v1/transactions", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "update_transaction":
transaction_id = arguments.pop("transaction_id")
response = await client.put(f"/v1/transactions/{transaction_id}", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_tags":
response = await client.get("/v1/tags")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_assets":
response = await client.get("/v1/assets")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "create_asset":
response = await client.post("/v1/assets", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "update_asset":
asset_id = arguments.pop("asset_id")
response = await client.put(f"/v1/assets/{asset_id}", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_budget_summary":
params = {k: v for k, v in arguments.items() if v is not None}
response = await client.get("/v1/budgets", params=params)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "upsert_budget":
response = await client.put("/v1/budgets", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_recurring_items":
response = await client.get("/v1/recurring_items")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_plaid_accounts":
response = await client.get("/v1/plaid_accounts")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "trigger_plaid_fetch":
response = await client.post("/v1/plaid_accounts/fetch", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "get_all_crypto":
response = await client.get("/v1/crypto")
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
elif name == "update_manual_crypto":
crypto_id = arguments.pop("crypto_id")
response = await client.put(f"/v1/crypto/manual/{crypto_id}", json=arguments)
response.raise_for_status()
return [TextContent(type="text", text=json.dumps(response.json(), indent=2))]
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except httpx.HTTPStatusError as e:
error_msg = f"HTTP {e.response.status_code}: {e.response.text}"
return [TextContent(type="text", text=f"Error: {error_msg}")]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def async_main():
"""Main server entry point (async)"""
import sys
# Handle command line arguments
if len(sys.argv) > 1:
if sys.argv[1] in ['--help', '-h']:
print("Lunch Money MCP Server")
print("=" * 30)
print("A Model Context Protocol server for Lunch Money API")
print()
print("Usage:")
print(" lunchmoney-mcp # Start MCP server")
print(" lunchmoney-mcp --help # Show this help")
print(" lunchmoney-mcp --version # Show version")
print()
print("Environment Variables:")
print(" LUNCHMONEY_ACCESS_TOKEN # Your Lunch Money API token")
print(" # Get from: https://my.lunchmoney.app/developers")
print()
print("Documentation: https://github.com/MatiasL13/lunchmoney-mcp")
return
elif sys.argv[1] in ['--version', '-v']:
from . import __version__
print(f"lunchmoney-mcp {__version__}")
return
# Validate access token before starting the server
validate_access_token()
# Start the MCP server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def main():
"""Synchronous entry point for command line"""
asyncio.run(async_main())
if __name__ == "__main__":
main()