main.py•3.92 kB
"""YNAB Assistant MCP server."""
from typing import Any
import ynab
from fastmcp import FastMCP
from mcp.types import ToolAnnotations
from accounts import get_account_by_id, get_accounts
from auth import get_api_client, get_api_configuration
from categories import (
get_categories,
get_category_by_id,
get_month_category_by_id,
update_category,
update_month_category,
)
from months import get_budget_month, get_budget_months
from transactions import (
create_transaction,
create_transactions,
delete_transaction,
get_transaction_by_id,
get_transactions,
get_transactions_by_account,
get_transactions_by_category,
get_transactions_by_month,
get_transactions_by_payee,
update_transaction,
update_transactions,
)
mcp = FastMCP("YNAB Assistant")
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
def get_user() -> dict[str, Any]:
"""Get authenticated YNAB user information.
Retrieves the authenticated user's YNAB profile information including user ID,
username, email, and account dates. Use this to verify authentication and get
basic user profile details.
Returns:
dict[str, Any]: User information containing id, username, email,
last_accessed_date, and created_date.
Raises:
Exception: If the API call fails or auth token is not set.
"""
configuration = get_api_configuration()
with get_api_client(configuration) as api_client:
api_instance = ynab.UserApi(api_client)
try:
api_response = api_instance.get_user()
user = api_response.data.user # type: ignore[attr-defined]
# Convert to dict to see what fields are actually available
user_dict = user.model_dump() # type: ignore[attr-defined]
# Return the user information dictionary
return user_dict
except Exception as e:
msg = f"Error fetching user info: {e!s}"
raise Exception(msg) from e
# Register account tools (read-only)
mcp.tool(get_accounts, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_account_by_id, annotations=ToolAnnotations(readOnlyHint=True))
# Register category tools
# Read operations
mcp.tool(get_categories, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_category_by_id, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_month_category_by_id, annotations=ToolAnnotations(readOnlyHint=True))
# Write operations
mcp.tool(update_category, annotations=ToolAnnotations(readOnlyHint=False))
mcp.tool(update_month_category, annotations=ToolAnnotations(readOnlyHint=False))
# Register month tools (read-only)
mcp.tool(get_budget_months, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_budget_month, annotations=ToolAnnotations(readOnlyHint=True))
# Register transaction tools
# Read operations
mcp.tool(get_transaction_by_id, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_transactions, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_transactions_by_account, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_transactions_by_category, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_transactions_by_month, annotations=ToolAnnotations(readOnlyHint=True))
mcp.tool(get_transactions_by_payee, annotations=ToolAnnotations(readOnlyHint=True))
# Write operations
mcp.tool(create_transaction, annotations=ToolAnnotations(readOnlyHint=False))
mcp.tool(create_transactions, annotations=ToolAnnotations(readOnlyHint=False))
mcp.tool(update_transaction, annotations=ToolAnnotations(readOnlyHint=False))
mcp.tool(update_transactions, annotations=ToolAnnotations(readOnlyHint=False))
# Destructive operation
mcp.tool(
delete_transaction,
annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True),
)
if __name__ == "__main__":
mcp.run(transport="http", port=8000)