This section describes the Machine-to-Machine (M2M) authentication flow for connecting to a Gracenote Video MCP server. This approach enables backend services to authenticate without user interaction. Gracenote Video MCP server uses AWS Cognito service as an OAuth provider.
The following sample code demonstrates the acquisition of the Cognito JWT token given the client credentials.
Code
import base64import hashlibimport hmacimport boto3def compute_secret_hash(username: str, client_id: str, client_secret: str) -> str: """ Compute SECRET_HASH for Cognito authentication. Required when the Cognito App Client has a client secret configured. The hash is HMAC-SHA256 of (username + client_id) using client_secret as key. """ message = username + client_id dig = hmac.new( key=client_secret.encode('utf-8'), msg=message.encode('utf-8'), digestmod=hashlib.sha256 ).digest() return base64.b64encode(dig).decode()def get_cognito_id_token( region: str, client_id: str, client_secret: str, username: str, password: str,) -> str: """ Obtain an ID token from Cognito using USER_PASSWORD_AUTH flow. Args: region: AWS region (e.g., 'us-west-2') client_id: Cognito App Client ID client_secret: Cognito App Client Secret (empty string if none) username: Service account username password: Service account password Returns: JWT ID token string Raises: botocore.exceptions.ClientError: If authentication fails """ client = boto3.client('cognito-idp', region_name=region) auth_params = { 'USERNAME': username, 'PASSWORD': password, } # Include SECRET_HASH if client secret is configured if client_secret: auth_params['SECRET_HASH'] = compute_secret_hash( username, client_id, client_secret ) response = client.initiate_auth( ClientId=client_id, AuthFlow='USER_PASSWORD_AUTH', AuthParameters=auth_params, ) return response['AuthenticationResult']['IdToken']
MCP Client Authentication Example
The following sample code demonstrates the authentication with the MCP server using the get_cognito_id_token() function provided above.
Code
import osimport asynciofrom contextlib import AsyncExitStackfrom mcp import ClientSessionfrom mcp.client.streamable_http import streamablehttp_clientclass MCPClient: """ MCP Client with Cognito M2M authentication. """ def __init__( self, mcp_host: str, mcp_port: str, cognito_region: str, cognito_client_id: str, cognito_client_secret: str, cognito_username: str, cognito_password: str, ): # Build MCP server URL scheme = "https" if mcp_port == "443" else "http" self.url = f"{scheme}://{mcp_host}:{mcp_port}/mcp" # Store Cognito credentials self.cognito_region = cognito_region self.cognito_client_id = cognito_client_id self.cognito_client_secret = cognito_client_secret self.cognito_username = cognito_username self.cognito_password = cognito_password # Session management self.headers = {"Accept": "text/event-stream"} self.session = None self.exit_stack = AsyncExitStack() def _get_token(self) -> str: """Fetch ID token from Cognito.""" return get_cognito_id_token( region=self.cognito_region, client_id=self.cognito_client_id, client_secret=self.cognito_client_secret, username=self.cognito_username, password=self.cognito_password, ) async def connect(self): """ Connect to the MCP server with Cognito authentication. """ # Get fresh token and set Authorization header token = self._get_token() self.headers["Authorization"] = f"Bearer {token}" # Establish MCP connection transport = await self.exit_stack.enter_async_context( streamablehttp_client(self.url, headers=self.headers) ) read_stream, write_stream, _ = transport self.session = await self.exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await self.session.initialize() print(f"Connected to MCP Server at {self.url}") async def list_tools(self): """List available tools from the MCP server.""" if not self.session: raise RuntimeError("Not connected. Call connect() first.") result = await self.session.list_tools() return result.tools async def call_tool(self, tool_name: str, arguments: dict): """Call a tool on the MCP server.""" if not self.session: raise RuntimeError("Not connected. Call connect() first.") result = await self.session.call_tool(tool_name, arguments=arguments) return result async def close(self): """Close the connection.""" await self.exit_stack.aclose()# Usage exampleasync def main(): client = MCPClient( mcp_host=os.getenv("MCP_SERVER_HOST"), mcp_port=os.getenv("MCP_SERVER_PORT", "443"), cognito_region=os.getenv("COGNITO_USER_POOL_REGION", "us-west-2"), cognito_client_id=os.getenv("COGNITO_CLIENT_ID"), cognito_client_secret=os.getenv("COGNITO_CLIENT_SECRET", ""), cognito_username=os.getenv("COGNITO_USERNAME"), cognito_password=os.getenv("COGNITO_PASSWORD"), ) try: await client.connect() # List available tools tools = await client.list_tools() print(f"Available tools: {[t.name for t in tools]}") # Call a tool (example) # result = await client.call_tool("my_tool", {"param": "value"}) # print(result) finally: await client.close()if __name__ == "__main__": asyncio.run(main())
Token Expiration
Tokens expire after 1 hour by default. For long-running processes:
Refresh before expiration: Get a new token before the current one expires (sample code)
Handle 401 errors: Catch authentication failures and retry with a fresh token
Code
import timeclass MCPClientWithRefresh(MCPClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.token_acquired_at = None self.token_expires_in = 3600 # 1 hour default def _should_refresh_token(self) -> bool: """Check if token needs refresh (5 min buffer).""" if not self.token_acquired_at: return True elapsed = time.time() - self.token_acquired_at return elapsed > (self.token_expires_in - 300) # 5 min buffer async def ensure_valid_token(self): """Refresh token if needed.""" if self._should_refresh_token(): token = self._get_token() self.headers["Authorization"] = f"Bearer {token}" self.token_acquired_at = time.time()
Error Handling
Common Errors
Error
Cause
Solution
NotAuthorizedException
Invalid username/password
Verify credentials
UserNotConfirmedException
User not confirmed
Confirm user is validated with Gracenote
401 Unauthorized
Invalid/expired token
Get fresh token
invalid_token
Server rejects token
Reach out to Gracenote
Error Handling Example
Code
from botocore.exceptions import ClientErrorfrom botocore.exceptions import ClientErrortry: token = get_cognito_id_token(...)except ClientError as e: error_code = e.response['Error']['Code'] if error_code == 'NotAuthorizedException': print("Invalid credentials - check username/password") elif error_code == 'UserNotConfirmedException': print("User must be confirmed before authentication") elif error_code == 'UserNotFoundException': print("User does not exist") else: print(f"Authentication failed: {e}") raise
Security Best Practices
Store credentials securely: Use AWS Secrets Manager, HashiCorp Vault, or environment variables (not in code)
Use HTTPS: Always connect to MCP servers over HTTPS in production
Troubleshooting
The following is a self contained Python example that obtains the Cognito JWT token given the client credentials and inspects its claims, without verification (for debugging purposes). To run this example, create a .env file in the local directory, containing the environment variables described above.
Code
import osimport boto3import hmacimport hashlibimport base64import jsonfrom dotenv import load_dotenvdef compute_secret_hash(username: str, client_id: str, client_secret: str) -> str: """ Compute SECRET_HASH for Cognito authentication. Required when the Cognito App Client has a client secret configured. """ message = username + client_id dig = hmac.new( key=client_secret.encode('utf-8'), msg=message.encode('utf-8'), digestmod=hashlib.sha256 ).digest() return base64.b64encode(dig).decode()def get_cognito_id_token( region: str, client_id: str, client_secret: str, username: str, password: str,) -> str: """ Obtain an ID token from Cognito using USER_PASSWORD_AUTH flow. """ if not all([region, client_id, username, password]): raise ValueError("Missing required Cognito configuration (Region, Client ID, Username, or Password).") print(f"Attempting authentication for user: {username} in region: {region}") client = boto3.client('cognito-idp', region_name=region) auth_params = { 'USERNAME': username, 'PASSWORD': password, } # Include SECRET_HASH if client secret is configured if client_secret: auth_params['SECRET_HASH'] = compute_secret_hash( username, client_id, client_secret ) try: response = client.initiate_auth( ClientId=client_id, AuthFlow='USER_PASSWORD_AUTH', AuthParameters=auth_params, ) return response['AuthenticationResult']['IdToken'] except client.exceptions.NotAuthorizedException: raise PermissionError("Authentication failed: Incorrect username or password.") except client.exceptions.UserNotFoundException: raise PermissionError("Authentication failed: User does not exist.") except Exception as e: raise RuntimeError(f"An error occurred during authentication: {str(e)}")def decode_jwt_payload(token: str) -> dict: """Decode JWT payload for debugging (no signature verification).""" parts = token.split('.') if len(parts) != 3: raise ValueError("Invalid JWT format") payload_b64 = parts[1] # Add padding for base64 padding = 4 - len(payload_b64) % 4 if padding != 4: payload_b64 += '=' * padding payload_bytes = base64.urlsafe_b64decode(payload_b64) return json.loads(payload_bytes)def main(): # Load environment variables from .env file load_dotenv() # Fetch configuration region = os.getenv("COGNITO_USER_POOL_REGION", "us-west-2") client_id = os.getenv("COGNITO_CLIENT_ID") client_secret = os.getenv("COGNITO_CLIENT_SECRET", "") username = os.getenv("COGNITO_USERNAME") password = os.getenv("COGNITO_PASSWORD") try: # Acquire Token token = get_cognito_id_token( region=region, client_id=client_id, client_secret=client_secret, username=username, password=password ) print("\n--- Authentication Successful ---") print("ID Token acquired:") print("-" * 20) print(token) print("-" * 20) claims = decode_jwt_payload(token) print(f"Issuer: {claims.get('iss')}") print(f"Audience: {claims.get('aud')}") print(f"Expires: {claims.get('exp')}") except Exception as e: print(f"\n❌ Error: {e}")if __name__ == "__main__": main()
