"""Authentication tool for the OWA MCP server. Provides a `login` tool that accepts session cookies from the user, saves them to disk, loads them into the OWA client, and verifies the session. The user obtains cookies from their browser (DevTools → Application → Cookies), pastes them as a ``name=value`` string (one cookie per line), and passes them to this tool. """ import json from mcp.server.fastmcp import Context from owa_mcp.owa_client import OWAClient, SessionExpiredError from owa_mcp.server import AppContext, mcp from owa_mcp.server_lifecycle import require_client def _get_app_ctx(ctx: Context) -> AppContext: """Extract the AppContext from the MCP lifespan context.""" return ctx.request_context.lifespan_context def _get_client(ctx: Context) -> OWAClient: """Extract the OWAClient from the MCP lifespan context.""" return require_client(_get_app_ctx(ctx).client) def _session_is_valid(client: OWAClient) -> bool: """Quick check: can we reach the inbox?""" try: client._ensure_loaded() payload = { "__type": "GetFolderJsonRequest:#Exchange", "Header": { "__type": "JsonRequestHeaders:#Exchange", "RequestServerVersion": "Exchange2013", }, "Body": { "__type": "GetFolderRequest:#Exchange", "FolderShape": { "__type": "FolderResponseShape:#Exchange", "BaseShape": "IdOnly", }, "FolderIds": [ { "__type": "DistinguishedFolderId:#Exchange", "Id": "inbox", } ], }, } data = client.request("GetFolder", payload) items = client.extract_items(data) return bool(items) except Exception: return False @mcp.tool() async def login( cookies: str, ctx: Context = None, ) -> str: """Authenticate to Exchange OWA using session cookies from the browser. Paste the cookies you copied from your browser (DevTools → Application → Cookies) as a ``name=value`` string — one cookie per line. Example cookies string:: X-OWA-CANARY=abc123 CookieAuth1=def456 ... Args: cookies: Session cookies in ``name=value`` format, one per line. Obtain from browser DevTools → Application → Cookies → copy all cookies as ``name=value`` pairs. Returns: JSON result with success status and message. """ client = _get_client(ctx) # Parse and validate cookies cookies_str = cookies.strip() if not cookies_str: return json.dumps( { "success": False, "error": "Cookies string is empty. Paste your browser cookies as name=value pairs, one per line.", } ) # Save to disk try: client.cookie_file.write_text(cookies_str + "\n") except Exception as e: return json.dumps( {"success": False, "error": f"Failed to write cookie file: {e}"} ) # Load into memory try: client.load_cookies_from_string(cookies_str) except SessionExpiredError as e: return json.dumps({"success": False, "error": str(e)}) # Verify session if _session_is_valid(client): return json.dumps( { "success": True, "message": f"Logged in. Session cookies saved to {client.cookie_file}.", } ) else: return json.dumps( { "success": False, "error": "Cookies loaded but session verification failed. The cookies may have expired.", } )