EislM0203andClaude Opus 5
c9a1375fb9
Two long-standing correctness gaps, both flagged in fin-research-frontier. A1 — POST /api/transactions could write half a linked pair TransactionCreate inherited the full six-value type Literal and accepted a free-standing linked_account_id, so any client could create a lone `transfer` row with no partner. That is exactly the balance corruption invariant #1 exists to prevent: the orphan counts toward SUM(amount - fee) and paired-delete can never find a partner to remove it with. TransactionCreate now narrows type to the single-leg kinds and rejects a caller-supplied linked_account_id. TransactionBase stays wide on purpose — reads must keep describing historical rows of every type. splitwise_pay deliberately stays allowed. The two-leg creator uses that type, but a standalone row carrying splitwise_partner_share is also a legitimate record, and is what `fin tx add --type splitwise_pay --partner-share` writes. Narrowing to expense/income would have broken it. B3 — no idempotency, so any retry double-booked Migration i9d0e1f2a3b4 adds a nullable client_ref to transactions. POST /api/transactions, /batch, /transfer and /settlement accept it, and crud.find_by_client_ref is checked before writing, so retrying a timed-out request replays the original rows instead of booking them again. NULL means "no idempotency requested" and never collides. Deviation from the frontier's sketch, which called for unique-when-not-null: the index is NOT unique. A transfer writes two rows and a batch writes many, all under one key, so uniqueness cannot live on the column. Suppression is a check-then-write inside the creators instead — sound here because SQLite serialises writers and the scheduler never writes transactions. Exposed on the MCP create_transaction/create_transfer/create_settlement tools, since a retrying agent is the case B3 was written about. The server instructions no longer tell agents that writes are non-idempotent. Tests: 60 passing, up from 47. Thirteen new cover retries collapsing to one write across all four endpoints, differing keys staying separate, NULL keys not colliding, and the orphan-leg guard refusing transfer/settlement/manual linkage while the four legitimate single-leg types still work. Migration verified both directions on a scratch DB seeded at the previous head: existing rows survive with NULL client_ref, index is created, and downgrade drops cleanly with rows intact. Verified live over HTTP that a transfer submitted three times with one client_ref yields one pair of legs. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L3qYLuu9jgXsCRXmJ18YGY
1111 lines
46 KiB
Python
1111 lines
46 KiB
Python
"""MCP server wrapping FinClient.
|
||
|
||
Entry point: `fin-mcp`.
|
||
|
||
Transport is selected with FIN_MCP_TRANSPORT:
|
||
stdio (default) — for local clients that spawn the process
|
||
streamable-http — for running as a network service; binds
|
||
FIN_MCP_HOST:FIN_MCP_PORT and serves at /mcp
|
||
|
||
Default mode: read-only (15 tools).
|
||
Set FIN_MCP_ALLOW_WRITES=1 to additionally enable 24 write tools.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from typing import Any, Literal
|
||
|
||
try:
|
||
from mcp.server.fastmcp import FastMCP
|
||
from mcp.types import ToolAnnotations
|
||
except ImportError as exc:
|
||
raise ImportError(
|
||
"fin-mcp requires the 'mcp' extra: pip install 'fin-cli[mcp]'"
|
||
) from exc
|
||
|
||
from fin_cli.client import FinClient
|
||
|
||
TRUTHY = {"1", "true", "yes", "on"}
|
||
|
||
|
||
class _LazyClient:
|
||
"""Defers FinClient construction until the first tool call.
|
||
|
||
FinClient raises when FIN_API_URL is unset. Doing that at import/startup
|
||
kills the process before the MCP handshake, so the client only ever sees
|
||
"server failed to start" with no cause. Deferring it turns the same
|
||
misconfiguration into a readable error on the first tool result.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._client: FinClient | None = None
|
||
|
||
def __getattr__(self, name: str) -> Any:
|
||
if self._client is None:
|
||
self._client = FinClient()
|
||
return getattr(self._client, name)
|
||
|
||
|
||
def _match_by_name(rows: list[dict[str, Any]] | None, name: str) -> list[dict[str, Any]]:
|
||
"""Resolve a human name to rows, preferring exact matches.
|
||
|
||
An exact case-insensitive hit wins outright, so searching "Food" cannot be
|
||
drowned out by "Fast Food". Only when nothing matches exactly do we fall
|
||
back to substring matching and let the caller disambiguate.
|
||
"""
|
||
needle = name.strip().lower()
|
||
rows = rows or []
|
||
exact = [r for r in rows if (r.get("name") or "").strip().lower() == needle]
|
||
if exact:
|
||
return exact
|
||
return [r for r in rows if needle in (r.get("name") or "").lower()]
|
||
|
||
|
||
def _instructions(allow_writes: bool) -> str:
|
||
mode_line = (
|
||
"writes ENABLED"
|
||
if allow_writes
|
||
else (
|
||
"READ-ONLY — write tools are disabled; "
|
||
"set FIN_MCP_ALLOW_WRITES=1 on the server to enable"
|
||
)
|
||
)
|
||
base = (
|
||
"fin is a single-user, self-hosted personal finance app tracking bank accounts, "
|
||
"investments, RSU grants, Splitwise balances, receivables, and net worth in EUR.\n\n"
|
||
"Tool return convention: every tool returns {\"result\": <payload>} where <payload> "
|
||
"is the raw API JSON (list, dict, or null). Unwrap result before displaying.\n\n"
|
||
f"Current mode: {mode_line}.\n\n"
|
||
"Sign convention: amounts are negative for money out (expenses, transfers from an "
|
||
"account), positive for money in (income, transfers to an account)."
|
||
)
|
||
if not allow_writes:
|
||
return base
|
||
return base + (
|
||
"\n\nWrite discipline (applies to every non-read tool):\n"
|
||
"- Confirm amounts, accounts and dates with the user before writing.\n"
|
||
"- Pass a unique client_ref on create_transaction, create_transfer and "
|
||
"create_settlement. If the call fails ambiguously, retry with the SAME "
|
||
"client_ref — the write is returned rather than repeated, so you cannot "
|
||
"double-book. Tools without a client_ref parameter are not yet protected: "
|
||
"there, list recent transactions and check for an identical row first.\n"
|
||
"- Accounts and categories are addressed by integer id, never by name. "
|
||
"Resolve the id with find_account / find_category first; if the lookup "
|
||
"returns nothing or more than one row, ask the user instead of guessing."
|
||
)
|
||
|
||
|
||
def create_server(
|
||
client: FinClient | None = None,
|
||
allow_writes: bool | None = None,
|
||
) -> FastMCP:
|
||
"""Create the fin MCP server.
|
||
|
||
Args:
|
||
client: FinClient to use; if None, built from FIN_API_URL / FIN_API_TOKEN env vars.
|
||
allow_writes: if None, read from FIN_MCP_ALLOW_WRITES env var.
|
||
"""
|
||
if allow_writes is None:
|
||
allow_writes = os.getenv("FIN_MCP_ALLOW_WRITES", "").lower() in TRUTHY
|
||
if client is None:
|
||
# Built lazily on first tool call. Constructing here would abort the
|
||
# process before the MCP handshake when FIN_API_URL is unset, leaving
|
||
# the calling client with nothing but "server failed to start".
|
||
client = _LazyClient()
|
||
|
||
mcp = FastMCP("fin", instructions=_instructions(allow_writes))
|
||
|
||
# ------------------------------------------------------------------
|
||
# Read tools (always registered)
|
||
# ------------------------------------------------------------------
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def health() -> dict[str, Any]:
|
||
"""Check that the fin API is reachable and healthy."""
|
||
return {"result": client.health()}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_net_worth() -> dict[str, Any]:
|
||
"""Return the current net worth breakdown in EUR, grouped by account."""
|
||
return {"result": client.get_net_worth()}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_net_worth_history(
|
||
range: Literal["1M", "3M", "6M", "1Y", "all"] = "all"
|
||
) -> dict[str, Any]:
|
||
"""Return net worth snapshots ordered by date ascending.
|
||
|
||
Args:
|
||
range: Time range — one of: 1M, 3M, 6M, 1Y, all (default: all).
|
||
"""
|
||
return {"result": client.get_net_worth_history(range=range)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def list_accounts() -> dict[str, Any]:
|
||
"""List all active accounts (bank, investment, RSU, Splitwise, receivable, payable)."""
|
||
return {"result": client.list_accounts()}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def list_categories() -> dict[str, Any]:
|
||
"""List all spending and income categories."""
|
||
return {"result": client.list_categories()}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def find_account(name: str) -> dict[str, Any]:
|
||
"""Resolve an account name to the integer id that other tools require.
|
||
|
||
Every other tool addresses accounts by integer id, so use this instead
|
||
of listing all accounts and matching names by hand.
|
||
|
||
An exact case-insensitive match wins outright; otherwise all substring
|
||
matches are returned. An empty list means no such account — ask the
|
||
user rather than guessing an id. More than one result means the name
|
||
was ambiguous, so ask which one they meant.
|
||
|
||
Args:
|
||
name: Full or partial account name, e.g. "checking".
|
||
"""
|
||
return {"result": _match_by_name(client.list_accounts(), name)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def find_category(name: str) -> dict[str, Any]:
|
||
"""Resolve a category name to the integer id that other tools require.
|
||
|
||
Same matching rules as find_account: exact match wins, otherwise
|
||
substring matches. Empty means no such category — do not invent an id,
|
||
and do not silently create one.
|
||
|
||
Args:
|
||
name: Full or partial category name, e.g. "groceries".
|
||
"""
|
||
return {"result": _match_by_name(client.list_categories(), name)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def list_transactions(
|
||
account_id: int | None = None,
|
||
category_id: int | None = None,
|
||
month: str | None = None,
|
||
limit: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""List transactions, optionally filtered.
|
||
|
||
Args:
|
||
account_id: Filter to one account (integer id).
|
||
category_id: Filter to one category (integer id).
|
||
month: Filter to one calendar month in YYYY-MM format.
|
||
limit: Maximum number of rows to return (most recent first).
|
||
"""
|
||
return {
|
||
"result": client.list_transactions(
|
||
account_id=account_id,
|
||
category_id=category_id,
|
||
month=month,
|
||
limit=limit,
|
||
)
|
||
}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_holdings(account_id: int) -> dict[str, Any]:
|
||
"""Return holdings with live P&L for one investment or RSU account.
|
||
|
||
Args:
|
||
account_id: Integer id of an investment or RSU account.
|
||
"""
|
||
return {"result": client.get_holdings(account_id=account_id)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def list_stock_transactions(account_id: int | None = None) -> dict[str, Any]:
|
||
"""List stock transactions (buy/sell/rsu_vest), optionally filtered by account.
|
||
|
||
Args:
|
||
account_id: If provided, return only trades for this account.
|
||
"""
|
||
return {"result": client.list_stock_transactions(account_id=account_id)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def list_rsu_grants(include_vested: bool = False) -> dict[str, Any]:
|
||
"""List RSU grants.
|
||
|
||
Args:
|
||
include_vested: If True, include already-vested grants (default: False).
|
||
"""
|
||
return {"result": client.list_rsu_grants(include_vested=include_vested)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def preview_rsu_schedule(
|
||
account_id: int,
|
||
ticker: str,
|
||
grant_date: str,
|
||
total_gross_shares: float,
|
||
cliff_months: int = 12,
|
||
total_months: int = 36,
|
||
frequency_months: int = 3,
|
||
withholding_pct: float = 0.5,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Preview the vesting schedule for a hypothetical RSU grant without writing anything.
|
||
|
||
Pure computation — safe in read-only mode; no data is stored.
|
||
|
||
Args:
|
||
account_id: RSU account to associate with.
|
||
ticker: Stock ticker symbol (1-12 chars, alphanumeric/dot/dash).
|
||
grant_date: Grant date in YYYY-MM-DD format.
|
||
total_gross_shares: Total gross shares in the grant (>0).
|
||
cliff_months: Months until first vest (1-60, default 12).
|
||
total_months: Total vesting period in months (12-120, default 36).
|
||
frequency_months: Vesting frequency in months (1-12, default 3).
|
||
withholding_pct: Tax withholding fraction (0-1, default 0.5).
|
||
notes: Optional notes (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"account_id": account_id,
|
||
"ticker": ticker,
|
||
"grant_date": grant_date,
|
||
"total_gross_shares": total_gross_shares,
|
||
"cliff_months": cliff_months,
|
||
"total_months": total_months,
|
||
"frequency_months": frequency_months,
|
||
"withholding_pct": withholding_pct,
|
||
}
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
return {"result": client.preview_rsu_schedule(data=data)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_spending_month(month: str, account_id: int | None = None) -> dict[str, Any]:
|
||
"""Return spending breakdown by category for one calendar month.
|
||
|
||
Args:
|
||
month: Month in YYYY-MM format (required).
|
||
account_id: If provided, restrict to one account.
|
||
"""
|
||
return {"result": client.get_spending_month(month=month, account_id=account_id)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_spending_summary(
|
||
mode: Literal["yearly", "custom"],
|
||
year: int | None = None,
|
||
start: str | None = None,
|
||
end: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Aggregate spending over a time range.
|
||
|
||
Args:
|
||
mode: Aggregation mode — "yearly" (requires year) or "custom" (requires start + end).
|
||
year: Calendar year (e.g. 2026); used with mode=yearly.
|
||
start: Start date in YYYY-MM-DD format; used with mode=custom.
|
||
end: End date in YYYY-MM-DD format; used with mode=custom.
|
||
"""
|
||
params: dict[str, Any] = {"mode": mode}
|
||
if year is not None:
|
||
params["year"] = year
|
||
if start is not None:
|
||
params["start"] = start
|
||
if end is not None:
|
||
params["end"] = end
|
||
return {"result": client.get_spending_summary(**params)}
|
||
|
||
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
|
||
def get_fx_rate(
|
||
from_currency: str = "EUR",
|
||
to_currency: str = "USD",
|
||
) -> dict[str, Any]:
|
||
"""Return the cached FX rate between two currencies.
|
||
|
||
Args:
|
||
from_currency: Source currency code (default: EUR).
|
||
to_currency: Target currency code (default: USD).
|
||
"""
|
||
return {
|
||
"result": client.get_fx_rate(
|
||
from_currency=from_currency, to_currency=to_currency
|
||
)
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Write tools (registered only when allow_writes is True)
|
||
# ------------------------------------------------------------------
|
||
|
||
if allow_writes:
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_account(
|
||
name: str,
|
||
type: Literal["bank", "investment", "rsu", "splitwise", "receivable", "payable"],
|
||
currency: str = "EUR",
|
||
notes: str | None = None,
|
||
manual_override_balance: float | None = None,
|
||
is_active: bool = True,
|
||
initial_balance: float | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create a new account.
|
||
|
||
Args:
|
||
name: Account name (max 100 chars).
|
||
type: One of: bank, investment, rsu, splitwise, receivable, payable.
|
||
currency: Currency code (default EUR).
|
||
notes: Optional notes.
|
||
manual_override_balance: If set, overrides the computed balance display.
|
||
is_active: Whether the account is active (default True).
|
||
initial_balance: If set, creates an opening-balance income transaction.
|
||
"""
|
||
data: dict[str, Any] = {"name": name, "type": type, "currency": currency, "is_active": is_active}
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
if manual_override_balance is not None:
|
||
data["manual_override_balance"] = manual_override_balance
|
||
if initial_balance is not None:
|
||
data["initial_balance"] = initial_balance
|
||
return {"result": client.create_account(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def update_account(
|
||
account_id: int,
|
||
name: str | None = None,
|
||
type: Literal["bank", "investment", "rsu", "splitwise", "receivable", "payable"] | None = None,
|
||
currency: str | None = None,
|
||
notes: str | None = None,
|
||
manual_override_balance: float | None = None,
|
||
is_active: bool | None = None,
|
||
clear_manual_override_balance: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""Update fields on an existing account. Omitted fields are left unchanged.
|
||
|
||
Args:
|
||
account_id: Integer id of the account to update.
|
||
name: New account name (max 100 chars).
|
||
type: New account type.
|
||
currency: New currency code.
|
||
notes: New notes text.
|
||
manual_override_balance: New override for the displayed balance.
|
||
is_active: Set False to deactivate, True to reactivate.
|
||
clear_manual_override_balance: Set True to remove an existing
|
||
override so the balance is computed from transactions again.
|
||
"""
|
||
data: dict[str, Any] = {}
|
||
if name is not None:
|
||
data["name"] = name
|
||
if type is not None:
|
||
data["type"] = type
|
||
if currency is not None:
|
||
data["currency"] = currency
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
if manual_override_balance is not None:
|
||
data["manual_override_balance"] = manual_override_balance
|
||
if is_active is not None:
|
||
data["is_active"] = is_active
|
||
if clear_manual_override_balance:
|
||
data["manual_override_balance"] = None
|
||
return {"result": client.update_account(account_id=account_id, data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(
|
||
readOnlyHint=False, destructiveHint=False, idempotentHint=True
|
||
)
|
||
)
|
||
def deactivate_account(account_id: int) -> dict[str, Any]:
|
||
"""Deactivate (soft-delete) an account. Reversible via update_account with is_active=True.
|
||
|
||
Args:
|
||
account_id: Integer id of the account to deactivate.
|
||
"""
|
||
return {"result": client.deactivate_account(account_id=account_id)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def adjust_account_balance(account_id: int, target_balance: float) -> dict[str, Any]:
|
||
"""Adjust a bank account's balance to a target value by creating a correction transaction.
|
||
|
||
Args:
|
||
account_id: Integer id of a bank account.
|
||
target_balance: The desired balance in the account's currency.
|
||
"""
|
||
return {
|
||
"result": client.adjust_account_balance(
|
||
account_id=account_id, target_balance=target_balance
|
||
)
|
||
}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_category(
|
||
name: str,
|
||
color: str,
|
||
icon: str,
|
||
type: Literal["expense", "income"] = "expense",
|
||
) -> dict[str, Any]:
|
||
"""Create a new spending or income category.
|
||
|
||
Args:
|
||
name: Category name (max 100 chars).
|
||
color: CSS color string (max 20 chars), e.g. "#e74c3c".
|
||
icon: Emoji or icon name (max 20 chars).
|
||
type: Either "expense" (default) or "income".
|
||
"""
|
||
data: dict[str, Any] = {"name": name, "color": color, "icon": icon, "type": type}
|
||
return {"result": client.create_category(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def update_category(
|
||
category_id: int,
|
||
name: str | None = None,
|
||
color: str | None = None,
|
||
icon: str | None = None,
|
||
type: Literal["expense", "income"] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Update fields on an existing category. Omitted fields are left unchanged.
|
||
|
||
Args:
|
||
category_id: Integer id of the category to update.
|
||
name: New category name (max 100 chars).
|
||
color: New CSS color string (max 20 chars), e.g. "#e74c3c".
|
||
icon: New emoji or icon name (max 20 chars).
|
||
type: New category type.
|
||
"""
|
||
data: dict[str, Any] = {}
|
||
if name is not None:
|
||
data["name"] = name
|
||
if color is not None:
|
||
data["color"] = color
|
||
if icon is not None:
|
||
data["icon"] = icon
|
||
if type is not None:
|
||
data["type"] = type
|
||
return {"result": client.update_category(category_id=category_id, data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
|
||
)
|
||
def delete_category(category_id: int) -> dict[str, Any]:
|
||
"""Delete a category. Transactions referencing it will have their category set to null.
|
||
|
||
Args:
|
||
category_id: Integer id of the category to delete.
|
||
"""
|
||
return {"result": client.delete_category(category_id=category_id)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_transaction(
|
||
date: str,
|
||
account_id: int,
|
||
amount: float,
|
||
type: Literal["expense", "income"],
|
||
currency: str = "EUR",
|
||
category_id: int | None = None,
|
||
description: str | None = None,
|
||
fee: float = 0,
|
||
exchange_rate_used: float | None = None,
|
||
client_ref: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create a single expense or income transaction.
|
||
|
||
Handles standalone expenses and income only. Transfers, settlements and
|
||
Splitwise entries are two-leg writes that MUST go through their own tools —
|
||
create_transfer, create_settlement, create_split_expense or
|
||
create_splitwise_i_paid. Writing one leg here would leave the ledger
|
||
unbalanced, so this tool cannot express those types.
|
||
|
||
Args:
|
||
date: Transaction date in YYYY-MM-DD format.
|
||
account_id: Integer id of the bank account.
|
||
amount: Amount in the account's currency. Negative for expenses, positive for income.
|
||
type: "expense" or "income" only.
|
||
currency: Currency code (default EUR).
|
||
category_id: Optional integer category id.
|
||
description: Optional description (max 500 chars).
|
||
fee: Fee amount (included in balance calculation; default 0).
|
||
exchange_rate_used: Optional FX rate used for this transaction.
|
||
client_ref: Optional idempotency key. If a write with this key already
|
||
exists it is returned unchanged instead of booking a second time.
|
||
Set one before any write you might have to retry.
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"date": date,
|
||
"account_id": account_id,
|
||
"amount": amount,
|
||
"type": type,
|
||
"currency": currency,
|
||
"fee": fee,
|
||
}
|
||
if category_id is not None:
|
||
data["category_id"] = category_id
|
||
if description is not None:
|
||
data["description"] = description
|
||
if exchange_rate_used is not None:
|
||
data["exchange_rate_used"] = exchange_rate_used
|
||
if client_ref is not None:
|
||
data["client_ref"] = client_ref
|
||
return {"result": client.create_transaction(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def update_transaction(
|
||
transaction_id: int,
|
||
amount: float | None = None,
|
||
description: str | None = None,
|
||
category_id: int | None = None,
|
||
date: str | None = None,
|
||
clear_category: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""Update an existing expense or income transaction. Omitted fields are unchanged.
|
||
|
||
Note: the API rejects updates to linked transfer/settlement legs with 422.
|
||
To change a transfer, delete the whole group and re-enter.
|
||
|
||
Args:
|
||
transaction_id: Integer id of the transaction to update.
|
||
amount: New amount (negative for expenses, positive for income).
|
||
description: New description (max 500 chars).
|
||
category_id: New category integer id.
|
||
date: New date in YYYY-MM-DD format.
|
||
clear_category: Set True to remove the category entirely.
|
||
"""
|
||
data: dict[str, Any] = {}
|
||
if amount is not None:
|
||
data["amount"] = amount
|
||
if description is not None:
|
||
data["description"] = description
|
||
if category_id is not None:
|
||
data["category_id"] = category_id
|
||
if date is not None:
|
||
data["date"] = date
|
||
if clear_category:
|
||
data["category_id"] = None
|
||
return {
|
||
"result": client.update_transaction(
|
||
transaction_id=transaction_id, data=data
|
||
)
|
||
}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
|
||
)
|
||
def delete_transaction(transaction_id: int) -> dict[str, Any]:
|
||
"""Delete a transaction. If the transaction is part of a transfer/settlement group,
|
||
the entire linked group is deleted atomically.
|
||
|
||
Args:
|
||
transaction_id: Integer id of the transaction (or any leg of a linked group).
|
||
"""
|
||
return {"result": client.delete_transaction(transaction_id=transaction_id)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_transfer(
|
||
from_account_id: int,
|
||
to_account_id: int,
|
||
amount: float,
|
||
date: str,
|
||
currency: str = "EUR",
|
||
fee: float = 0,
|
||
description: str = "Transfer",
|
||
to_amount: float | None = None,
|
||
to_currency: str | None = None,
|
||
client_ref: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create an atomic two-leg transfer between two accounts.
|
||
|
||
For cross-currency transfers, provide to_amount and to_currency.
|
||
|
||
Args:
|
||
from_account_id: Source account integer id.
|
||
to_account_id: Destination account integer id.
|
||
amount: Amount in the source currency (positive).
|
||
date: Transfer date in YYYY-MM-DD format.
|
||
currency: Source currency code (default EUR).
|
||
fee: Fee deducted from source account (default 0).
|
||
description: Description for both legs (default "Transfer").
|
||
to_amount: Amount received in destination currency (for FX transfers).
|
||
to_currency: Destination currency code (for FX transfers).
|
||
client_ref: Optional idempotency key. If a write with this key already
|
||
exists it is returned unchanged instead of booking a second time.
|
||
Set one before any write you might have to retry.
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"from_account_id": from_account_id,
|
||
"to_account_id": to_account_id,
|
||
"amount": amount,
|
||
"currency": currency,
|
||
"fee": fee,
|
||
"date": date,
|
||
"description": description,
|
||
}
|
||
if to_amount is not None:
|
||
data["to_amount"] = to_amount
|
||
if to_currency is not None:
|
||
data["to_currency"] = to_currency
|
||
if client_ref is not None:
|
||
data["client_ref"] = client_ref
|
||
return {"result": client.create_transfer(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_settlement(
|
||
bank_account_id: int,
|
||
splitwise_account_id: int,
|
||
amount: float,
|
||
date: str,
|
||
direction: Literal["bank_to_splitwise", "splitwise_to_bank"],
|
||
currency: str = "EUR",
|
||
client_ref: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create an atomic Splitwise settlement between a bank account and a Splitwise account.
|
||
|
||
Args:
|
||
bank_account_id: Integer id of the bank account.
|
||
splitwise_account_id: Integer id of the Splitwise account.
|
||
amount: Settlement amount (positive).
|
||
date: Settlement date in YYYY-MM-DD format.
|
||
direction: "bank_to_splitwise" (you paid the Splitwise balance)
|
||
or "splitwise_to_bank" (Splitwise partner paid you).
|
||
currency: Currency code (default EUR).
|
||
client_ref: Optional idempotency key. If a write with this key already
|
||
exists it is returned unchanged instead of booking a second time.
|
||
Set one before any write you might have to retry.
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"bank_account_id": bank_account_id,
|
||
"splitwise_account_id": splitwise_account_id,
|
||
"amount": amount,
|
||
"currency": currency,
|
||
"date": date,
|
||
"direction": direction,
|
||
}
|
||
if client_ref is not None:
|
||
data["client_ref"] = client_ref
|
||
return {"result": client.create_settlement(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_split_expense(
|
||
bank_account_id: int,
|
||
receivable_account_id: int,
|
||
total_amount: float,
|
||
partner_share: float,
|
||
date: str,
|
||
currency: str = "EUR",
|
||
category_id: int | None = None,
|
||
description: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create a shared bill where you paid and a partner owes you their share.
|
||
|
||
Records the full expense on the bank account and a receivable for the partner's share.
|
||
|
||
Args:
|
||
bank_account_id: Integer id of the bank account you paid from.
|
||
receivable_account_id: Integer id of the receivable account for the partner.
|
||
total_amount: Total bill amount (>0).
|
||
partner_share: Partner's share of the bill (>0, must be less than total_amount).
|
||
date: Date in YYYY-MM-DD format.
|
||
currency: Currency code (default EUR).
|
||
category_id: Optional integer category id.
|
||
description: Optional description (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"bank_account_id": bank_account_id,
|
||
"receivable_account_id": receivable_account_id,
|
||
"total_amount": total_amount,
|
||
"partner_share": partner_share,
|
||
"currency": currency,
|
||
"date": date,
|
||
}
|
||
if category_id is not None:
|
||
data["category_id"] = category_id
|
||
if description is not None:
|
||
data["description"] = description
|
||
return {"result": client.create_split_expense(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_splitwise_i_paid(
|
||
bank_account_id: int,
|
||
splitwise_account_id: int,
|
||
total_amount: float,
|
||
partner_share: float,
|
||
date: str,
|
||
currency: str = "EUR",
|
||
category_id: int | None = None,
|
||
description: str = "Splitwise: I paid",
|
||
) -> dict[str, Any]:
|
||
"""Record a Splitwise expense where you paid the full bill and the group owes you.
|
||
|
||
Args:
|
||
bank_account_id: Integer id of the bank account you paid from.
|
||
splitwise_account_id: Integer id of the Splitwise account.
|
||
total_amount: Total bill amount (>0).
|
||
partner_share: Partner's share (>0, must be less than total_amount).
|
||
date: Date in YYYY-MM-DD format.
|
||
currency: Currency code (default EUR).
|
||
category_id: Optional integer category id.
|
||
description: Description (default "Splitwise: I paid").
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"bank_account_id": bank_account_id,
|
||
"splitwise_account_id": splitwise_account_id,
|
||
"total_amount": total_amount,
|
||
"partner_share": partner_share,
|
||
"currency": currency,
|
||
"date": date,
|
||
"description": description,
|
||
}
|
||
if category_id is not None:
|
||
data["category_id"] = category_id
|
||
return {"result": client.create_splitwise_i_paid(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def reconcile(
|
||
bank_account_id: int,
|
||
other_account_id: int,
|
||
amount: float,
|
||
date: str,
|
||
description: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Settle a receivable or payable against a bank account.
|
||
|
||
For receivables (someone pays you back): bank receives +amount, other account −amount.
|
||
For payables (you pay someone back): bank sends −amount, other account +amount.
|
||
Both legs are typed 'settlement' and do not count as income or spending.
|
||
|
||
Args:
|
||
bank_account_id: Integer id of the bank account.
|
||
other_account_id: Integer id of the payable or receivable account.
|
||
amount: Amount to settle (>0).
|
||
date: Settlement date in YYYY-MM-DD format.
|
||
description: Optional description.
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"bank_account_id": bank_account_id,
|
||
"other_account_id": other_account_id,
|
||
"amount": amount,
|
||
"date": date,
|
||
}
|
||
if description is not None:
|
||
data["description"] = description
|
||
return {"result": client.reconcile(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_stock_transaction(
|
||
account_id: int,
|
||
ticker: str,
|
||
type: Literal["buy", "sell", "rsu_vest"],
|
||
shares: float,
|
||
price_per_share: float,
|
||
total_cost: float,
|
||
date: str,
|
||
fee: float = 0,
|
||
currency: str = "USD",
|
||
exchange_rate_used: float | None = None,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Record a stock buy, sell, or RSU vest. Updates the holding immediately.
|
||
|
||
For a buy funded by a bank account use funded_buy instead (records
|
||
both the trade and the cash withdrawal atomically).
|
||
|
||
Args:
|
||
account_id: Integer id of an investment or RSU account.
|
||
ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
|
||
type: "buy", "sell", or "rsu_vest".
|
||
shares: Number of shares (>0).
|
||
price_per_share: Price per share (>0).
|
||
total_cost: Total cost including fees (>=0).
|
||
date: Trade date in YYYY-MM-DD format.
|
||
fee: Commission/fee (default 0).
|
||
currency: Currency of the price (default USD).
|
||
exchange_rate_used: Optional FX rate used.
|
||
notes: Optional notes (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"account_id": account_id,
|
||
"ticker": ticker,
|
||
"type": type,
|
||
"shares": shares,
|
||
"price_per_share": price_per_share,
|
||
"fee": fee,
|
||
"total_cost": total_cost,
|
||
"date": date,
|
||
"currency": currency,
|
||
}
|
||
if exchange_rate_used is not None:
|
||
data["exchange_rate_used"] = exchange_rate_used
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
return {"result": client.create_stock_transaction(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def funded_buy(
|
||
account_id: int,
|
||
ticker: str,
|
||
shares: float,
|
||
price_per_share: float,
|
||
total_cost: float,
|
||
funded_by_account_id: int,
|
||
date: str,
|
||
type: Literal["buy", "sell"] = "buy",
|
||
fee: float = 0,
|
||
currency: str = "USD",
|
||
exchange_rate_used: float | None = None,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Buy stock and simultaneously deduct the cost from a bank account (atomic two-leg write).
|
||
|
||
Args:
|
||
account_id: Integer id of the investment account receiving the shares.
|
||
ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
|
||
shares: Number of shares purchased (>0).
|
||
price_per_share: Price per share (>0).
|
||
total_cost: Total cost including fees (>=0).
|
||
funded_by_account_id: Integer id of the bank account to debit.
|
||
date: Trade date in YYYY-MM-DD format.
|
||
type: "buy" or "sell" (default "buy").
|
||
fee: Commission/fee (default 0).
|
||
currency: Currency of the price (default USD).
|
||
exchange_rate_used: Optional FX rate used.
|
||
notes: Optional notes (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"account_id": account_id,
|
||
"ticker": ticker,
|
||
"type": type,
|
||
"shares": shares,
|
||
"price_per_share": price_per_share,
|
||
"fee": fee,
|
||
"total_cost": total_cost,
|
||
"date": date,
|
||
"currency": currency,
|
||
"funded_by_account_id": funded_by_account_id,
|
||
}
|
||
if exchange_rate_used is not None:
|
||
data["exchange_rate_used"] = exchange_rate_used
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
return {"result": client.funded_buy(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
|
||
)
|
||
def delete_stock_transaction(stock_txn_id: int) -> dict[str, Any]:
|
||
"""Delete a stock transaction. Holdings are recomputed server-side after deletion.
|
||
|
||
Args:
|
||
stock_txn_id: Integer id of the stock transaction to delete.
|
||
"""
|
||
return {"result": client.delete_stock_transaction(stock_txn_id=stock_txn_id)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def create_rsu_grant(
|
||
account_id: int,
|
||
ticker: str,
|
||
vest_date: str,
|
||
gross_shares: float,
|
||
withholding_pct: float = 0.5,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Create an RSU grant record (does not vest shares — use vest_rsu_grant for that).
|
||
|
||
Args:
|
||
account_id: Integer id of an RSU account.
|
||
ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
|
||
vest_date: Planned vesting date in YYYY-MM-DD format.
|
||
gross_shares: Total gross shares in the grant (>0).
|
||
withholding_pct: Tax withholding fraction (0-1, default 0.5).
|
||
notes: Optional notes (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"account_id": account_id,
|
||
"ticker": ticker,
|
||
"vest_date": vest_date,
|
||
"gross_shares": gross_shares,
|
||
"withholding_pct": withholding_pct,
|
||
}
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
return {"result": client.create_rsu_grant(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
|
||
)
|
||
def delete_rsu_grant(grant_id: int) -> dict[str, Any]:
|
||
"""Delete an RSU grant. Already-vested grants are refused (API returns 422).
|
||
|
||
Args:
|
||
grant_id: Integer id of the RSU grant to delete.
|
||
"""
|
||
return {"result": client.delete_rsu_grant(grant_id=grant_id)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def vest_rsu_grant(
|
||
grant_id: int,
|
||
withholding_pct_override: float | None = None,
|
||
net_shares: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Vest an RSU grant, creating a stock transaction for the net shares received.
|
||
|
||
Args:
|
||
grant_id: Integer id of the RSU grant to vest.
|
||
withholding_pct_override: Override the grant's withholding percentage (0-1).
|
||
net_shares: Override the computed net shares (integer >=1).
|
||
"""
|
||
data: dict[str, Any] = {}
|
||
if withholding_pct_override is not None:
|
||
data["withholding_pct_override"] = withholding_pct_override
|
||
if net_shares is not None:
|
||
data["net_shares"] = net_shares
|
||
return {"result": client.vest_rsu_grant(grant_id=grant_id, data=data or None)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def generate_rsu_schedule(
|
||
account_id: int,
|
||
ticker: str,
|
||
grant_date: str,
|
||
total_gross_shares: float,
|
||
cliff_months: int = 12,
|
||
total_months: int = 36,
|
||
frequency_months: int = 3,
|
||
withholding_pct: float = 0.5,
|
||
notes: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Generate and persist the full RSU vesting schedule (creates grant records).
|
||
|
||
Use preview_rsu_schedule to see the schedule without saving anything.
|
||
|
||
Args:
|
||
account_id: RSU account integer id.
|
||
ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
|
||
grant_date: Grant date in YYYY-MM-DD format.
|
||
total_gross_shares: Total gross shares in the grant (>0).
|
||
cliff_months: Months until first vest (1-60, default 12).
|
||
total_months: Total vesting period in months (12-120, default 36).
|
||
frequency_months: Vesting frequency in months (1-12, default 3).
|
||
withholding_pct: Tax withholding fraction (0-1, default 0.5).
|
||
notes: Optional notes (max 500 chars).
|
||
"""
|
||
data: dict[str, Any] = {
|
||
"account_id": account_id,
|
||
"ticker": ticker,
|
||
"grant_date": grant_date,
|
||
"total_gross_shares": total_gross_shares,
|
||
"cliff_months": cliff_months,
|
||
"total_months": total_months,
|
||
"frequency_months": frequency_months,
|
||
"withholding_pct": withholding_pct,
|
||
}
|
||
if notes is not None:
|
||
data["notes"] = notes
|
||
return {"result": client.generate_rsu_schedule(data=data)}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def refresh_prices() -> dict[str, Any]:
|
||
"""Trigger a background refresh of all stock prices and FX rates.
|
||
|
||
Returns 202 immediately; the refresh runs asynchronously.
|
||
"""
|
||
return {"result": client.refresh_prices()}
|
||
|
||
@mcp.tool(
|
||
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
|
||
)
|
||
def fetch_ticker_price(ticker: str) -> dict[str, Any]:
|
||
"""Fetch and cache the current price for a single ticker symbol from yfinance.
|
||
|
||
WARNING: the upstream fetch is synchronous and retries internally.
|
||
Unknown tickers routinely exceed the client's 30s timeout and come
|
||
back as a request error rather than a result. Do not call in a loop.
|
||
|
||
Args:
|
||
ticker: Ticker symbol to fetch (e.g. "AAPL", "MSFT").
|
||
"""
|
||
return {"result": client.fetch_ticker_price(ticker=ticker)}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Resources — stable reference data every task needs. Exposing these
|
||
# alongside the list_* tools lets a client pull them into context once
|
||
# instead of spending a tool call per turn re-listing them.
|
||
# ------------------------------------------------------------------
|
||
|
||
@mcp.resource(
|
||
"fin://accounts",
|
||
name="accounts",
|
||
description=(
|
||
"All active fin accounts, with the integer id, type, currency and "
|
||
"balance of each. Look up account ids here before writing."
|
||
),
|
||
mime_type="application/json",
|
||
)
|
||
def accounts_resource() -> str:
|
||
return json.dumps(client.list_accounts(), indent=2, default=str)
|
||
|
||
@mcp.resource(
|
||
"fin://categories",
|
||
name="categories",
|
||
description=(
|
||
"All spending and income categories, with the integer id of each. "
|
||
"Look up category ids here before writing."
|
||
),
|
||
mime_type="application/json",
|
||
)
|
||
def categories_resource() -> str:
|
||
return json.dumps(client.list_categories(), indent=2, default=str)
|
||
|
||
return mcp
|
||
|
||
|
||
def main() -> None:
|
||
"""Entry point for the fin-mcp console script.
|
||
|
||
Transport comes from FIN_MCP_TRANSPORT (default "stdio"). Under
|
||
"streamable-http" the server binds FIN_MCP_HOST:FIN_MCP_PORT and serves
|
||
MCP at /mcp, which is the shape used when running it as a container.
|
||
"""
|
||
transport = os.getenv("FIN_MCP_TRANSPORT", "stdio").strip().lower()
|
||
if transport not in ("stdio", "streamable-http", "sse"):
|
||
raise SystemExit(
|
||
f"FIN_MCP_TRANSPORT={transport!r} is not valid; "
|
||
"expected one of: stdio, streamable-http, sse"
|
||
)
|
||
|
||
mcp = create_server()
|
||
|
||
if transport != "stdio":
|
||
mcp.settings.host = os.getenv("FIN_MCP_HOST", "0.0.0.0")
|
||
mcp.settings.port = int(os.getenv("FIN_MCP_PORT", "8000"))
|
||
mcp.settings.streamable_http_path = os.getenv("FIN_MCP_PATH", "/mcp")
|
||
# No session affinity in front of this, so keep every request
|
||
# self-contained rather than pinning clients to one replica.
|
||
mcp.settings.stateless_http = True
|
||
|
||
mcp.run(transport=transport)
|