diff --git a/.vscode/tasks.json b/.vscode/tasks.json index 4c61307..c541f61 100644 --- a/.vscode/tasks.json +++ b/.vscode/tasks.json @@ -133,5 +133,22 @@ }, "problemMatcher": [] }, + { + "label": "bunx @anthropic-ai/dxt pack", + "detail": "Pack the DXT package", + "type": "shell", + "command": "bunx", + "args": ["@anthropic-ai/dxt", "pack"], + "group": { + "kind": "build", + "isDefault": false + }, + "presentation": { + "reveal": "always", + "panel": "new", + "focus": true + }, + "problemMatcher": [] + } ] } diff --git a/README.md b/README.md index 6c13063..21cca9c 100644 --- a/README.md +++ b/README.md @@ -36,17 +36,16 @@ Suggest improvements for my CV to target this job posting https://www.linkedin.c > [!NOTE] > July 2025: All tools are currently functional and actively maintained. If you encounter any issues, please report them in the [GitHub issues](https://github.com/stickerdaniel/linkedin-mcp-server/issues). ---- +
+
## šŸ³ Docker Setup (Recommended - Universal) **Prerequisites:** Make sure you have [Docker](https://www.docker.com/get-started/) installed and running. -**Zero setup required** - just add the mcp server to your client config and replace email and password with your linkedin credentials. - ### Installation -**Claude Desktop:** +**Client Configuration:** ```json { "mcpServers": { @@ -54,20 +53,52 @@ Suggest improvements for my CV to target this job posting https://www.linkedin.c "command": "docker", "args": [ "run", "-i", "--rm", - "-e", "LINKEDIN_EMAIL", - "-e", "LINKEDIN_PASSWORD", + "-e", "LINKEDIN_COOKIE", "stickerdaniel/linkedin-mcp-server", "--no-setup" ], "env": { - "LINKEDIN_EMAIL": "your.email@example.com", - "LINKEDIN_PASSWORD": "your_password" + "LINKEDIN_COOKIE": "XXXXXX...", } } } } ``` +### Getting the LinkedIn Cookie +
+šŸ³ Docker get-cookie method + +**Run the server with the `--get-cookie` flag:** +```bash +docker run -i --rm \ + -e LINKEDIN_EMAIL="your.email@example.com" \ + -e LINKEDIN_PASSWORD="your_password" \ + stickerdaniel/linkedin-mcp-server \ + --get-cookie +``` +Copy the cookie from the output and set it as `LINKEDIN_COOKIE` in your client configuration. If this fails with a captcha challenge, use the method below. +
+
+šŸŒ Chrome DevTools Guide + +1. Open LinkedIn and login +2. Open Chrome DevTools (F12 or right-click ā†’ Inspect) +3. Go to **Application** > **Storage** > **Cookies** > **https://www.linkedin.com** +4. Find the cookie named `li_at` +5. Copy the **Value** field (this is your LinkedIn session cookie) +6. Use this value as your `LINKEDIN_COOKIE` in the configuration + +
+
+ +> [!NOTE] +> The cookie will expire during the next 30 days. Just get the new cookie and update your config. + +> [!TIP] +> There are also many cookie manager extensions that you can use to easily get the cookie. + +### Docker Setup Help
šŸ”§ Configuration @@ -83,6 +114,8 @@ Suggest improvements for my CV to target this job posting https://www.linkedin.c - `--host HOST` - HTTP server host (default: 127.0.0.1) - `--port PORT` - HTTP server port (default: 8000) - `--path PATH` - HTTP server path (default: /mcp) +- `--get-cookie` - Attempt to login with email and password and extract the LinkedIn cookie +- `--cookie {cookie}` - Pass a specific LinkedIn cookie for login **HTTP Mode Example (for web-based MCP clients):** ```bash @@ -116,6 +149,9 @@ docker run -i --rm \ - You might get a captcha challenge if you logged in a lot of times in a short period of time, then try again later or follow the [local setup instructions](#-local-setup-develop--contribute) to run the server manually in --no-headless mode where you can debug the login process (solve captcha manually)
+
+
+ ## šŸ“¦ Claude Desktop (DXT Extension) **Prerequisites:** [Claude Desktop](https://claude.ai/download) and [Docker](https://www.docker.com/get-started/) installed @@ -126,6 +162,40 @@ docker run -i --rm \ 3. Configure your LinkedIn credentials when prompted 4. Start using LinkedIn tools immediately +### Getting the LinkedIn Cookie +
+šŸ³ Docker get-cookie method + +**Run the server with the `--get-cookie` flag:** +```bash +docker run -i --rm \ + -e LINKEDIN_EMAIL="your.email@example.com" \ + -e LINKEDIN_PASSWORD="your_password" \ + stickerdaniel/linkedin-mcp-server \ + --get-cookie +``` +Copy the cookie from the output and set it as `LINKEDIN_COOKIE` in your client configuration. If this fails with a captcha challenge, use the method below. +
+
+šŸŒ Chrome DevTools Guide + +1. Open LinkedIn and login +2. Open Chrome DevTools (F12 or right-click ā†’ Inspect) +3. Go to **Application** > **Storage** > **Cookies** > **https://www.linkedin.com** +4. Find the cookie named `li_at` +5. Copy the **Value** field (this is your LinkedIn session cookie) +6. Use this value as your `LINKEDIN_COOKIE` in the configuration + +
+
+ +> [!NOTE] +> The cookie will expire during the next 30 days. Just get the new cookie and update your config. + +> [!TIP] +> There are also many cookie manager extensions that you can use to easily get the cookie. + +### DXT Extension Setup Help
ā— Troubleshooting @@ -139,6 +209,9 @@ docker run -i --rm \ - You might get a captcha challenge if you logged in a lot of times in a short period of time, then try again later or follow the [local setup instructions](#-local-setup-develop--contribute) to run the server manually in --no-headless mode where you can debug the login process (solve captcha manually)
+
+
+ ## šŸ Local Setup (Develop & Contribute) **Prerequisites:** [Chrome browser](https://www.google.com/chrome/) and [Git](https://git-scm.com/downloads) installed @@ -170,21 +243,26 @@ uv sync --group dev uv run pre-commit install # 5. Start the server once manually -# (you will be prompted to enter your LinkedIn credentials, and they are securely stored in your OS keychain) +# You will be prompted to enter your LinkedIn credentials, and they will be securely stored in your OS keychain +# Once logged in, your cookie will be stored in your OS keychain and used for subsequent runs until it expires uv run main.py --no-headless --no-lazy-init ``` +### Local Setup Help
šŸ”§ Configuration **CLI Options:** - `--no-headless` - Show browser window (debugging) - `--debug` - Enable detailed logging -- `--no-setup` - Skip credential prompts (make sure to set `LINKEDIN_EMAIL` and `LINKEDIN_PASSWORD` in env or or run the server once manualy, then it will be stored in your OS keychain and you can run the server without credentials) +- `--no-setup` - Skip credential prompts (make sure to set `LINKEDIN_COOKIE` or `LINKEDIN_EMAIL` and `LINKEDIN_PASSWORD` in env or that you run the server once manually, so the authentication is stored in your OS keychain and you can run the server without credentials) - `--no-lazy-init` - Login to LinkedIn immediately instead of waiting for the first tool call +- `--get-cookie` - Login with email and password and extract the LinkedIn cookie +- `--cookie {cookie}` - Pass a specific LinkedIn cookie for login +- `--help` - Show help **Claude Desktop:** -```json +```**json** { "mcpServers": { "linkedin": { @@ -217,7 +295,9 @@ uv run main.py --no-headless --no-lazy-init Feel free to open an [issue](https://github.com/stickerdaniel/linkedin-mcp-server/issues) or [PR](https://github.com/stickerdaniel/linkedin-mcp-server/pulls)! ---- + +
+
## License diff --git a/linkedin_mcp_server/authentication.py b/linkedin_mcp_server/authentication.py new file mode 100644 index 0000000..7d54b52 --- /dev/null +++ b/linkedin_mcp_server/authentication.py @@ -0,0 +1,171 @@ +# linkedin_mcp_server/authentication.py +""" +Pure authentication module for LinkedIn MCP Server. + +This module handles authentication without any driver dependencies. +""" + +import logging + +from linkedin_mcp_server.config import get_config +from linkedin_mcp_server.config.providers import ( + clear_cookie_from_keyring, + get_cookie_from_keyring, + save_cookie_to_keyring, +) +from linkedin_mcp_server.exceptions import CredentialsNotFoundError + +# Constants for cookie validation +MIN_RAW_COOKIE_LENGTH = 110 +MIN_COOKIE_LENGTH = MIN_RAW_COOKIE_LENGTH + len("li_at=") + +logger = logging.getLogger(__name__) + + +def has_authentication() -> bool: + """ + Check if authentication is available without triggering setup. + + Returns: + bool: True if authentication (cookie) is available, False otherwise + """ + config = get_config() + + # Check environment variable + if config.linkedin.cookie: + return True + + # Check keyring if enabled + if config.linkedin.use_keyring: + cookie = get_cookie_from_keyring() + if cookie: + return True + + return False + + +def get_authentication() -> str: + """ + Get LinkedIn cookie from available sources. + + Returns: + str: LinkedIn session cookie + + Raises: + CredentialsNotFoundError: If no authentication is available + """ + config = get_config() + + # First, try environment variable + if config.linkedin.cookie: + logger.info("Using LinkedIn cookie from environment") + return config.linkedin.cookie + + # Second, try keyring if enabled + if config.linkedin.use_keyring: + cookie = get_cookie_from_keyring() + if cookie: + logger.info("Using LinkedIn cookie from keyring") + return cookie + + # No authentication available + raise CredentialsNotFoundError("No LinkedIn cookie found") + + +def store_authentication(cookie: str) -> bool: + """ + Store LinkedIn cookie securely. + + Args: + cookie: LinkedIn session cookie to store + + Returns: + bool: True if storage was successful, False otherwise + """ + config = get_config() + + if config.linkedin.use_keyring: + success = save_cookie_to_keyring(cookie) + if success: + logger.info("Cookie stored securely in keyring") + else: + logger.warning("Could not store cookie in system keyring") + return success + else: + logger.info("Keyring disabled, cookie not stored") + return False + + +def clear_authentication() -> bool: + """ + Clear stored authentication. + + Returns: + bool: True if clearing was successful, False otherwise + """ + config = get_config() + + if config.linkedin.use_keyring: + success = clear_cookie_from_keyring() + if success: + logger.info("Authentication cleared from keyring") + else: + logger.warning("Could not clear authentication from keyring") + return success + else: + logger.info("Keyring disabled, nothing to clear") + return True + + +def validate_cookie_format(cookie: str) -> bool: + """ + Validate that the cookie has the expected format. + + Args: + cookie: Cookie string to validate + + Returns: + bool: True if cookie format is valid, False otherwise + """ + if not cookie: + return False + + # LinkedIn session cookies typically start with "li_at=" + if cookie.startswith("li_at=") and len(cookie) > MIN_COOKIE_LENGTH: + return True + + # Also accept raw cookie values (without li_at= prefix) + if ( + not cookie.startswith("li_at=") + and len(cookie) > MIN_RAW_COOKIE_LENGTH + and "=" not in cookie + ): + return True + + return False + + +def ensure_authentication() -> str: + """ + Ensure authentication is available, raising clear error if not. + + Returns: + str: LinkedIn session cookie + + Raises: + CredentialsNotFoundError: If no authentication is available with clear instructions + """ + try: + return get_authentication() + except CredentialsNotFoundError: + config = get_config() + + if config.chrome.non_interactive: + raise CredentialsNotFoundError( + "No LinkedIn cookie found. Please provide cookie via " + "environment variable (LINKEDIN_COOKIE) or run with --get-cookie to obtain one." + ) + else: + raise CredentialsNotFoundError( + "No LinkedIn authentication found. Please run setup to configure authentication." + ) diff --git a/linkedin_mcp_server/config/__init__.py b/linkedin_mcp_server/config/__init__.py index caa6151..a79d7ee 100644 --- a/linkedin_mcp_server/config/__init__.py +++ b/linkedin_mcp_server/config/__init__.py @@ -1,14 +1,15 @@ # src/linkedin_mcp_server/config/__init__.py -from typing import Optional import logging -from .schema import AppConfig, ChromeConfig, LinkedInConfig, ServerConfig +from typing import Optional + from .loaders import load_config from .providers import ( - get_credentials_from_keyring, - save_credentials_to_keyring, clear_credentials_from_keyring, + get_credentials_from_keyring, get_keyring_name, + save_credentials_to_keyring, ) +from .schema import AppConfig, ChromeConfig, LinkedInConfig, ServerConfig logger = logging.getLogger(__name__) diff --git a/linkedin_mcp_server/config/loaders.py b/linkedin_mcp_server/config/loaders.py index ca1ea27..771d963 100644 --- a/linkedin_mcp_server/config/loaders.py +++ b/linkedin_mcp_server/config/loaders.py @@ -1,10 +1,11 @@ # src/linkedin_mcp_server/config/loaders.py -import os import argparse import logging +import os from typing import Optional -from .schema import AppConfig + from .providers import get_chromedriver_paths +from .schema import AppConfig logger = logging.getLogger(__name__) @@ -33,6 +34,9 @@ def load_from_env(config: AppConfig) -> AppConfig: if password := os.environ.get("LINKEDIN_PASSWORD"): config.linkedin.password = password + if cookie := os.environ.get("LINKEDIN_COOKIE"): + config.linkedin.cookie = cookie + # ChromeDriver configuration if chromedriver := os.environ.get("CHROMEDRIVER"): config.chrome.chromedriver_path = chromedriver @@ -124,6 +128,18 @@ def load_from_args(config: AppConfig) -> AppConfig: help="Specify the path to the ChromeDriver executable", ) + parser.add_argument( + "--get-cookie", + action="store_true", + help="Login with credentials and display cookie for Docker setup", + ) + + parser.add_argument( + "--cookie", + type=str, + help="Specify LinkedIn cookie directly", + ) + args = parser.parse_args() # Update configuration with parsed arguments @@ -157,6 +173,11 @@ def load_from_args(config: AppConfig) -> AppConfig: if args.chromedriver: config.chrome.chromedriver_path = args.chromedriver + if hasattr(args, "get_cookie") and args.get_cookie: + config.server.get_cookie = True + if args.cookie: + config.linkedin.cookie = args.cookie + return config diff --git a/linkedin_mcp_server/config/providers.py b/linkedin_mcp_server/config/providers.py index 31a3977..aab3054 100644 --- a/linkedin_mcp_server/config/providers.py +++ b/linkedin_mcp_server/config/providers.py @@ -10,6 +10,7 @@ SERVICE_NAME = "linkedin_mcp_server" EMAIL_KEY = "linkedin_email" PASSWORD_KEY = "linkedin_password" +COOKIE_KEY = "linkedin_cookie" logger = logging.getLogger(__name__) @@ -74,6 +75,27 @@ def clear_credentials_from_keyring() -> bool: return False +def get_cookie_from_keyring() -> Optional[str]: + """Retrieve LinkedIn cookie from system keyring.""" + return get_secret_from_keyring(COOKIE_KEY) + + +def save_cookie_to_keyring(cookie: str) -> bool: + """Save LinkedIn cookie to system keyring.""" + return set_secret_in_keyring(COOKIE_KEY, cookie) + + +def clear_cookie_from_keyring() -> bool: + """Clear stored cookie from the keyring.""" + try: + keyring.delete_password(SERVICE_NAME, COOKIE_KEY) + logger.info(f"Cookie removed from {get_keyring_name()}") + return True + except KeyringError as e: + logger.error(f"Error clearing cookie: {e}") + return False + + def get_chromedriver_paths() -> List[str]: """Get possible ChromeDriver paths based on the platform.""" paths = [ diff --git a/linkedin_mcp_server/config/schema.py b/linkedin_mcp_server/config/schema.py index 55d912f..cf3e012 100644 --- a/linkedin_mcp_server/config/schema.py +++ b/linkedin_mcp_server/config/schema.py @@ -19,6 +19,7 @@ class LinkedInConfig: email: Optional[str] = None password: Optional[str] = None + cookie: Optional[str] = None use_keyring: bool = True @@ -30,6 +31,7 @@ class ServerConfig: lazy_init: bool = True debug: bool = False setup: bool = True + get_cookie: bool = False # HTTP transport configuration host: str = "127.0.0.1" port: int = 8000 diff --git a/linkedin_mcp_server/config/secrets.py b/linkedin_mcp_server/config/secrets.py index bafe01c..c01dc1b 100644 --- a/linkedin_mcp_server/config/secrets.py +++ b/linkedin_mcp_server/config/secrets.py @@ -17,7 +17,7 @@ def get_credentials() -> Dict[str, str]: - """Get LinkedIn credentials from config, keyring, or prompt.""" + """Get LinkedIn credentials from config, keyring, or prompt (legacy for --get-cookie).""" config = get_config() # First, try configuration (includes environment variables) diff --git a/linkedin_mcp_server/drivers/chrome.py b/linkedin_mcp_server/drivers/chrome.py index 3c21db2..7b6243c 100644 --- a/linkedin_mcp_server/drivers/chrome.py +++ b/linkedin_mcp_server/drivers/chrome.py @@ -1,16 +1,15 @@ -# src/linkedin_mcp_server/drivers/chrome.py +# linkedin_mcp_server/drivers/chrome.py """ Chrome driver management for LinkedIn scraping. This module handles the creation and management of Chrome WebDriver instances. +Simplified to focus only on driver management without authentication setup. """ import logging import os -import sys from typing import Dict, Optional -import inquirer # type: ignore from linkedin_scraper.exceptions import ( CaptchaRequiredError, InvalidCredentialsError, @@ -25,12 +24,10 @@ from selenium.webdriver.chrome.service import Service from linkedin_mcp_server.config import get_config -from linkedin_mcp_server.config.providers import clear_credentials_from_keyring -from linkedin_mcp_server.config.secrets import get_credentials -from linkedin_mcp_server.exceptions import ( - CredentialsNotFoundError, - DriverInitializationError, -) +from linkedin_mcp_server.exceptions import DriverInitializationError + +# Constants +DEFAULT_USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36" # Global driver storage to reuse sessions active_drivers: Dict[str, webdriver.Chrome] = {} @@ -38,23 +35,17 @@ logger = logging.getLogger(__name__) -def get_or_create_driver() -> Optional[webdriver.Chrome]: +def create_chrome_driver() -> webdriver.Chrome: """ - Get existing driver or create a new one using the configured settings. + Create a new Chrome WebDriver instance with proper configuration. Returns: - Optional[webdriver.Chrome]: Chrome WebDriver instance or None if initialization fails - in non-interactive mode + webdriver.Chrome: Configured Chrome WebDriver instance Raises: - WebDriverException: If the driver cannot be created and not in non-interactive mode + WebDriverException: If driver creation fails """ config = get_config() - session_id = "default" # We use a single session for simplicity - - # Return existing driver if available - if session_id in active_drivers: - return active_drivers[session_id] # Set up Chrome options chrome_options = Options() @@ -64,244 +55,179 @@ def get_or_create_driver() -> Optional[webdriver.Chrome]: if config.chrome.headless: chrome_options.add_argument("--headless=new") - # Add essential options for stability (compatible with both Grid and direct) + # Add essential options for stability chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-gpu") chrome_options.add_argument("--window-size=1920,1080") chrome_options.add_argument("--disable-extensions") chrome_options.add_argument("--disable-background-timer-throttling") - chrome_options.add_argument( - "--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36" - ) + + # Set user agent (configurable with sensible default) + user_agent = getattr(config.chrome, "user_agent", DEFAULT_USER_AGENT) + chrome_options.add_argument(f"--user-agent={user_agent}") # Add any custom browser arguments from config for arg in config.chrome.browser_args: chrome_options.add_argument(arg) # Initialize Chrome driver - try: - logger.info("Initializing Chrome WebDriver...") + logger.info("Initializing Chrome WebDriver...") - # Use ChromeDriver path from environment or config - chromedriver_path = ( - os.environ.get("CHROMEDRIVER_PATH") or config.chrome.chromedriver_path - ) + # Use ChromeDriver path from environment or config + chromedriver_path = ( + os.environ.get("CHROMEDRIVER_PATH") or config.chrome.chromedriver_path + ) - if chromedriver_path: - logger.info(f"Using ChromeDriver at path: {chromedriver_path}") - service = Service(executable_path=chromedriver_path) - driver = webdriver.Chrome(service=service, options=chrome_options) - else: - logger.info("Using auto-detected ChromeDriver") - driver = webdriver.Chrome(options=chrome_options) + if chromedriver_path: + logger.info(f"Using ChromeDriver at path: {chromedriver_path}") + service = Service(executable_path=chromedriver_path) + driver = webdriver.Chrome(service=service, options=chrome_options) + else: + logger.info("Using auto-detected ChromeDriver") + driver = webdriver.Chrome(options=chrome_options) - logger.info("Chrome WebDriver initialized successfully") + logger.info("Chrome WebDriver initialized successfully") - # Add a page load timeout for safety - driver.set_page_load_timeout(60) + # Add a page load timeout for safety + driver.set_page_load_timeout(60) - # Try to log in with retry loop - max_retries = 3 - for attempt in range(max_retries): - try: - if login_to_linkedin(driver): - logger.info("Successfully logged in to LinkedIn") - active_drivers[session_id] = driver - return driver - except ( - CaptchaRequiredError, - InvalidCredentialsError, - SecurityChallengeError, - TwoFactorAuthError, - RateLimitError, - LoginTimeoutError, - CredentialsNotFoundError, - ) as e: - if config.chrome.non_interactive: - # In non-interactive mode, propagate the error - driver.quit() - raise e - else: - # In interactive mode, handle the error and potentially retry - should_retry = handle_login_error(e) - if should_retry and attempt < max_retries - 1: - logger.info(f"Retry attempt {attempt + 2}/{max_retries}") - continue - else: - # Clean up driver on final failure - driver.quit() - return None - except Exception as e: - error_msg = f"Error creating web driver: {e}" - logger.error( - error_msg, - extra={"exception_type": type(e).__name__, "exception_message": str(e)}, - ) + # Set shorter implicit wait for faster cookie validation + driver.implicitly_wait(10) - if config.chrome.non_interactive: - raise DriverInitializationError(error_msg) - else: - raise WebDriverException(error_msg) + return driver -def login_to_linkedin(driver: webdriver.Chrome) -> bool: +def login_with_cookie(driver: webdriver.Chrome, cookie: str) -> bool: """ - Log in to LinkedIn using stored or provided credentials. + Log in to LinkedIn using session cookie. Args: driver: Chrome WebDriver instance + cookie: LinkedIn session cookie Returns: bool: True if login was successful, False otherwise + """ + try: + from linkedin_scraper import actions # type: ignore - Raises: - Various login-related errors from linkedin-scraper + logger.info("Attempting cookie authentication...") + + # Set shorter timeout for faster failure detection + driver.set_page_load_timeout(15) + + actions.login(driver, cookie=cookie) + + # Quick check - if we're on login page, cookie is invalid + current_url = driver.current_url + if "login" in current_url or "uas/login" in current_url: + logger.warning("Cookie authentication failed - redirected to login page") + return False + elif ( + "feed" in current_url + or "mynetwork" in current_url + or "linkedin.com/in/" in current_url + ): + logger.info("Cookie authentication successful") + return True + else: + logger.warning("Cookie authentication failed - unexpected page") + return False + + except Exception as e: + logger.warning(f"Cookie authentication failed: {e}") + return False + finally: + # Restore normal timeout + driver.set_page_load_timeout(60) + + +def login_to_linkedin(driver: webdriver.Chrome, authentication: str) -> None: """ - config = get_config() + Log in to LinkedIn using provided authentication. - # Get LinkedIn credentials from config - try: - credentials = get_credentials() - except CredentialsNotFoundError as e: - if config.chrome.non_interactive: - raise e - # Only prompt if not in non-interactive mode - from linkedin_mcp_server.config.secrets import prompt_for_credentials + Args: + driver: Chrome WebDriver instance + authentication: LinkedIn session cookie - credentials = prompt_for_credentials() + Raises: + Various login-related errors from linkedin-scraper or this module + """ + # Try cookie authentication + if login_with_cookie(driver, authentication): + logger.info("Successfully logged in to LinkedIn using cookie") + return - if not credentials: - raise CredentialsNotFoundError("No credentials available") + # If we get here, cookie authentication failed + logger.error("Cookie authentication failed") - # Login to LinkedIn using enhanced linkedin-scraper - logger.info("Logging in to LinkedIn...") + # Clear invalid cookie from keyring + from linkedin_mcp_server.authentication import clear_authentication - from linkedin_scraper import actions # type: ignore + clear_authentication() + logger.info("Cleared invalid cookie from authentication storage") - # Use linkedin-scraper login but with simplified error handling + # Check current page to determine the issue try: - actions.login( - driver, - credentials["email"], - credentials["password"], - interactive=not config.chrome.non_interactive, - ) - - logger.info("Successfully logged in to LinkedIn") - return True - - except Exception: - # Check current page to determine the real issue - current_url = driver.current_url + current_url: str = driver.current_url if "checkpoint/challenge" in current_url: - # We're on a challenge page - this is the real issue, not credentials if "security check" in driver.page_source.lower(): raise SecurityChallengeError( challenge_url=current_url, message="LinkedIn requires a security challenge. Please complete it manually and restart the application.", ) else: - raise CaptchaRequiredError( - captcha_url=current_url, - ) - - elif "feed" in current_url or "mynetwork" in current_url: - # Actually logged in successfully despite the exception - logger.info("Successfully logged in to LinkedIn") - return True - + raise CaptchaRequiredError(captcha_url=current_url) else: - # Check for actual credential issues - page_source = driver.page_source.lower() - if any( - pattern in page_source - for pattern in ["wrong email", "wrong password", "incorrect", "invalid"] - ): - raise InvalidCredentialsError("Invalid LinkedIn email or password.") - elif "too many" in page_source: - raise RateLimitError( - "Too many login attempts. Please wait and try again later." - ) - else: - raise LoginTimeoutError( - "Login failed. Please check your credentials and network connection." - ) + raise InvalidCredentialsError( + "Cookie authentication failed - cookie may be expired or invalid" + ) + except Exception as e: + # If we can't determine the specific error, raise a generic one + raise LoginTimeoutError(f"Login failed: {str(e)}") -def handle_login_error(error: Exception) -> bool: - """Handle login errors in interactive mode. - Returns: - bool: True if user wants to retry, False if they want to exit +def get_or_create_driver(authentication: str) -> webdriver.Chrome: """ - config = get_config() + Get existing driver or create a new one and login. - logger.error(f"\nāŒ {str(error)}") + Args: + authentication: LinkedIn session cookie for login - if config.chrome.headless: - logger.info( - "šŸ” Try running with visible browser window: uv run main.py --no-headless" - ) - - # Only allow retry for credential errors - if isinstance(error, InvalidCredentialsError): - retry = inquirer.prompt( - [ - inquirer.Confirm( - "retry", - message="Would you like to try with different credentials?", - default=True, - ), - ] - ) - if retry and retry.get("retry", False): - clear_credentials_from_keyring() - logger.info("āœ… Credentials cleared from keyring.") - logger.info("šŸ”„ Retrying with new credentials...") - return True + Returns: + webdriver.Chrome: Chrome WebDriver instance, logged in and ready + + Raises: + DriverInitializationError: If driver creation fails + Various login-related errors: If login fails + """ + session_id = "default" # We use a single session for simplicity - return False + # Return existing driver if available + if session_id in active_drivers: + logger.info("Using existing Chrome WebDriver session") + return active_drivers[session_id] + try: + # Create new driver + driver = create_chrome_driver() -def initialize_driver() -> None: - """ - Initialize the driver based on global configuration. - """ - config = get_config() + # Login to LinkedIn + login_to_linkedin(driver, authentication) - if config.server.lazy_init: - logger.info( - "Using lazy initialization - driver will be created on first tool call" - ) - if config.linkedin.email and config.linkedin.password: - logger.info("LinkedIn credentials found in configuration") - else: - logger.info( - "No LinkedIn credentials found - will look for stored credentials on first use" - ) - return + # Store successful driver + active_drivers[session_id] = driver + logger.info("Chrome WebDriver session created and authenticated successfully") - # Validate chromedriver can be found - if config.chrome.chromedriver_path: - logger.info(f"āœ… ChromeDriver found at: {config.chrome.chromedriver_path}") - os.environ["CHROMEDRIVER"] = config.chrome.chromedriver_path - else: - logger.info("āš ļø ChromeDriver not found in common locations.") - logger.info("āš” Continuing with automatic detection...") - logger.info( - "šŸ’” Tip: install ChromeDriver and set the CHROMEDRIVER environment variable" - ) + return driver - # Create driver and log in - try: - driver = get_or_create_driver() - if driver: - logger.info("āœ… Web driver initialized successfully") - else: - # Driver creation failed - always raise an error - raise DriverInitializationError("Failed to initialize web driver") + except WebDriverException as e: + error_msg = f"Error creating web driver: {e}" + logger.error(error_msg) + raise DriverInitializationError(error_msg) except ( CaptchaRequiredError, InvalidCredentialsError, @@ -309,83 +235,56 @@ def initialize_driver() -> None: TwoFactorAuthError, RateLimitError, LoginTimeoutError, - CredentialsNotFoundError, ) as e: - # Always re-raise login-related errors so main.py can handle them + # Login-related errors - clean up driver if it was created + if session_id in active_drivers: + active_drivers[session_id].quit() + del active_drivers[session_id] raise e - except WebDriverException as e: - if config.chrome.non_interactive: - raise DriverInitializationError( - f"Failed to initialize web driver: {str(e)}" - ) - logger.error(f"āŒ Failed to initialize web driver: {str(e)}") - handle_driver_error() -def handle_driver_error() -> None: +def close_all_drivers() -> None: + """Close all active drivers and clean up resources.""" + global active_drivers + + for session_id, driver in active_drivers.items(): + try: + logger.info(f"Closing Chrome WebDriver session: {session_id}") + driver.quit() + except Exception as e: + logger.warning(f"Error closing driver {session_id}: {e}") + + active_drivers.clear() + logger.info("All Chrome WebDriver sessions closed") + + +def get_active_driver() -> Optional[webdriver.Chrome]: """ - Handle ChromeDriver initialization errors by providing helpful options. + Get the currently active driver without creating a new one. + + Returns: + Optional[webdriver.Chrome]: Active driver if available, None otherwise """ - config = get_config() + session_id = "default" + return active_drivers.get(session_id) - # Skip interactive handling in non-interactive mode - if config.chrome.non_interactive: - logger.error( - "āŒ ChromeDriver is required for this application to work properly." - ) - sys.exit(1) - - questions = [ - inquirer.List( - "chromedriver_action", - message="What would you like to do?", - choices=[ - ("Specify ChromeDriver path manually", "specify"), - ("Get help installing ChromeDriver", "help"), - ("Exit", "exit"), - ], - ), - ] - answers = inquirer.prompt(questions) - - if answers["chromedriver_action"] == "specify": - path = inquirer.prompt( - [inquirer.Text("custom_path", message="Enter ChromeDriver path")] - )["custom_path"] - - if os.path.exists(path): - # Update config with the new path - config.chrome.chromedriver_path = path - os.environ["CHROMEDRIVER"] = path - logger.info(f"āœ… ChromeDriver path set to: {path}") - logger.info( - "šŸ’” Please restart the application to use the new ChromeDriver path." - ) - logger.info(" Example: uv run main.py") - sys.exit(0) - else: - logger.warning(f"āš ļø Warning: The specified path does not exist: {path}") - logger.info("šŸ’” Please check the path and restart the application.") - sys.exit(1) - - elif answers["chromedriver_action"] == "help": - logger.info("\nšŸ“‹ ChromeDriver Installation Guide:") - logger.info( - "1. Find your Chrome version: Chrome menu > Help > About Google Chrome" - ) - logger.info( - "2. Download matching ChromeDriver: https://chromedriver.chromium.org/downloads" - ) - logger.info("3. Place ChromeDriver in a location on your PATH") - logger.info(" - macOS/Linux: /usr/local/bin/ is recommended") - logger.info( - " - Windows: Add to a directory in your PATH or specify the full path\n" - ) - - if inquirer.prompt( - [inquirer.Confirm("try_again", message="Try again?", default=True)] - )["try_again"]: - initialize_driver() - - logger.error("āŒ ChromeDriver is required for this application to work properly.") - sys.exit(1) + +def capture_session_cookie(driver: webdriver.Chrome) -> Optional[str]: + """ + Capture LinkedIn session cookie from driver. + + Args: + driver: Chrome WebDriver instance + + Returns: + Optional[str]: Session cookie if found, None otherwise + """ + try: + # Get li_at cookie which is the main LinkedIn session cookie + cookie = driver.get_cookie("li_at") + if cookie and cookie.get("value"): + return f"li_at={cookie['value']}" + return None + except Exception as e: + logger.warning(f"Failed to capture session cookie: {e}") + return None diff --git a/linkedin_mcp_server/error_handler.py b/linkedin_mcp_server/error_handler.py index 1e9d3d2..745f306 100644 --- a/linkedin_mcp_server/error_handler.py +++ b/linkedin_mcp_server/error_handler.py @@ -68,9 +68,9 @@ def convert_exception_to_response( """ if isinstance(exception, CredentialsNotFoundError): return { - "error": "credentials_not_found", + "error": "authentication_not_found", "message": str(exception), - "resolution": "Provide LinkedIn credentials via environment variables", + "resolution": "Provide LinkedIn cookie via LINKEDIN_COOKIE environment variable or run setup", } elif isinstance(exception, InvalidCredentialsError): @@ -161,17 +161,18 @@ def safe_get_driver(): Safely get or create a driver with proper error handling. Returns: - Driver instance or None if initialization fails + Driver instance Raises: - LinkedInMCPError: If driver initialization fails in non-interactive mode + LinkedInMCPError: If driver initialization fails """ + from linkedin_mcp_server.authentication import ensure_authentication from linkedin_mcp_server.drivers.chrome import get_or_create_driver - driver = get_or_create_driver() - if not driver: - from linkedin_mcp_server.exceptions import DriverInitializationError + # Get authentication first + authentication = ensure_authentication() - raise DriverInitializationError("Failed to initialize Chrome driver") + # Create driver with authentication + driver = get_or_create_driver(authentication) return driver diff --git a/linkedin_mcp_server/logging_config.py b/linkedin_mcp_server/logging_config.py index 98616ed..c5871a2 100644 --- a/linkedin_mcp_server/logging_config.py +++ b/linkedin_mcp_server/logging_config.py @@ -84,7 +84,8 @@ def configure_logging(debug: bool = False, json_format: bool = False) -> None: debug: Whether to enable debug logging json_format: Whether to use JSON formatting for logs """ - log_level = logging.DEBUG if debug else logging.INFO + # Set end-user appropriate logging level: WARNING for production, DEBUG for debug mode + log_level = logging.DEBUG if debug else logging.WARNING if json_format: formatter = MCPJSONFormatter() @@ -104,6 +105,7 @@ def configure_logging(debug: bool = False, json_format: bool = False) -> None: console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) - # Set specific loggers - logging.getLogger("selenium").setLevel(logging.WARNING) - logging.getLogger("urllib3").setLevel(logging.WARNING) + # Set specific loggers to reduce noise + logging.getLogger("selenium").setLevel(logging.ERROR) + logging.getLogger("urllib3").setLevel(logging.ERROR) + logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR) diff --git a/linkedin_mcp_server/setup.py b/linkedin_mcp_server/setup.py new file mode 100644 index 0000000..793a1fd --- /dev/null +++ b/linkedin_mcp_server/setup.py @@ -0,0 +1,337 @@ +# linkedin_mcp_server/setup.py +""" +Interactive setup module for LinkedIn MCP Server. + +This module handles interactive setup flows and authentication configuration. +""" + +import logging +import os +from contextlib import contextmanager +from typing import Dict, Iterator + +import inquirer +from selenium import webdriver +from selenium.webdriver.chrome.options import Options + +from linkedin_mcp_server.authentication import store_authentication +from linkedin_mcp_server.config import get_config +from linkedin_mcp_server.config.providers import ( + get_credentials_from_keyring, + save_credentials_to_keyring, +) +from linkedin_mcp_server.config.schema import AppConfig +from linkedin_mcp_server.exceptions import CredentialsNotFoundError + +logger = logging.getLogger(__name__) + + +def get_credentials_for_setup() -> Dict[str, str]: + """ + Get LinkedIn credentials for setup purposes. + + Returns: + Dict[str, str]: Dictionary with email and password + + Raises: + CredentialsNotFoundError: If credentials cannot be obtained + """ + config = get_config() + + # First, try configuration (includes environment variables) + if config.linkedin.email and config.linkedin.password: + logger.info("Using LinkedIn credentials from configuration") + return {"email": config.linkedin.email, "password": config.linkedin.password} + + # Second, try keyring if enabled + if config.linkedin.use_keyring: + credentials = get_credentials_from_keyring() + if credentials["email"] and credentials["password"]: + logger.info("Using LinkedIn credentials from keyring") + return {"email": credentials["email"], "password": credentials["password"]} + + # If in non-interactive mode and no credentials found, raise error + if config.chrome.non_interactive: + raise CredentialsNotFoundError( + "No LinkedIn credentials found. Please provide credentials via " + "environment variables (LINKEDIN_EMAIL, LINKEDIN_PASSWORD) for setup." + ) + + # Otherwise, prompt for credentials + return prompt_for_credentials() + + +def prompt_for_credentials() -> Dict[str, str]: + """ + Prompt user for LinkedIn credentials. + + Returns: + Dict[str, str]: Dictionary with email and password + + Raises: + KeyboardInterrupt: If user cancels input + """ + config: AppConfig = get_config() + + print("šŸ”‘ LinkedIn credentials required for setup") + questions = [ + inquirer.Text("email", message="LinkedIn Email"), + inquirer.Password("password", message="LinkedIn Password"), + ] + credentials: Dict[str, str] = inquirer.prompt(questions) + + if not credentials: + raise KeyboardInterrupt("Credential input was cancelled") + + # Store credentials securely in keyring if enabled + if config.linkedin.use_keyring: + if save_credentials_to_keyring(credentials["email"], credentials["password"]): + logger.info("Credentials stored securely in keyring") + else: + logger.warning("Could not store credentials in system keyring") + + return credentials + + +@contextmanager +def temporary_chrome_driver() -> Iterator[webdriver.Chrome]: + """ + Context manager for creating temporary Chrome driver with automatic cleanup. + + Yields: + webdriver.Chrome: Configured Chrome WebDriver instance + + Raises: + Exception: If driver creation fails + """ + config: AppConfig = get_config() + + logger.info("Creating temporary browser for cookie capture...") + + # Set up Chrome options for cookie capture + chrome_options = Options() + if config.chrome.headless: + chrome_options.add_argument("--headless=new") + + # Add essential options + # chrome_options.add_argument("--no-sandbox") + # chrome_options.add_argument("--disable-dev-shm-usage") + # chrome_options.add_argument("--disable-gpu") + # chrome_options.add_argument("--window-size=3456,2234") + + driver = None + try: + # Create temporary driver + chromedriver_path = ( + os.environ.get("CHROMEDRIVER_PATH") or config.chrome.chromedriver_path + ) + + if chromedriver_path: + from selenium.webdriver.chrome.service import Service + + service = Service(executable_path=chromedriver_path) + driver = webdriver.Chrome(service=service, options=chrome_options) + else: + driver = webdriver.Chrome(options=chrome_options) + + driver.set_page_load_timeout(60) + yield driver + + finally: + if driver: + driver.quit() + + +def capture_cookie_from_credentials(email: str, password: str) -> str: + """ + Login with credentials and capture session cookie using temporary driver. + + Args: + email: LinkedIn email + password: LinkedIn password + + Returns: + str: Captured session cookie + + Raises: + Exception: If login or cookie capture fails + """ + with temporary_chrome_driver() as driver: + # Login using linkedin-scraper + from linkedin_scraper import actions + + config: AppConfig = get_config() + interactive: bool = not config.chrome.non_interactive + logger.info(f"Logging in to LinkedIn... Interactive: {interactive}") + actions.login( + driver, + email, + password, + timeout=60, # longer timeout for login (captcha, mobile verification, etc.) + interactive=interactive, # Respect configuration setting + ) + + # Capture cookie + cookie_obj: Dict[str, str] = driver.get_cookie("li_at") + if cookie_obj and cookie_obj.get("value"): + cookie: str = cookie_obj["value"] + logger.info("Successfully captured session cookie") + return cookie + else: + raise Exception("Failed to capture session cookie from browser") + + +def test_cookie_validity(cookie: str) -> bool: + """ + Test if a cookie is valid by attempting to use it with a temporary driver. + + Args: + cookie: LinkedIn session cookie to test + + Returns: + bool: True if cookie is valid, False otherwise + """ + try: + with temporary_chrome_driver() as driver: + from linkedin_mcp_server.drivers.chrome import login_with_cookie + + return login_with_cookie(driver, cookie) + except Exception as e: + logger.warning(f"Cookie validation failed: {e}") + return False + + +def prompt_for_cookie() -> str: + """ + Prompt user to input LinkedIn cookie directly. + + Returns: + str: LinkedIn session cookie + + Raises: + KeyboardInterrupt: If user cancels input + ValueError: If cookie format is invalid + """ + print("šŸŖ Please provide your LinkedIn session cookie") + cookie = inquirer.text("LinkedIn Cookie") + + if not cookie: + raise KeyboardInterrupt("Cookie input was cancelled") + + # Normalize cookie format + if cookie.startswith("li_at="): + cookie: str = cookie.split("li_at=")[1] + + return cookie + + +def run_interactive_setup() -> str: + """ + Run interactive setup to configure authentication. + + Returns: + str: Configured LinkedIn session cookie + + Raises: + Exception: If setup fails + """ + print("šŸ”— LinkedIn MCP Server Setup") + print("Choose how you'd like to authenticate:") + + # Ask user for setup method + setup_method = inquirer.list_input( + "Setup method", + choices=[ + ("I have a LinkedIn cookie", "cookie"), + ("Login with email/password to get cookie", "credentials"), + ], + default="cookie", + ) + + if setup_method == "cookie": + # User provides cookie directly + cookie = prompt_for_cookie() + + # Test the cookie with a temporary driver + print("šŸ” Testing provided cookie...") + if test_cookie_validity(cookie): + # Store the valid cookie + store_authentication(cookie) + logger.info("āœ… Authentication configured successfully") + return cookie + else: + print("āŒ The provided cookie is invalid or expired") + retry = inquirer.confirm( + "Would you like to try with email/password instead?", default=True + ) + if not retry: + raise Exception("Setup cancelled - invalid cookie provided") + + # Fall through to credentials flow + setup_method = "credentials" + + if setup_method == "credentials": + # Get credentials and attempt login with retry + max_retries = 3 + for attempt in range(max_retries): + try: + credentials = get_credentials_for_setup() + + print("šŸ”‘ Logging in to capture session cookie...") + cookie = capture_cookie_from_credentials( + credentials["email"], credentials["password"] + ) + + # Store the captured cookie + store_authentication(cookie) + logger.info("āœ… Authentication configured successfully") + return cookie + + except Exception as e: + logger.error(f"Login failed: {e}") + print(f"āŒ Login failed: {e}") + + if attempt < max_retries - 1: + retry = inquirer.confirm( + "Would you like to try with different credentials?", + default=True, + ) + if not retry: + break + # Clear stored credentials to prompt for new ones + from linkedin_mcp_server.config.providers import ( + clear_credentials_from_keyring, + ) + + clear_credentials_from_keyring() + else: + raise Exception(f"Setup failed after {max_retries} attempts") + + raise Exception("Setup cancelled by user") + + # This should never be reached, but ensures type checker knows all paths are covered + raise Exception("Unexpected setup flow completion") + + +def run_cookie_extraction_setup() -> str: + """ + Run setup specifically for cookie extraction (--get-cookie mode). + + Returns: + str: Captured LinkedIn session cookie for display + + Raises: + Exception: If setup fails + """ + logger.info("šŸ”— LinkedIn MCP Server - Cookie Extraction mode started") + print("šŸ”— LinkedIn MCP Server - Cookie Extraction") + + # Get credentials + credentials: Dict[str, str] = get_credentials_for_setup() + + # Capture cookie + cookie: str = capture_cookie_from_credentials( + credentials["email"], credentials["password"] + ) + + return cookie diff --git a/main.py b/main.py index d070325..10a32ef 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,11 @@ # main.py """ LinkedIn MCP Server - A Model Context Protocol server for LinkedIn integration. + +Clean architecture with clear phase separation: +1. Authentication Setup Phase +2. Driver Management Phase +3. Server Runtime Phase """ import logging @@ -17,14 +22,17 @@ TwoFactorAuthError, ) +from linkedin_mcp_server.authentication import ( + ensure_authentication, + has_authentication, +) from linkedin_mcp_server.cli import print_claude_config - -# Import the new centralized configuration from linkedin_mcp_server.config import get_config -from linkedin_mcp_server.drivers.chrome import initialize_driver -from linkedin_mcp_server.exceptions import LinkedInMCPError +from linkedin_mcp_server.drivers.chrome import close_all_drivers, get_or_create_driver +from linkedin_mcp_server.exceptions import CredentialsNotFoundError, LinkedInMCPError from linkedin_mcp_server.logging_config import configure_logging from linkedin_mcp_server.server import create_mcp_server, shutdown_handler +from linkedin_mcp_server.setup import run_cookie_extraction_setup, run_interactive_setup logger = logging.getLogger(__name__) @@ -43,87 +51,269 @@ def choose_transport_interactive() -> Literal["stdio", "streamable-http"]: ) ] answers = inquirer.prompt(questions) + + if not answers: + raise KeyboardInterrupt("Transport selection cancelled by user") + return answers["transport"] +def get_cookie_and_exit() -> None: + """Get LinkedIn cookie and exit (for Docker setup).""" + config = get_config() + + # Configure logging - prioritize debug mode over non_interactive + configure_logging( + debug=config.server.debug, + json_format=config.chrome.non_interactive and not config.server.debug, + ) + + logger.info("LinkedIn MCP Server - Cookie Extraction mode started") + + try: + # Run cookie extraction setup + cookie = run_cookie_extraction_setup() + + logger.info("Cookie extraction successful") + print("āœ… Login successful!") + print("šŸŖ LinkedIn Cookie extracted:") + print(cookie) + + # Try to copy to clipboard + try: + import pyperclip + + pyperclip.copy(cookie) + print( + "šŸ“‹ Cookie copied to clipboard! Now you can set the LINKEDIN_COOKIE environment variable in your configuration" + ) + except Exception as e: + logger.warning(f"Could not copy to clipboard: {e}") + print("āš ļø Copy the cookie above manually") + + except Exception as e: + logger.error(f"Error getting cookie: {e}") + + # Provide specific guidance for security challenges + error_msg = str(e).lower() + if "security challenge" in error_msg or "captcha" in error_msg: + print("āŒ LinkedIn security challenge detected") + print("šŸ’” Try one of these solutions:") + print( + " 1. Use an existing LinkedIn cookie from your browser instead (see instructions below)" + ) + print( + " 2. Use --no-headless flag (manual installation required, does not work with Docker) and solve the security challenge manually" + ) + print("\nšŸŖ To get your LinkedIn cookie manually:") + print(" 1. Login to LinkedIn in your browser") + print(" 2. Open Developer Tools (F12)") + print(" 3. Go to Application/Storage > Cookies > linkedin.com") + print(" 4. Copy the 'li_at' cookie value") + print(" 5. Set LINKEDIN_COOKIE environment variable or use --cookie flag") + elif "invalid credentials" in error_msg: + print("āŒ Invalid LinkedIn credentials") + print("šŸ’” Please check your email and password") + else: + print("āŒ Failed to obtain cookie - check your credentials") + sys.exit(1) + + sys.exit(0) + + +def ensure_authentication_ready() -> str: + """ + Phase 1: Ensure authentication is ready before any drivers are created. + + Returns: + str: Valid LinkedIn session cookie + + Raises: + CredentialsNotFoundError: If authentication setup fails + """ + config = get_config() + + # Check if authentication already exists + if has_authentication(): + try: + return ensure_authentication() + except CredentialsNotFoundError: + # Authentication exists but might be invalid, continue to setup + pass + + # If in non-interactive mode and no auth, fail immediately + if config.chrome.non_interactive: + raise CredentialsNotFoundError( + "No LinkedIn authentication found. Please provide cookie via " + "environment variable (LINKEDIN_COOKIE) or run with --get-cookie to obtain one." + ) + + # Run interactive setup + logger.info("Setting up LinkedIn authentication...") + return run_interactive_setup() + + +def initialize_driver_with_auth(authentication: str) -> None: + """ + Phase 2: Initialize driver using existing authentication. + + Args: + authentication: LinkedIn session cookie + + Raises: + Various exceptions if driver creation or login fails + """ + config = get_config() + + if config.server.lazy_init: + logger.info( + "Using lazy initialization - driver will be created on first tool call" + ) + return + + logger.info("Initializing Chrome WebDriver and logging in...") + + try: + # Create driver and login with provided authentication + get_or_create_driver(authentication) + logger.info("āœ… Web driver initialized and authenticated successfully") + + except Exception as e: + logger.error(f"Failed to initialize driver: {e}") + raise e + + def main() -> None: - """Initialize and run the LinkedIn MCP server.""" + """Main application entry point with clear phase separation.""" + logger.info("šŸ”— LinkedIn MCP Server šŸ”—") print("šŸ”— LinkedIn MCP Server šŸ”—") print("=" * 40) - # Get configuration using the new centralized system + # Get configuration config = get_config() - # Configure logging + # Handle --get-cookie flag immediately + if config.server.get_cookie: + get_cookie_and_exit() + + # Configure logging - prioritize debug mode over non_interactive configure_logging( debug=config.server.debug, - json_format=config.chrome.non_interactive, # Use JSON format in non-interactive mode + json_format=config.chrome.non_interactive and not config.server.debug, ) logger.debug(f"Server configuration: {config}") - # Initialize the driver with configuration (initialize driver checks for lazy init options) + # Phase 1: Ensure Authentication is Ready try: - initialize_driver() + authentication = ensure_authentication_ready() + print("āœ… Authentication ready") + logger.info("Authentication ready") + except CredentialsNotFoundError as e: + logger.error(f"Authentication setup failed: {e}") + print( + "\nāŒ Authentication required - please provide LinkedIn cookie or credentials" + ) + sys.exit(1) + except KeyboardInterrupt: + print("\n\nšŸ‘‹ Setup cancelled by user") + sys.exit(0) + except Exception as e: + logger.error(f"Unexpected error during authentication setup: {e}") + print("\nāŒ Setup failed - please try again") + sys.exit(1) + + # Phase 2: Initialize Driver (if not lazy) + try: + initialize_driver_with_auth(authentication) + except InvalidCredentialsError as e: + logger.error(f"Driver initialization failed with invalid credentials: {e}") + + # Cookie was already cleared in driver layer + # In interactive mode, try setup again + if not config.chrome.non_interactive and config.server.setup: + print(f"\nāŒ {str(e)}") + print("šŸ”„ Starting interactive setup for new authentication...") + try: + new_authentication = run_interactive_setup() + # Try again with new authentication + initialize_driver_with_auth(new_authentication) + logger.info("āœ… Successfully authenticated with new credentials") + except Exception as setup_error: + logger.error(f"Setup failed: {setup_error}") + print(f"\nāŒ Setup failed: {setup_error}") + sys.exit(1) + else: + print(f"\nāŒ {str(e)}") + if not config.server.lazy_init: + sys.exit(1) except ( LinkedInMCPError, CaptchaRequiredError, - InvalidCredentialsError, SecurityChallengeError, TwoFactorAuthError, RateLimitError, LoginTimeoutError, ) as e: - logger.error( - f"Failed to initialize driver: {str(e)}", - extra={"error_type": type(e).__name__, "error_details": str(e)}, - ) - - # Always terminate if login fails and we're not using lazy initialization + logger.error(f"Driver initialization failed: {e}") + print(f"\nāŒ {str(e)}") if not config.server.lazy_init: - print(f"\nāŒ {str(e)}") - sys.exit(1) - - # In lazy init mode with non-interactive, still exit on error - if config.chrome.non_interactive: sys.exit(1) - else: - print(f"\nāŒ Error: {str(e)}") - print("šŸ’” Tip: Check your credentials and try again.") + except Exception as e: + logger.error(f"Unexpected error during driver initialization: {e}") + print(f"\nāŒ Driver initialization failed: {e}") + if not config.server.lazy_init: sys.exit(1) - # Decide transport - transport = config.server.transport - if config.server.setup: - transport = choose_transport_interactive() + # Phase 3: Server Runtime + try: + # Decide transport + transport = config.server.transport + if config.server.setup: + print("\nšŸš€ Server ready! Choose transport mode:") + transport = choose_transport_interactive() - # Print configuration for Claude if in setup mode and using stdio transport - if config.server.setup and transport == "stdio": - print_claude_config() + # Print configuration for Claude if in setup mode and using stdio transport + if config.server.setup and transport == "stdio": + print_claude_config() - # Create and run the MCP server - mcp = create_mcp_server() + # Create and run the MCP server + mcp = create_mcp_server() - # Start server - print(f"\nšŸš€ Running LinkedIn MCP server ({transport.upper()} mode)...") - if transport == "streamable-http": - print( - f"šŸ“” HTTP server will be available at http://{config.server.host}:{config.server.port}{config.server.path}" - ) - mcp.run( - transport=transport, - host=config.server.host, - port=config.server.port, - path=config.server.path, - ) - else: - mcp.run(transport=transport) + # Start server + print(f"\nšŸš€ Running LinkedIn MCP server ({transport.upper()} mode)...") + if transport == "streamable-http": + print( + f"šŸ“” HTTP server will be available at http://{config.server.host}:{config.server.port}{config.server.path}" + ) + mcp.run( + transport=transport, + host=config.server.host, + port=config.server.port, + path=config.server.path, + ) + else: + mcp.run(transport=transport) + + except KeyboardInterrupt: + print("\n\nšŸ‘‹ Server stopped by user") + exit_gracefully(0) + except Exception as e: + logger.error(f"Server runtime error: {e}") + print(f"\nāŒ Server error: {e}") + exit_gracefully(1) def exit_gracefully(exit_code: int = 0) -> None: """Exit the application gracefully, cleaning up resources.""" print("\nšŸ‘‹ Shutting down LinkedIn MCP server...") + + # Clean up drivers + close_all_drivers() + + # Clean up server shutdown_handler() + sys.exit(exit_code) diff --git a/manifest.json b/manifest.json index 7f49b0d..950a5ba 100644 --- a/manifest.json +++ b/manifest.json @@ -24,8 +24,7 @@ "command": "docker", "args": [ "run", "-i", "--rm", - "-e", "LINKEDIN_EMAIL=${user_config.linkedin_email}", - "-e", "LINKEDIN_PASSWORD=${user_config.linkedin_password}", + "-e", "LINKEDIN_COOKIE=${user_config.linkedin_cookie}", "stickerdaniel/linkedin-mcp-server" ] } @@ -53,7 +52,7 @@ "properties": { "company_url": { "type": "string", - "description": "LinkedIn company URL (e.g., https://www.linkedin.com/company/company-name/)" + "description": "LinkedIn company URL (e.g., https://www.linkedin.com/company/docker/)" } }, "required": ["company_url"] @@ -84,16 +83,9 @@ } ], "user_config": { - "linkedin_email": { - "title": "LinkedIn Email", - "description": "Your LinkedIn account email address", - "type": "string", - "required": true, - "sensitive": false - }, - "linkedin_password": { - "title": "LinkedIn Password", - "description": "Your LinkedIn account password", + "linkedin_cookie": { + "title": "LinkedIn Cookie", + "description": "LinkedIn li_at session cookie. Follow the instructions in the README to get it.", "type": "string", "required": true, "sensitive": true diff --git a/uv.lock b/uv.lock index 527cc63..6a1638f 100644 --- a/uv.lock +++ b/uv.lock @@ -702,7 +702,7 @@ dev = [ [[package]] name = "linkedin-scraper" version = "2.11.5" -source = { git = "https://github.com/stickerdaniel/linkedin_scraper.git#1d6ff82f8b0950b060529b12102a674cfabad1bb" } +source = { git = "https://github.com/stickerdaniel/linkedin_scraper.git#30f448df90af834bafb7d9e4caebfd0032605163" } dependencies = [ { name = "lxml" }, { name = "python-dotenv" },