Plaid Transactions MCP Server

#!/usr/bin/env python3 """ Plaid Account Linker - Securely add new bank accounts to your Plaid MCP. This script: 1. Reads your Plaid credentials from macOS Keychain (never from files) 2. Opens Plaid Link in your browser 3. Stores the access token securely in Keychain Usage: python add_account.py """ import subprocess import sys import json import webbrowser import http.server import socketserver import urllib.parse import threading import time from typing import Optional, Tuple import signal # Plaid API configuration PLAID_API_URL = "https://production.plaid.com" REDIRECT_PORT = 8765 REDIRECT_URI = f"http://localhost:{REDIRECT_PORT}/callback" # Global to capture the public token from callback received_public_token: Optional[str] = None server_should_stop = threading.Event() def get_keychain_value(service_name: str, account_name: Optional[str] = None) -> str: """Retrieve a value from macOS Keychain.""" 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) if result.returncode != 0: raise Exception(f"Failed to retrieve {service_name} from keychain") return result.stdout.strip() 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.""" # First try to delete existing entry (ignore errors if doesn't exist) subprocess.run( ["security", "delete-generic-password", "-s", service_name, "-a", account_name], capture_output=True ) # Add the new entry result = subprocess.run( ["security", "add-generic-password", "-s", service_name, "-a", account_name, "-w", value], capture_output=True, text=True ) return result.returncode == 0 def plaid_request(endpoint: str, data: dict) -> dict: """Make a request to Plaid API using curl (avoids dependency issues).""" import json # Get credentials from keychain client_id = get_keychain_value("PLAID_CLIENT_ID") secret = get_keychain_value("PLAID_SECRET") data["client_id"] = client_id data["secret"] = secret # Use curl for the request result = subprocess.run( [ "curl", "-s", "-X", "POST", f"{PLAID_API_URL}{endpoint}", "-H", "Content-Type: application/json", "-d", json.dumps(data) ], capture_output=True, text=True ) if result.returncode != 0: raise Exception(f"API request failed: {result.stderr}") return json.loads(result.stdout) def create_link_token() -> str: """Create a Plaid Link token.""" response = plaid_request("/link/token/create", { "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"] def exchange_public_token(public_token: str) -> Tuple[str, str]: """Exchange public token for access token. Returns (access_token, item_id).""" response = plaid_request("/item/public_token/exchange", { "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"] def get_institution_name(access_token: str) -> str: """Get the institution name for a linked item.""" # Get credentials from keychain client_id = get_keychain_value("PLAID_CLIENT_ID") secret = get_keychain_value("PLAID_SECRET") # Get item info item_response = subprocess.run( [ "curl", "-s", "-X", "POST", f"{PLAID_API_URL}/item/get", "-H", "Content-Type: application/json", "-d", json.dumps({ "client_id": client_id, "secret": secret, "access_token": access_token }) ], capture_output=True, text=True ) item_data = json.loads(item_response.stdout) institution_id = item_data.get("item", {}).get("institution_id", "") if not institution_id: return "Unknown_Bank" # Get institution name inst_response = subprocess.run( [ "curl", "-s", "-X", "POST", f"{PLAID_API_URL}/institutions/get_by_id", "-H", "Content-Type: application/json", "-d", json.dumps({ "client_id": client_id, "secret": secret, "institution_id": institution_id, "country_codes": ["US"] }) ], capture_output=True, text=True ) inst_data = json.loads(inst_response.stdout) name = inst_data.get("institution", {}).get("name", "Unknown_Bank") # Replace spaces with underscores for keychain account name return name.replace(" ", "_") class CallbackHandler(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 received_public_token parsed = urllib.parse.urlparse(self.path) if parsed.path == "/callback": # Parse query parameters params = urllib.parse.parse_qs(parsed.query) if "public_token" in params: received_public_token = params["public_token"][0] # Send success response 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 your terminal.</p> </div> </body> </html> """) # Signal server to stop server_should_stop.set() else: # Error or user cancelled 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> """) server_should_stop.set() elif parsed.path == "/link": # Serve the Plaid Link page 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(); // Auto-open Plaid Link setTimeout(() => handler.open(), 500); </script> </body> </html> """ self.wfile.write(html.encode()) else: self.send_response(404) self.end_headers() def run_server(link_token: str): """Run the temporary callback server.""" with socketserver.TCPServer(("", REDIRECT_PORT), CallbackHandler) as httpd: httpd.link_token = link_token httpd.timeout = 1 print(f" Callback server running on port {REDIRECT_PORT}") while not server_should_stop.is_set(): httpd.handle_request() def main(): print("\n" + "=" * 50) print(" Plaid Account Linker") print("=" * 50) print("\nThis will open your browser to securely link a new bank account.") print("Your access token will be stored in macOS Keychain.\n") # Verify we can access keychain credentials print("[1/5] Verifying Plaid credentials in Keychain...") try: get_keychain_value("PLAID_CLIENT_ID") get_keychain_value("PLAID_SECRET") print(" Credentials found.") except Exception as e: print(f"\n ERROR: {e}") print("\n Please ensure your Plaid credentials are in Keychain:") print(' security add-generic-password -s "PLAID_CLIENT_ID" -a "plaid" -w "your_client_id"') print(' security add-generic-password -s "PLAID_SECRET" -a "plaid" -w "your_secret"') sys.exit(1) # Create link token print("\n[2/5] Creating Plaid Link token...") try: link_token = create_link_token() print(" Link token created.") except Exception as e: print(f"\n ERROR: {e}") sys.exit(1) # Start callback server in background print("\n[3/5] Starting temporary callback server...") server_thread = threading.Thread(target=run_server, args=(link_token,), daemon=True) server_thread.start() # Open browser print("\n[4/5] Opening browser for Plaid Link...") time.sleep(0.5) webbrowser.open(f"http://localhost:{REDIRECT_PORT}/link") print("\n" + "-" * 50) print(" Waiting for you to complete the Link flow...") print(" (Press Ctrl+C to cancel)") print("-" * 50 + "\n") # Wait for callback try: while not server_should_stop.is_set(): time.sleep(0.5) except KeyboardInterrupt: print("\n\nCancelled by user.") sys.exit(0) # Check if we got a token if not received_public_token: print("\nNo account was linked. Exiting.") sys.exit(0) # Exchange token print("[5/5] Exchanging token and saving to Keychain...") try: access_token, item_id = exchange_public_token(received_public_token) # Get institution name institution_name = get_institution_name(access_token) # Store in keychain with the format expected by the MCP account_name = f"access_token_{item_id}_{institution_name}" if store_keychain_value("PlaidTracker", account_name, access_token): print(f"\n" + "=" * 50) print(" SUCCESS!") print("=" * 50) print(f"\n Institution: {institution_name.replace('_', ' ')}") print(f" Stored as: {account_name[:40]}...") print(f"\n Your new account is now available in Plaid MCP!") print(" Try: 'Show me my accounts' in Claude\n") else: print("\n ERROR: Failed to store token in Keychain") sys.exit(1) except Exception as e: print(f"\n ERROR: {e}") sys.exit(1) if __name__ == "__main__": main()