LinkedIn MCP Server

#!/usr/bin/env python3 """ LinkedIn OAuth Helper Simple utility to help obtain LinkedIn access tokens for testing """ import argparse import base64 import json import urllib.parse import webbrowser from typing import Optional import requests class LinkedInOAuthHelper: def __init__(self, client_id: str, client_secret: str, redirect_uri: str = "http://localhost:8002/auth/callback"): self.client_id = client_id self.client_secret = client_secret self.redirect_uri = redirect_uri self.authorization_base_url = "https://www.linkedin.com/oauth/v2/authorization" self.token_url = "https://www.linkedin.com/oauth/v2/accessToken" def get_authorization_url(self, scopes: list = None) -> str: """Generate LinkedIn authorization URL""" if scopes is None: scopes = [ 'r_liteprofile', 'r_emailaddress', 'w_member_social', 'r_organization_social' ] params = { 'response_type': 'code', 'client_id': self.client_id, 'redirect_uri': self.redirect_uri, 'scope': ' '.join(scopes), 'state': 'linkedin_mcp_auth' # Simple state for CSRF protection } url = f"{self.authorization_base_url}?{urllib.parse.urlencode(params)}" return url def exchange_code_for_token(self, authorization_code: str) -> dict: """Exchange authorization code for access token""" data = { 'grant_type': 'authorization_code', 'code': authorization_code, 'redirect_uri': self.redirect_uri, 'client_id': self.client_id, 'client_secret': self.client_secret } headers = { 'Content-Type': 'application/x-www-form-urlencoded' } response = requests.post(self.token_url, data=data, headers=headers) response.raise_for_status() return response.json() def get_user_profile(self, access_token: str) -> dict: """Test the access token by getting user profile""" headers = { 'Authorization': f'Bearer {access_token}', 'X-Restli-Protocol-Version': '2.0.0' } # Get basic profile info profile_url = "https://api.linkedin.com/v2/people/~?projection=(id,firstName,lastName,headline)" response = requests.get(profile_url, headers=headers) response.raise_for_status() return response.json() def interactive_oauth_flow(): """Interactive OAuth flow for getting access token""" print("🔑 LinkedIn OAuth Helper") print("========================") print() # Get client credentials client_id = input("Enter your LinkedIn Client ID: ").strip() if not client_id: print("❌ Client ID is required") return client_secret = input("Enter your LinkedIn Client Secret: ").strip() if not client_secret: print("❌ Client Secret is required") return # Create OAuth helper oauth = LinkedInOAuthHelper(client_id, client_secret) # Generate authorization URL auth_url = oauth.get_authorization_url() print("\n🚀 OAuth Flow Steps:") print("1. Opening LinkedIn authorization page in your browser...") print("2. Login and authorize the application") print("3. Copy the authorization code from the redirect URL") print() print(f"Authorization URL: {auth_url}") print() # Open browser try: webbrowser.open(auth_url) print("✅ Browser opened successfully") except Exception as e: print(f"⚠️ Could not open browser: {e}") print("Please manually copy the URL above into your browser") print() print("After authorizing, you'll be redirected to:") print(f"http://localhost:8002/auth/callback?code=AUTHORIZATION_CODE&state=linkedin_mcp_auth") print() # Get authorization code from user auth_code = input("Enter the authorization code from the redirect URL: ").strip() if not auth_code: print("❌ Authorization code is required") return try: # Exchange code for token print("\n🔄 Exchanging authorization code for access token...") token_response = oauth.exchange_code_for_token(auth_code) access_token = token_response.get('access_token') if not access_token: print("❌ Failed to get access token") print(f"Response: {token_response}") return print("✅ Successfully obtained access token!") print() # Test the token print("🧪 Testing access token...") try: profile = oauth.get_user_profile(access_token) print(f"✅ Token works! Connected as: {profile.get('firstName', {}).get('localized', {}).get('en_US', 'Unknown')} {profile.get('lastName', {}).get('localized', {}).get('en_US', '')}") except Exception as e: print(f"⚠️ Token might have limited permissions: {e}") print() print("📋 Token Information:") print(f"Access Token: {access_token}") print(f"Token Type: {token_response.get('token_type', 'Bearer')}") print(f"Expires In: {token_response.get('expires_in', 'Unknown')} seconds") # Save to .env file save_to_env = input("\n💾 Save token to .env file? (y/N): ").strip().lower() if save_to_env in ['y', 'yes']: try: with open('.env', 'a') as f: f.write(f"\n# LinkedIn access token generated on {datetime.now().isoformat()}\n") f.write(f"LINKEDIN_ACCESS_TOKEN={access_token}\n") f.write(f"LINKEDIN_CLIENT_ID={client_id}\n") f.write(f"LINKEDIN_CLIENT_SECRET={client_secret}\n") print("✅ Token saved to .env file") except Exception as e: print(f"❌ Failed to save to .env: {e}") print() print("🎉 OAuth flow completed successfully!") print() print("Next steps:") print("1. Set LINKEDIN_ACCESS_TOKEN environment variable") print("2. Test with: python test_linkedin.py") print("3. Use the LinkedIn MCP client at http://localhost:8002/docs") return access_token except Exception as e: print(f"❌ Error during token exchange: {e}") return None def manual_oauth_instructions(): """Show manual OAuth instructions""" print("📖 Manual LinkedIn OAuth Setup") print("==============================") print() print("If you prefer to set up OAuth manually, follow these steps:") print() print("1. 🏢 Create LinkedIn App:") print(" - Go to https://developer.linkedin.com/") print(" - Click 'Create App'") print(" - Fill in app details and create") print() print("2. 🔧 Configure App:") print(" - Go to 'Auth' tab") print(" - Add redirect URL: http://localhost:8002/auth/callback") print(" - Request permissions: r_liteprofile, r_emailaddress, w_member_social") print() print("3. 🔑 Generate Access Token:") print(" - Construct authorization URL:") print(" https://www.linkedin.com/oauth/v2/authorization?") print(" response_type=code&") print(" client_id=YOUR_CLIENT_ID&") print(" redirect_uri=http://localhost:8002/auth/callback&") print(" scope=r_liteprofile%20r_emailaddress%20w_member_social") print() print("4. 🔄 Exchange Code:") print(" - Visit URL, authorize, get code from redirect") print(" - POST to https://www.linkedin.com/oauth/v2/accessToken") print(" - With: grant_type=authorization_code, code=CODE, redirect_uri, client_id, client_secret") print() print("5. ✅ Use Token:") print(" - Set LINKEDIN_ACCESS_TOKEN environment variable") print(" - Test with the LinkedIn MCP client") def main(): parser = argparse.ArgumentParser(description="LinkedIn OAuth Helper for MCP Client") parser.add_argument("--client-id", help="LinkedIn Client ID") parser.add_argument("--client-secret", help="LinkedIn Client Secret") parser.add_argument("--manual", action="store_true", help="Show manual setup instructions") parser.add_argument("--test-token", help="Test an existing access token") args = parser.parse_args() if args.manual: manual_oauth_instructions() return if args.test_token: print("🧪 Testing LinkedIn Access Token") print("================================") try: oauth = LinkedInOAuthHelper("test", "test") # Client creds not needed for testing profile = oauth.get_user_profile(args.test_token) print("✅ Token is valid!") print(f"User: {profile.get('firstName', {}).get('localized', {}).get('en_US', 'Unknown')} {profile.get('lastName', {}).get('localized', {}).get('en_US', '')}") print(f"LinkedIn ID: {profile.get('id', 'Unknown')}") except Exception as e: print(f"❌ Token test failed: {e}") return if args.client_id and args.client_secret: # Non-interactive mode oauth = LinkedInOAuthHelper(args.client_id, args.client_secret) auth_url = oauth.get_authorization_url() print(f"Authorization URL: {auth_url}") print("Visit this URL to authorize, then use the authorization code with --code parameter") else: # Interactive mode import datetime interactive_oauth_flow() if __name__ == "__main__": main()