create_transaction
Add a new transaction in YNAB by specifying account ID, amount, payee, and optional details like category and memo using the MCP YNAB Server.
Instructions
Create a new transaction in YNAB.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| account_id | Yes | ||
| amount | Yes | Amount in dollars | |
| category_name | No | ||
| memo | No | ||
| payee_name | Yes |
Implementation Reference
- src/mcp_ynab/server.py:309-349 (handler)The main handler function for the 'create_transaction' tool. It is decorated with @mcp.tool() which registers it and defines the schema from type annotations. Creates a new transaction in YNAB using the TransactionsApi.
@mcp.tool() async def create_transaction( account_id: str, amount: Annotated[float, Field(description="Amount in dollars")], payee_name: str, category_name: Optional[str] = None, memo: Optional[str] = None, ) -> Dict[str, Any]: """Create a new transaction in YNAB.""" async with await get_ynab_client() as client: transactions_api = TransactionsApi(client) budgets_api = BudgetsApi(client) amount_milliunits = int(amount * 1000) # Use preferred budget ID if available, otherwise fetch a list of budgets budget_id = ynab_resources.get_preferred_budget_id() if not budget_id: budgets_response = budgets_api.get_budgets() budget_id = budgets_response.data.budgets[0].id category_id = None if category_name: category_id = await _find_category_id(client, budget_id, category_name) # Create transaction data transaction = NewTransaction( account_id=account_id, date=date.today(), amount=amount_milliunits, payee_name=payee_name, memo=memo, category_id=category_id, ) wrapper = PostTransactionsWrapper(transaction=transaction) response = transactions_api.create_transaction(budget_id, wrapper) if response.data and response.data.transaction: return response.data.transaction.to_dict() return {} - src/mcp_ynab/server.py:297-307 (helper)Helper utility function used by create_transaction to resolve category_name to a category_id by searching through category groups.
async def _find_category_id(client: ApiClient, budget_id: str, category_name: str) -> Optional[str]: """Find a category ID by name.""" categories_api = CategoriesApi(client) categories_response = categories_api.get_categories(budget_id) categories = categories_response.data.category_groups for group in categories: for cat in group.categories: if cat.name.lower() == category_name.lower(): return cat.id return None