"""Paste redirect URL as argument: python exchange_code.py "https://127.0.0.1/?code=..." """
import base64, json, time, sys, httpx, os
from pathlib import Path
from urllib.parse import parse_qs, urlparse
from dotenv import load_dotenv
load_dotenv()
CALLBACK_URL = os.getenv('SCHWAB_CALLBACK_URL', 'https://127.0.0.1:8182/callback')
TOKEN_PATH = Path(os.getenv('SCHWAB_TOKEN_PATH', '~/.schwab-mcp/token.json')).expanduser()
def load_client_credentials():
"""Load client credentials from env or existing token file."""
client_id = os.getenv('SCHWAB_CLIENT_ID')
client_secret = os.getenv('SCHWAB_CLIENT_SECRET')
if client_id and client_secret:
return client_id, client_secret
try:
with open(TOKEN_PATH) as f:
data = json.load(f)
return data.get('client_id'), data.get('client_secret')
except (FileNotFoundError, json.JSONDecodeError):
return None, None
CLIENT_ID, CLIENT_SECRET = load_client_credentials()
if not CLIENT_ID or not CLIENT_SECRET:
raise SystemExit("Missing SCHWAB_CLIENT_ID or SCHWAB_CLIENT_SECRET. "
"Set them in your environment or add client_id/client_secret to the token file.")
auth_code = parse_qs(urlparse(sys.argv[1]).query)['code'][0]
encoded = base64.b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()
r = httpx.post('https://api.schwabapi.com/v1/oauth/token',
headers={'Authorization': f'Basic {encoded}', 'Content-Type': 'application/x-www-form-urlencoded'},
data={'grant_type': 'authorization_code', 'code': auth_code, 'redirect_uri': CALLBACK_URL})
if r.status_code == 200:
d = r.json()
TOKEN_PATH.parent.mkdir(parents=True, exist_ok=True)
json.dump({'access_token': d['access_token'], 'refresh_token': d['refresh_token'],
'expires_at': time.time() + d['expires_in'], 'token_type': 'Bearer',
'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET}, open(TOKEN_PATH, 'w'), indent=2)
print(f'Success! Token saved to {TOKEN_PATH}')
else:
print(f'Error {r.status_code}: {r.text}')
