You Need A Budget (YNAB) MCP

Instructions

Implementation Reference

• src/ynab_mcp_server/server.py:484-532 (handler)

  The core handler function within the @server.call_tool() decorator that processes the tool call, validates arguments using ManageScheduledTransactionInput, extracts budget_id, and executes create, update, or delete actions via ynab_client.
  elif name == "manage-scheduled-transaction": args = ManageScheduledTransactionInput.model_validate(arguments or {}) budget_id = await _get_budget_id(args.model_dump()) action = args.action if action == "create": transaction_data = { k: v for k, v in args.transaction_data.model_dump().items() if v is not None } if "amount" in transaction_data: transaction_data["amount"] = int(transaction_data["amount"]) await ynab_client.create_scheduled_transaction( budget_id=budget_id, transaction=SaveScheduledTransaction(**transaction_data) ) return [ types.TextContent( type="text", text="Successfully created scheduled transaction.", ) ] elif action == "update": transaction_data = { k: v for k, v in args.transaction_data.model_dump().items() if v is not None } if "amount" in transaction_data: transaction_data["amount"] = int(transaction_data["amount"]) await ynab_client.update_scheduled_transaction( budget_id=budget_id, transaction_id=args.transaction_id, transaction=SaveScheduledTransaction(**transaction_data), ) return [ types.TextContent( type="text", text=f"Successfully updated scheduled transaction {args.transaction_id}.", ) ] elif action == "delete": await ynab_client.delete_scheduled_transaction( budget_id=budget_id, transaction_id=args.transaction_id ) return [ types.TextContent( type="text", text=f"Successfully deleted scheduled transaction {args.transaction_id}.", ) ]
• src/ynab_mcp_server/tool_models.py:280-310 (schema)

  Pydantic model for input validation and schema definition used in tool registration and handler. Includes action enum reference, fields for id and data, and model validators for action-specific requirements.
  class ManageScheduledTransactionInput(BudgetIdInput): action: ManageScheduledTransactionAction = Field(..., description="The action to perform.") transaction_id: Optional[str] = Field(None, description="The ID of the scheduled transaction to update or delete.") transaction_data: Optional[ScheduledTransaction] = Field(None, description="The data for the scheduled transaction to create or update.") @model_validator(mode='before') @classmethod def check_fields_for_action(cls, values): action = values.get('action') if not action: raise ValueError("'action' is a required field.") transaction_id = values.get('transaction_id') transaction_data = values.get('transaction_data') if action == 'create': if not transaction_data: raise ValueError("'transaction_data' is required for the 'create' action.") if transaction_id: raise ValueError("'transaction_id' should not be provided for the 'create' action.") elif action == 'update': if not transaction_id or not transaction_data: raise ValueError("'transaction_id' and 'transaction_data' are required for the 'update' action.") elif action == 'delete': if not transaction_id: raise ValueError("'transaction_id' is required for the 'delete' action.") if transaction_data: raise ValueError("'transaction_data' should not be provided for the 'delete' action.") return values
• src/ynab_mcp_server/tool_models.py:253-272 (schema)

  Nested Pydantic model defining the structure of the scheduled transaction data used in create/update actions.
  class ScheduledTransaction(BaseModel): account_id: str = Field(..., description="The ID of the account for the transaction.") date: str = Field(..., description="The transaction date in YYYY-MM-DD format.") amount: float = Field(..., description="The transaction amount in milliunits.") frequency: str = Field(..., description="The frequency of the scheduled transaction (e.g. 'daily', 'weekly', 'monthly').") payee_id: Optional[str] = Field(None, description="The ID of the payee.") payee_name: Optional[str] = Field( None, description="The name of the payee. If not provided, a new payee will be created." ) category_id: Optional[str] = Field( None, description="The ID of the category for the transaction." ) memo: Optional[str] = Field(None, description="A memo for the transaction.") flag_color: Optional[str] = Field( None, description="The flag color of the transaction.", ) import_id: Optional[str] = Field( None, description="A unique import ID for the transaction. Use for idempotency." )
• src/ynab_mcp_server/server.py:110-113 (registration)

  Tool registration in the @server.list_tools() handler, defining name, description, and input schema reference.
  types.Tool( name="manage-scheduled-transaction", description="Create, update, or delete a single scheduled (recurring) transaction. Use this to manage recurring bills or savings transfers.", inputSchema=ManageScheduledTransactionInput.model_json_schema(),
• src/ynab_mcp_server/ynab_client.py:260-286 (helper)

  Supporting methods in YNABClient class that wrap the YNAB SDK API calls for creating, updating, and deleting scheduled transactions, invoked by the tool handler.
  async def create_scheduled_transaction( self, budget_id: str, transaction: SaveScheduledTransaction ): wrapper = PostScheduledTransactionWrapper(scheduled_transaction=transaction) return await self._run_sync( self._scheduled_transactions_api.create_scheduled_transaction, budget_id, wrapper, ) async def update_scheduled_transaction( self, budget_id: str, transaction_id: str, transaction: SaveScheduledTransaction ): wrapper = PutScheduledTransactionWrapper(scheduled_transaction=transaction) return await self._run_sync( self._scheduled_transactions_api.update_scheduled_transaction, budget_id, transaction_id, wrapper, ) async def delete_scheduled_transaction(self, budget_id: str, transaction_id: str): return await self._run_sync( self._scheduled_transactions_api.delete_scheduled_transaction, budget_id, transaction_id, )