complete_authentication
Finalize OAuth authentication for Kroger by processing the authorization callback URL after user consent in the browser.
Instructions
Complete the OAuth flow using the redirect URL from Kroger.
After opening the auth URL in your browser and authorizing the app,
you'll be redirected to a callback URL. Copy that entire URL and
pass it to this tool to complete the authentication process.
Args:
redirect_url: The full URL from your browser after authorization
Returns:
Dictionary indicating authentication status
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| redirect_url | Yes |
Implementation Reference
- src/kroger_mcp/tools/auth.py:89-195 (handler)The core handler function for the 'complete_authentication' tool. It processes the OAuth redirect URL, validates the state parameter, extracts the authorization code, and exchanges it for access tokens using PKCE verification. Handles errors and stores no persistent state beyond globals.
@mcp.tool() async def complete_authentication(redirect_url: str, ctx: Context = None) -> Dict[str, Any]: """ Complete the OAuth flow using the redirect URL from Kroger. After opening the auth URL in your browser and authorizing the app, you'll be redirected to a callback URL. Copy that entire URL and pass it to this tool to complete the authentication process. Args: redirect_url: The full URL from your browser after authorization Returns: Dictionary indicating authentication status """ global _pkce_params, _auth_state if not _pkce_params or not _auth_state: if ctx: await ctx.error("Authentication flow not started") return { "error": True, "message": "Authentication flow not started. Please use start_authentication first." } try: # Parse the redirect URL parsed_url = urlparse(redirect_url) query_params = parse_qs(parsed_url.query) # Extract code and state if 'code' not in query_params: if ctx: await ctx.error("Authorization code not found in redirect URL") return { "error": True, "message": "Authorization code not found in redirect URL. Please check the URL and try again." } auth_code = query_params['code'][0] received_state = query_params.get('state', [None])[0] # Verify state parameter to prevent CSRF attacks if received_state != _auth_state: if ctx: await ctx.error(f"State mismatch: expected {_auth_state}, got {received_state}") return { "error": True, "message": "State parameter mismatch. This could indicate a CSRF attack. Please try authenticating again." } # Get client credentials client_id = os.environ.get("KROGER_CLIENT_ID") client_secret = os.environ.get("KROGER_CLIENT_SECRET") redirect_uri = os.environ.get("KROGER_REDIRECT_URI", "http://localhost:8000/callback") if not client_id or not client_secret: if ctx: await ctx.error("Missing Kroger API credentials") return { "error": True, "message": "Missing Kroger API credentials. Please set KROGER_CLIENT_ID and KROGER_CLIENT_SECRET." } # Initialize Kroger API client kroger = KrogerAPI() # Exchange the authorization code for tokens with the code verifier if ctx: await ctx.info(f"Exchanging authorization code for tokens with code_verifier") # Use the code_verifier from the PKCE parameters token_info = kroger.authorization.get_token_with_authorization_code( auth_code, code_verifier=_pkce_params["code_verifier"] ) # Clear PKCE parameters and state after successful exchange _pkce_params = None _auth_state = None if ctx: await ctx.info(f"Authentication successful!") # Return success response return { "success": True, "message": "Authentication successful! You can now use Kroger API tools that require authentication.", "token_info": { "expires_in": token_info.get("expires_in"), "token_type": token_info.get("token_type"), "scope": token_info.get("scope"), "has_refresh_token": "refresh_token" in token_info } } except Exception as e: error_message = str(e) if ctx: await ctx.error(f"Authentication error: {error_message}") return { "error": True, "message": f"Authentication failed: {error_message}" } - src/kroger_mcp/tools/auth_tools.py:7-10 (registration)Registers the authentication tools by calling register_auth_tools from auth.py.
def register_tools(mcp): """Register authentication tools with the FastMCP server""" register_auth_tools(mcp) - src/kroger_mcp/server.py:78-78 (registration)Invokes the registration of auth tools in the main server setup, making complete_authentication available as an MCP tool.
auth_tools.register_tools(mcp) - src/kroger_mcp/tools/auth.py:25-28 (registration)Defines and registers both start_authentication and complete_authentication tools using @mcp.tool() decorators within the registration function.
def register_auth_tools(mcp): """Register authentication-specific tools with the FastMCP server""" @mcp.tool()