Plaid Transactions MCP Server

#!/usr/bin/env python3 """ Plaid Transactions MCP Server. This MCP server provides tools to interact with Plaid's Transactions API, enabling transaction syncing, searching, and category management. """ from typing import Optional, List, Dict, Any from enum import Enum from datetime import datetime, timedelta import json import subprocess import httpx import asyncio import webbrowser import http.server import socketserver import urllib.parse import threading from pydantic import BaseModel, Field, field_validator, ConfigDict from mcp.server.fastmcp import FastMCP # Initialize the MCP server mcp = FastMCP("plaid_mcp") # Constants PLAID_API_URL = "https://production.plaid.com" # Change to sandbox for testing CHARACTER_LIMIT = 25000 # Maximum response size in characters # Enums class ResponseFormat(str, Enum): """Output format for tool responses.""" MARKDOWN = "markdown" JSON = "json" # ============================================================================ # Utility Functions # ============================================================================ def _get_keychain_value(service_name: str, account_name: Optional[str] = None) -> str: """ Retrieve a value from macOS Keychain. Args: service_name: The service name to look up in keychain account_name: Optional account name for the keychain entry Returns: The value stored in keychain Raises: Exception if keychain access fails """ try: cmd = ["security", "find-generic-password", "-s", service_name] if account_name: cmd.extend(["-a", account_name]) cmd.append("-w") result = subprocess.run( cmd, capture_output=True, text=True, check=True ) return result.stdout.strip() except subprocess.CalledProcessError as e: raise Exception( f"Failed to retrieve {service_name} from keychain. " f"Please store it using: security add-generic-password -s '{service_name}' -a 'plaid' -w 'your_value'" ) def _get_all_plaid_accounts() -> List[Dict[str, str]]: """ Retrieve all Plaid access tokens from keychain (PlaidTracker service). Returns: List of dictionaries with account_name and access_token """ try: # Dump keychain and search for PlaidTracker access tokens result = subprocess.run( ["security", "dump-keychain"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True ) accounts = [] lines = result.stdout.split('\n') i = 0 while i < len(lines): line = lines[i] # Look for PlaidTracker service marker if '0x00000007 <blob>="PlaidTracker"' in line: # Look forward for the account name (within next 5 lines) for j in range(i+1, min(len(lines), i+6)): if '"acct"<blob>=' in lines[j]: # Extract account name account_line = lines[j] if 'access_token_' in account_line: # Extract the account name start = account_line.find('"access_token_') + 1 end = account_line.find('"', start) account_name = account_line[start:end] # Get the actual token try: token = _get_keychain_value("PlaidTracker", account_name) # Extract friendly name # Format: access_token_<40+char_hash>_<Bank_Name_With_Underscores> # Strategy: Remove prefix, then split on first underscore to remove hash friendly_name = account_name.replace("access_token_", "") # Split on first underscore to separate hash from bank name if "_" in friendly_name: parts = friendly_name.split("_", 1) if len(parts) > 1: # Take everything after the hash and replace underscores with spaces friendly_name = parts[1].replace("_", " ") else: friendly_name = friendly_name.replace("_", " ") else: friendly_name = friendly_name.replace("_", " ") accounts.append({ "account_name": account_name, "friendly_name": friendly_name, "access_token": token }) except: pass break i += 1 return accounts except Exception as e: return [] def _get_access_token_for_account(account_identifier: str) -> str: """ Get access token for a specific account by name or friendly name. Args: account_identifier: Either the full account name or friendly name Returns: The access token Raises: Exception if account not found """ accounts = _get_all_plaid_accounts() # Try to match by friendly name first (case-insensitive) for account in accounts: if account_identifier.lower() in account["friendly_name"].lower(): return account["access_token"] # Try to match by full account name for account in accounts: if account_identifier in account["account_name"]: return account["access_token"] # Not found available = [acc["friendly_name"] for acc in accounts] raise Exception( f"Account '{account_identifier}' not found. Available accounts: {', '.join(available)}" ) async def _make_plaid_request( endpoint: str, method: str = "POST", data: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Make a request to the Plaid API with authentication. Args: endpoint: API endpoint path (e.g., '/transactions/sync') method: HTTP method (default: POST) data: Request payload Returns: JSON response from Plaid API Raises: httpx.HTTPStatusError: If the API request fails """ # Retrieve credentials from keychain client_id = _get_keychain_value("PlaidTracker", "plaid_client_id") secret = _get_keychain_value("PlaidTracker", "plaid_secret") # Add credentials to request body if data is None: data = {} data["client_id"] = client_id data["secret"] = secret async with httpx.AsyncClient() as client: response = await client.request( method, f"{PLAID_API_URL}{endpoint}", json=data, timeout=60.0 ) response.raise_for_status() return response.json() def _handle_plaid_error(e: Exception) -> str: """ Convert Plaid API errors into clear, actionable error messages. Args: e: The exception that occurred Returns: User-friendly error message with guidance """ if isinstance(e, httpx.HTTPStatusError): status = e.response.status_code try: error_body = e.response.json() error_code = error_body.get("error_code", "") error_message = error_body.get("error_message", "") # Handle specific Plaid error codes if error_code == "INVALID_ACCESS_TOKEN": return ( "Error: Invalid access token. The token may have expired or been revoked. " "You need to re-link the account via Plaid Link to get a new access token." ) elif error_code == "ITEM_LOGIN_REQUIRED": return ( "Error: User needs to re-authenticate via Plaid Link. " "The connection to the financial institution has expired and requires re-login." ) elif error_code == "INVALID_API_KEYS": return ( "Error: Invalid Plaid API credentials. Please verify your PLAID_CLIENT_ID and PLAID_SECRET " "are correct in keychain." ) elif "RATE_LIMIT" in error_code: return ( "Error: Rate limit exceeded. Plaid limits requests to 100/minute in production. " "Please wait 60 seconds before trying again." ) return f"Error: Plaid API error ({error_code}): {error_message}" except Exception: # Fallback if response isn't JSON if status == 401: return "Error: Authentication failed. Please verify your Plaid credentials in keychain." elif status == 404: return "Error: Item not found. The access token may be invalid." elif status == 429: return "Error: Rate limit exceeded. Please wait 60 seconds before retrying." return f"Error: API request failed with status {status}" elif isinstance(e, httpx.TimeoutException): return "Error: Request timed out. The Plaid API may be slow. Please try again." elif "keychain" in str(e).lower(): return str(e) # Return keychain error as-is (already has guidance) return f"Error: Unexpected error occurred: {type(e).__name__} - {str(e)}" async def _get_accounts_with_masks(access_token: str) -> Dict[str, Dict[str, Any]]: """ Fetch account details including masks (last 4 digits) from Plaid. Args: access_token: Plaid access token Returns: Dict mapping account_id to account info including mask """ try: response = await _make_plaid_request("/accounts/get", data={ "access_token": access_token }) accounts = response.get("accounts", []) return { acc["account_id"]: { "mask": acc.get("mask", ""), "name": acc.get("name", ""), "official_name": acc.get("official_name", ""), "type": acc.get("type", ""), "subtype": acc.get("subtype", "") } for acc in accounts } except Exception: return {} def _format_transaction_markdown(txn: Dict[str, Any], account_masks: Dict[str, Dict[str, Any]] = None) -> str: """ Format a single transaction as markdown. Args: txn: Transaction data from Plaid account_masks: Optional dict mapping account_id to account info with mask Returns: Markdown-formatted transaction string """ lines = [] # Amount and merchant name amount = txn.get("amount", 0) name = txn.get("name", "Unknown") lines.append(f"### ${amount:.2f} - {name}") # Date date = txn.get("date", "") if date: lines.append(f"- **Date**: {date}") # Category category = txn.get("category", []) if category: category_str = " > ".join(category) lines.append(f"- **Category**: {category_str}") # Merchant name (if different from name) merchant_name = txn.get("merchant_name") if merchant_name and merchant_name != name: lines.append(f"- **Merchant**: {merchant_name}") # Account - show actual mask (last 4 digits) if available account_id = txn.get("account_id", "") if account_id: if account_masks and account_id in account_masks: mask = account_masks[account_id].get("mask", "") acc_name = account_masks[account_id].get("name", "") if mask: lines.append(f"- **Account**: ****{mask} ({acc_name})") else: lines.append(f"- **Account**: {acc_name}") else: lines.append(f"- **Account**: {account_id[-4:]}") # Fallback to last 4 chars of ID # Payment channel payment_channel = txn.get("payment_channel", "") if payment_channel: lines.append(f"- **Payment Channel**: {payment_channel}") # Pending status if txn.get("pending"): lines.append(f"- **Status**: Pending") # Transaction ID transaction_id = txn.get("transaction_id", "") if transaction_id: lines.append(f"- **ID**: {transaction_id}") lines.append("") # Empty line after each transaction return "\n".join(lines) def _truncate_response(response: str, data_description: str = "results") -> str: """ Truncate response if it exceeds CHARACTER_LIMIT and add helpful message. Args: response: The full response string data_description: Description of what's being truncated Returns: Truncated response with guidance message if needed """ if len(response) <= CHARACTER_LIMIT: return response # Truncate to 50% of limit to leave room for message truncated = response[:CHARACTER_LIMIT // 2] truncation_msg = ( f"\n\n---\n\n" f"⚠️ **Response Truncated**\n\n" f"The full response exceeded {CHARACTER_LIMIT:,} characters and was truncated. " f"To see more {data_description}, try:\n" f"- Use date_start/date_end to narrow the date range\n" f"- Filter by specific account_ids\n" f"- Use plaid_search_transactions to filter by merchant, category, or amount\n" ) return truncated + truncation_msg # ============================================================================ # Pydantic Input Models # ============================================================================ class TransactionSyncInput(BaseModel): """Input model for syncing transactions.""" model_config = ConfigDict( str_strip_whitespace=True, validate_assignment=True, extra='forbid' ) access_token: Optional[str] = Field( default=None, description="Plaid access token for the Item. If not provided, use account_name instead.", min_length=1 ) account_name: Optional[str] = Field( default=None, description="Friendly account name (e.g., 'American Express', 'Citibank'). Use plaid_list_accounts to see available accounts." ) cursor: Optional[str] = Field( default=None, description="Cursor for pagination (omit for initial sync). Example: 'eyJsYXN0X3VwZGF0ZWRfZGF0ZXRpbWUiOiIyMDIwLTA3LTI0VDE4OjQ5OjQ3WiJ9'" ) count: Optional[int] = Field( default=100, description="Number of transactions to fetch per page (default: 100, max: 500)", ge=1, le=500 ) account_ids: Optional[List[str]] = Field( default=None, description="Filter transactions to specific account IDs. Example: ['aAbBcC123dDeEfF456']" ) date_start: Optional[str] = Field( default=None, description="Start date for filtering transactions (YYYY-MM-DD format). Example: '2024-01-01'" ) date_end: Optional[str] = Field( default=None, description="End date for filtering transactions (YYYY-MM-DD format). Example: '2024-12-31'" ) response_format: ResponseFormat = Field( default=ResponseFormat.MARKDOWN, description="Output format: 'markdown' for human-readable or 'json' for machine-readable" ) class SearchTransactionsInput(BaseModel): """Input model for searching/filtering transactions.""" model_config = ConfigDict( str_strip_whitespace=True, validate_assignment=True, extra='forbid' ) access_token: Optional[str] = Field( default=None, description="Plaid access token for the Item. If not provided, use account_name instead.", min_length=1 ) account_name: Optional[str] = Field( default=None, description="Friendly account name (e.g., 'American Express', 'Citibank'). Use plaid_list_accounts to see available accounts." ) account_mask: Optional[str] = Field( default=None, description="Filter by card's last 4 digits (mask). Example: '3620', '1234'. Use plaid_list_accounts to see available masks." ) merchant_name: Optional[str] = Field( default=None, description="Filter by merchant name (partial match). Example: 'Starbucks', 'Amazon'" ) category: Optional[str] = Field( default=None, description="Filter by category (partial match). Example: 'Food and Drink', 'Travel'" ) min_amount: Optional[float] = Field( default=None, description="Minimum transaction amount. Example: 50.0" ) max_amount: Optional[float] = Field( default=None, description="Maximum transaction amount. Example: 500.0" ) date_start: Optional[str] = Field( default=None, description="Start date (YYYY-MM-DD). Example: '2024-01-01'" ) date_end: Optional[str] = Field( default=None, description="End date (YYYY-MM-DD). Example: '2024-12-31'" ) account_ids: Optional[List[str]] = Field( default=None, description="Filter to specific account IDs" ) response_format: ResponseFormat = Field( default=ResponseFormat.MARKDOWN, description="Output format: 'markdown' for human-readable or 'json' for machine-readable" ) # ============================================================================ # Tool Implementations # ============================================================================ @mcp.tool( name="plaid_list_accounts", annotations={ "title": "List Available Plaid Accounts", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False } ) async def plaid_list_accounts() -> str: """ List all available Plaid accounts stored in keychain. This tool retrieves all Plaid access tokens from your keychain and displays the account names. Use these names with other tools to specify which account to query. Returns: str: List of available accounts in Markdown format containing: - Friendly account names (e.g., "American Express", "Citibank Online") - Number of accounts found - Instructions on how to use account names with other tools Examples: - Use when: "What Plaid accounts do I have linked?" - Use when: "Show me my available accounts" - Always use this first to see which accounts you can query Error Handling: - Returns empty list if no accounts found in keychain - Returns error if keychain access fails """ try: accounts = _get_all_plaid_accounts() if not accounts: return ( "# No Plaid Accounts Found\n\n" "No Plaid access tokens found in keychain under 'PlaidTracker' service.\n" "Please ensure your accounts are linked and tokens are stored in keychain." ) lines = ["# Available Plaid Accounts", ""] lines.append(f"Found **{len(accounts)}** linked institution(s):") lines.append("") for i, account in enumerate(accounts, 1): lines.append(f"## {i}. {account['friendly_name']}") lines.append("") # Fetch sub-accounts (individual cards/accounts) with their masks try: sub_accounts = await _get_accounts_with_masks(account['access_token']) if sub_accounts: lines.append("| Card/Account | Type | Last 4 |") lines.append("|--------------|------|--------|") for acc_id, acc_info in sub_accounts.items(): acc_name = acc_info.get('name', 'Unknown') acc_type = acc_info.get('subtype', acc_info.get('type', '')) mask = acc_info.get('mask', 'N/A') lines.append(f"| {acc_name} | {acc_type} | **{mask}** |") lines.append("") else: lines.append("_No sub-accounts found_") lines.append("") except Exception: lines.append("_Could not fetch sub-account details_") lines.append("") lines.append("## How to Use") lines.append("") lines.append("Use these account names with other tools:") lines.append("- `plaid_sync_transactions` - Use `account_name` parameter") lines.append("- `plaid_search_transactions` - Use `account_name` parameter") lines.append("") lines.append("**Example**: To get transactions from American Express:") lines.append('```\naccount_name: "American Express"\n```') return "\n".join(lines) except Exception as e: return f"Error listing accounts: {str(e)}" @mcp.tool( name="plaid_sync_transactions", annotations={ "title": "Sync Plaid Transactions", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def plaid_sync_transactions(params: TransactionSyncInput) -> str: """ Sync all transactions from Plaid using cursor-based pagination. This tool retrieves transaction data including added, modified, and removed transactions since the last cursor position. It automatically handles pagination and can filter by accounts and date ranges. Args: params (TransactionSyncInput): Validated input parameters containing: - access_token (str): Plaid access token for the Item (required) - cursor (Optional[str]): Pagination cursor from previous sync (omit for initial sync) - count (Optional[int]): Transactions per page (default: 100, max: 500) - account_ids (Optional[List[str]]): Filter to specific accounts - date_start (Optional[str]): Filter start date (YYYY-MM-DD) - date_end (Optional[str]): Filter end date (YYYY-MM-DD) - response_format (ResponseFormat): Output format (default: markdown) Returns: str: Formatted response containing: Markdown format: - Human-readable transaction list grouped by added/modified/removed - Transaction details including amount, merchant, date, category - Pagination info and next cursor JSON format: { "added": [...], // New transactions "modified": [...], // Updated transactions "removed": [...], // Removed transaction IDs "next_cursor": str, // Use for next sync "has_more": bool, // More data available "total_added": int, "total_modified": int, "total_removed": int } Examples: - Use when: "Show me all my recent transactions" - Use when: "Sync my Plaid transactions from December 2024" - Use when: "Get transactions for my checking account" - Don't use when: Searching for specific merchants (use plaid_search_transactions) Error Handling: - Returns "Error: Invalid access token..." if token is expired - Returns "Error: Rate limit exceeded..." if too many requests - Returns "Error: User needs to re-authenticate..." if login required - Automatically truncates large responses with filtering guidance """ try: # Resolve access token from account name if needed access_token = params.access_token if not access_token and params.account_name: access_token = _get_access_token_for_account(params.account_name) elif not access_token: return "Error: Either access_token or account_name must be provided. Use plaid_list_accounts to see available accounts." # Build request payload request_data = { "access_token": access_token, "count": params.count } if params.cursor: request_data["cursor"] = params.cursor # Build options object options = {} if params.account_ids: options["account_ids"] = params.account_ids # Calculate days_requested for historical data # Note: Plaid's days_requested is always relative to TODAY, not an absolute range if params.date_start: start_date = datetime.strptime(params.date_start, "%Y-%m-%d") # Calculate days from start_date to TODAY to ensure we fetch enough historical data today = datetime.now() days_from_start = (today - start_date).days + 1 # Plaid max is 730 days (2 years) if days_from_start > 730: # Can only fetch last 730 days, will filter client-side options["days_requested"] = 730 else: options["days_requested"] = days_from_start elif params.date_end: # If only end date specified, fetch up to that point from today end_date = datetime.strptime(params.date_end, "%Y-%m-%d") today = datetime.now() days_to_end = (today - end_date).days if days_to_end < 0: # Future date, fetch default options["days_requested"] = 90 else: options["days_requested"] = min(days_to_end + 90, 730) if options: request_data["options"] = options # Fetch all pages automatically all_added = [] all_modified = [] all_removed = [] current_cursor = params.cursor has_more = True while has_more: if current_cursor: request_data["cursor"] = current_cursor response = await _make_plaid_request("/transactions/sync", data=request_data) all_added.extend(response.get("added", [])) all_modified.extend(response.get("modified", [])) all_removed.extend(response.get("removed", [])) has_more = response.get("has_more", False) current_cursor = response.get("next_cursor") # Fetch account masks (real last 4 digits) for display account_masks = await _get_accounts_with_masks(access_token) # Format response if params.response_format == ResponseFormat.MARKDOWN: lines = ["# Plaid Transaction Sync Results", ""] # Summary lines.append(f"**Added**: {len(all_added)} transactions") lines.append(f"**Modified**: {len(all_modified)} transactions") lines.append(f"**Removed**: {len(all_removed)} transactions") lines.append("") if current_cursor: lines.append(f"**Next Cursor**: `{current_cursor}`") lines.append("(Save this cursor for incremental syncs)") lines.append("") # Added transactions if all_added: lines.append("## Added Transactions") lines.append("") for txn in all_added: lines.append(_format_transaction_markdown(txn, account_masks)) # Modified transactions if all_modified: lines.append("## Modified Transactions") lines.append("") for txn in all_modified: lines.append(_format_transaction_markdown(txn, account_masks)) # Removed transactions if all_removed: lines.append("## Removed Transactions") lines.append("") for removed in all_removed: transaction_id = removed.get("transaction_id", "Unknown") lines.append(f"- {transaction_id}") lines.append("") if not all_added and not all_modified and not all_removed: lines.append("No new transaction updates found.") result = "\n".join(lines) return _truncate_response(result, "transactions") else: # JSON format result = { "added": all_added, "modified": all_modified, "removed": all_removed, "next_cursor": current_cursor, "has_more": False, # We fetched all pages "total_added": len(all_added), "total_modified": len(all_modified), "total_removed": len(all_removed) } json_str = json.dumps(result, indent=2) return _truncate_response(json_str, "transactions") except Exception as e: return _handle_plaid_error(e) @mcp.tool( name="plaid_get_transaction_categories", annotations={ "title": "Get Plaid Transaction Categories", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def plaid_get_transaction_categories() -> str: """ Retrieve all available transaction categories from Plaid. This tool returns the complete taxonomy of transaction categories that Plaid uses to categorize transactions. Useful for understanding categorization and filtering. Returns: str: Formatted list of categories: Markdown format: - Hierarchical list of categories with descriptions - Grouped by top-level category JSON format: { "categories": [ { "category_id": str, "hierarchy": [str], // e.g., ["Food and Drink", "Restaurants", "Coffee Shop"] "group": str } ] } Examples: - Use when: "What categories does Plaid use for transactions?" - Use when: "Show me all food-related categories" - Don't use when: You need actual transaction data (use plaid_sync_transactions) Error Handling: - Returns error if API request fails - No authentication required for this endpoint """ try: # Note: /categories/get doesn't require authentication response = await _make_plaid_request("/categories/get", data={}) categories = response.get("categories", []) # Format as markdown lines = ["# Plaid Transaction Categories", ""] lines.append(f"Total categories: {len(categories)}") lines.append("") # Group by top-level category grouped = {} for cat in categories: hierarchy = cat.get("hierarchy", []) if hierarchy: top_level = hierarchy[0] if top_level not in grouped: grouped[top_level] = [] grouped[top_level].append(cat) # Display grouped for top_level, cats in sorted(grouped.items()): lines.append(f"## {top_level}") lines.append("") for cat in cats: hierarchy = cat.get("hierarchy", []) category_id = cat.get("category_id", "") full_path = " > ".join(hierarchy) lines.append(f"- **{full_path}** (`{category_id}`)") lines.append("") return "\n".join(lines) except Exception as e: return _handle_plaid_error(e) @mcp.tool( name="plaid_search_transactions", annotations={ "title": "Search/Filter Plaid Transactions", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def plaid_search_transactions(params: SearchTransactionsInput) -> str: """ Search and filter transactions by merchant, category, amount, or date. This tool first syncs all transactions, then intelligently filters them based on your search criteria. Supports partial matching for merchant names and categories. Args: params (SearchTransactionsInput): Validated input parameters containing: - access_token (str): Plaid access token (required) - merchant_name (Optional[str]): Filter by merchant (partial match) - category (Optional[str]): Filter by category (partial match) - min_amount (Optional[float]): Minimum transaction amount - max_amount (Optional[float]): Maximum transaction amount - date_start (Optional[str]): Start date (YYYY-MM-DD) - date_end (Optional[str]): End date (YYYY-MM-DD) - account_ids (Optional[List[str]]): Filter to specific accounts - response_format (ResponseFormat): Output format (default: markdown) Returns: str: Filtered transactions in specified format Markdown format: - Human-readable list of matching transactions - Summary of filter criteria and match count JSON format: { "transactions": [...], "total_matches": int, "filters_applied": {...} } Examples: - Use when: "Find all Starbucks transactions" - Use when: "Show me transactions over $100 in December" - Use when: "List all travel expenses from last quarter" - Don't use when: You just want to sync all transactions (use plaid_sync_transactions) Error Handling: - Returns error if sync fails - Returns "No transactions match your criteria" if no results - Automatically truncates large result sets with filtering guidance """ try: # Resolve access token from account name if needed access_token = params.access_token if not access_token and params.account_name: access_token = _get_access_token_for_account(params.account_name) elif not access_token: return "Error: Either access_token or account_name must be provided. Use plaid_list_accounts to see available accounts." # Build request payload for sync request_data = { "access_token": access_token, "count": 500 # Fetch max per page } # Build options object options = {} if params.account_ids: options["account_ids"] = params.account_ids # Calculate days_requested for historical data # Note: Plaid's days_requested is always relative to TODAY, not an absolute range if params.date_start: start_date = datetime.strptime(params.date_start, "%Y-%m-%d") # Calculate days from start_date to TODAY to ensure we fetch enough historical data today = datetime.now() days_from_start = (today - start_date).days + 1 # Plaid max is 730 days (2 years) if days_from_start > 730: # Can only fetch last 730 days, will filter client-side options["days_requested"] = 730 else: options["days_requested"] = days_from_start elif params.date_end: # If only end date specified, fetch up to that point from today end_date = datetime.strptime(params.date_end, "%Y-%m-%d") today = datetime.now() days_to_end = (today - end_date).days if days_to_end < 0: # Future date, fetch default options["days_requested"] = 90 else: options["days_requested"] = min(days_to_end + 90, 730) if options: request_data["options"] = options # Fetch all transactions directly (work with raw data, not serialized strings) all_added = [] all_modified = [] current_cursor = None has_more = True while has_more: if current_cursor: request_data["cursor"] = current_cursor response = await _make_plaid_request("/transactions/sync", data=request_data) all_added.extend(response.get("added", [])) all_modified.extend(response.get("modified", [])) has_more = response.get("has_more", False) current_cursor = response.get("next_cursor") all_transactions = all_added + all_modified # Fetch account masks (real last 4 digits) for filtering and display account_masks = await _get_accounts_with_masks(access_token) # Build reverse mapping: mask -> list of account_ids (in case multiple accounts have same mask) mask_to_account_ids = {} for acc_id, acc_info in account_masks.items(): mask = acc_info.get("mask", "") if mask: if mask not in mask_to_account_ids: mask_to_account_ids[mask] = [] mask_to_account_ids[mask].append(acc_id) # If account_mask filter is provided, find matching account_ids target_account_ids = None if params.account_mask: target_account_ids = mask_to_account_ids.get(params.account_mask, []) if not target_account_ids: # Try partial match for mask, acc_ids in mask_to_account_ids.items(): if params.account_mask in mask or mask in params.account_mask: target_account_ids = acc_ids break if not target_account_ids: available_masks = list(mask_to_account_ids.keys()) return f"Error: No account found with mask '{params.account_mask}'. Available masks: {', '.join(available_masks)}" # Apply filters filtered = [] for txn in all_transactions: # Account mask filter (by account_id) if target_account_ids: if txn.get("account_id") not in target_account_ids: continue # Merchant name filter if params.merchant_name: name = (txn.get("name") or "").lower() merchant_name = (txn.get("merchant_name") or "").lower() search_term = params.merchant_name.lower() if search_term not in name and search_term not in merchant_name: continue # Category filter if params.category: categories = txn.get("category") or [] category_str = " ".join(categories).lower() if categories else "" if params.category.lower() not in category_str: continue # Amount filters amount = txn.get("amount", 0) if params.min_amount is not None and amount < params.min_amount: continue if params.max_amount is not None and amount > params.max_amount: continue # Date filters (already applied in sync, but double-check) if params.date_start: txn_date = txn.get("date", "") if txn_date < params.date_start: continue if params.date_end: txn_date = txn.get("date", "") if txn_date > params.date_end: continue filtered.append(txn) # Format response (account_masks already fetched above for filtering) if params.response_format == ResponseFormat.MARKDOWN: lines = ["# Transaction Search Results", ""] # Show filters applied lines.append("## Filters Applied") if params.account_mask: lines.append(f"- **Card (last 4)**: {params.account_mask}") if params.merchant_name: lines.append(f"- **Merchant**: {params.merchant_name}") if params.category: lines.append(f"- **Category**: {params.category}") if params.min_amount is not None: lines.append(f"- **Min Amount**: ${params.min_amount:.2f}") if params.max_amount is not None: lines.append(f"- **Max Amount**: ${params.max_amount:.2f}") if params.date_start: lines.append(f"- **Date Start**: {params.date_start}") if params.date_end: lines.append(f"- **Date End**: {params.date_end}") lines.append("") lines.append(f"**Found {len(filtered)} matching transactions** (out of {len(all_transactions)} total)") lines.append("") # Calculate total spend for this filter total_spend = sum(txn.get("amount", 0) for txn in filtered) lines.append(f"**Total**: ${total_spend:,.2f}") lines.append("") if not filtered: lines.append("No transactions match your search criteria. Try adjusting your filters.") return "\n".join(lines) # Display transactions lines.append("## Matching Transactions") lines.append("") for txn in filtered: lines.append(_format_transaction_markdown(txn, account_masks)) result = "\n".join(lines) return _truncate_response(result, "matching transactions") else: # JSON format total_spend = sum(txn.get("amount", 0) for txn in filtered) result = { "transactions": filtered, "total_matches": len(filtered), "total_spend": total_spend, "total_scanned": len(all_transactions), "filters_applied": { "account_mask": params.account_mask, "merchant_name": params.merchant_name, "category": params.category, "min_amount": params.min_amount, "max_amount": params.max_amount, "date_start": params.date_start, "date_end": params.date_end } } json_str = json.dumps(result, indent=2) return _truncate_response(json_str, "matching transactions") except Exception as e: return _handle_plaid_error(e) # ============================================================================ # Plaid Link Portal Tool # ============================================================================ # Global state for Link callback _link_received_public_token: Optional[str] = None _link_server_should_stop = threading.Event() _link_redirect_port = 8765 class _PlaidLinkCallbackHandler(http.server.SimpleHTTPRequestHandler): """Handle the OAuth callback from Plaid Link.""" def log_message(self, format, *args): """Suppress default logging.""" pass def do_GET(self): global _link_received_public_token parsed = urllib.parse.urlparse(self.path) if parsed.path == "/callback": params = urllib.parse.parse_qs(parsed.query) if "public_token" in params: _link_received_public_token = params["public_token"][0] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b""" <html> <head><title>Success</title></head> <body style="font-family: system-ui; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; background: #f0f0f0;"> <div style="text-align: center; background: white; padding: 40px; border-radius: 12px; box-shadow: 0 2px 10px rgba(0,0,0,0.1);"> <h1 style="color: #22c55e;">Account Linked Successfully!</h1> <p>You can close this window and return to Claude.</p> </div> </body> </html> """) _link_server_should_stop.set() else: self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b""" <html> <head><title>Cancelled</title></head> <body style="font-family: system-ui; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; background: #f0f0f0;"> <div style="text-align: center; background: white; padding: 40px; border-radius: 12px; box-shadow: 0 2px 10px rgba(0,0,0,0.1);"> <h1 style="color: #ef4444;">Linking Cancelled</h1> <p>You can close this window.</p> </div> </body> </html> """) _link_server_should_stop.set() elif parsed.path == "/link": link_token = getattr(self.server, 'link_token', '') self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() html = f""" <!DOCTYPE html> <html> <head> <title>Link Bank Account</title> <script src="https://cdn.plaid.com/link/v2/stable/link-initialize.js"></script> </head> <body style="font-family: system-ui; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; background: #f0f0f0;"> <div style="text-align: center; background: white; padding: 40px; border-radius: 12px; box-shadow: 0 2px 10px rgba(0,0,0,0.1);"> <h1>Link Your Bank Account</h1> <p>Click the button below to securely connect your account.</p> <button id="link-button" style="background: #0066ff; color: white; border: none; padding: 12px 24px; border-radius: 6px; font-size: 16px; cursor: pointer; margin-top: 20px;"> Connect Account </button> </div> <script> const handler = Plaid.create({{ token: '{link_token}', onSuccess: (public_token, metadata) => {{ window.location.href = '/callback?public_token=' + public_token; }}, onExit: (err, metadata) => {{ if (err) {{ console.error(err); }} window.location.href = '/callback?cancelled=true'; }}, }}); document.getElementById('link-button').onclick = () => handler.open(); setTimeout(() => handler.open(), 500); </script> </body> </html> """ self.wfile.write(html.encode()) else: self.send_response(404) self.end_headers() def _run_link_server(link_token: str): """Run the temporary callback server for Plaid Link.""" with socketserver.TCPServer(("", _link_redirect_port), _PlaidLinkCallbackHandler) as httpd: httpd.link_token = link_token httpd.timeout = 1 while not _link_server_should_stop.is_set(): httpd.handle_request() async def _create_link_token() -> str: """Create a Plaid Link token.""" response = await _make_plaid_request("/link/token/create", data={ "user": {"client_user_id": "plaid-mcp-user"}, "client_name": "Plaid MCP", "products": ["transactions"], "country_codes": ["US"], "language": "en" }) if "link_token" not in response: error_msg = response.get("error_message", "Unknown error") raise Exception(f"Failed to create link token: {error_msg}") return response["link_token"] async def _exchange_public_token(public_token: str) -> tuple: """Exchange public token for access token. Returns (access_token, item_id).""" response = await _make_plaid_request("/item/public_token/exchange", data={ "public_token": public_token }) if "access_token" not in response: error_msg = response.get("error_message", "Unknown error") raise Exception(f"Failed to exchange token: {error_msg}") return response["access_token"], response["item_id"] async def _get_institution_name(access_token: str) -> str: """Get the institution name for a linked item.""" # Get item info client_id = _get_keychain_value("PlaidTracker", "plaid_client_id") secret = _get_keychain_value("PlaidTracker", "plaid_secret") async with httpx.AsyncClient() as client: item_response = await client.post( f"{PLAID_API_URL}/item/get", json={"client_id": client_id, "secret": secret, "access_token": access_token}, timeout=60.0 ) item_data = item_response.json() institution_id = item_data.get("item", {}).get("institution_id", "") if not institution_id: return "Unknown_Bank" inst_response = await client.post( f"{PLAID_API_URL}/institutions/get_by_id", json={ "client_id": client_id, "secret": secret, "institution_id": institution_id, "country_codes": ["US"] }, timeout=60.0 ) inst_data = inst_response.json() name = inst_data.get("institution", {}).get("name", "Unknown_Bank") return name.replace(" ", "_") def _store_keychain_value(service_name: str, account_name: str, value: str) -> bool: """Store a value in macOS Keychain. Updates if exists, creates if not.""" subprocess.run( ["security", "delete-generic-password", "-s", service_name, "-a", account_name], capture_output=True ) result = subprocess.run( ["security", "add-generic-password", "-s", service_name, "-a", account_name, "-w", value], capture_output=True, text=True ) return result.returncode == 0 @mcp.tool( name="plaid_add_account", annotations={ "title": "Add New Bank Account via Plaid Link", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def plaid_add_account() -> str: """ Launch Plaid Link to connect a new bank account. This tool opens your browser to Plaid's secure Link interface where you can connect a new bank account. The access token will be automatically stored in your macOS Keychain. Returns: str: Status message indicating success or failure Examples: - Use when: "Add a new bank account" - Use when: "Link a new credit card" - Use when: "Connect my Chase account" Notes: - Opens a browser window for secure bank authentication - Waits up to 5 minutes for you to complete the flow - Stores credentials securely in Keychain """ global _link_received_public_token, _link_server_should_stop try: # Reset state _link_received_public_token = None _link_server_should_stop.clear() # Verify credentials try: _get_keychain_value("PlaidTracker", "plaid_client_id") _get_keychain_value("PlaidTracker", "plaid_secret") except Exception: return ( "# Error: Missing Plaid Credentials\n\n" "Please ensure your Plaid credentials are stored in Keychain:\n" "```\n" 'security add-generic-password -s "PlaidTracker" -a "plaid_client_id" -w "your_client_id"\n' 'security add-generic-password -s "PlaidTracker" -a "plaid_secret" -w "your_secret"\n' "```" ) # Create link token link_token = await _create_link_token() # Start callback server in background thread server_thread = threading.Thread( target=_run_link_server, args=(link_token,), daemon=True ) server_thread.start() # Give server time to start await asyncio.sleep(0.5) # Open browser webbrowser.open(f"http://localhost:{_link_redirect_port}/link") # Wait for callback (max 5 minutes) timeout = 300 # 5 minutes start_time = asyncio.get_event_loop().time() while not _link_server_should_stop.is_set(): await asyncio.sleep(0.5) if asyncio.get_event_loop().time() - start_time > timeout: _link_server_should_stop.set() return "# Timeout\n\nThe Plaid Link session timed out after 5 minutes. Please try again." # Check if we got a token if not _link_received_public_token: return "# Cancelled\n\nNo account was linked. The Plaid Link flow was cancelled or closed." # Exchange token access_token, item_id = await _exchange_public_token(_link_received_public_token) # Get institution name institution_name = await _get_institution_name(access_token) # Store in keychain account_name = f"access_token_{item_id}_{institution_name}" if _store_keychain_value("PlaidTracker", account_name, access_token): friendly_name = institution_name.replace("_", " ") return ( f"# Account Linked Successfully!\n\n" f"**Institution**: {friendly_name}\n\n" f"Your new account is now available. Use `plaid_list_accounts` to see all linked accounts." ) else: return "# Error\n\nFailed to store access token in Keychain. Please try again." except Exception as e: return _handle_plaid_error(e) # ============================================================================ # Main Entry Point # ============================================================================ if __name__ == "__main__": mcp.run()