II. Implementation

Connecting to the MCP Server

Overview

This page shows how to connect to the Gracenote MCP Server directly over HTTPS from your own Python code. Use this approach when building your own application or integration.

Already tested with the STDIO proxy? Great — this page replaces the proxy with a direct connection you control. If you haven't tested yet, try the Quick Start (CLI/Proxy) first.

This page provides a reusable Python module (mcp_utils.py) that handles Cognito authentication, token refresh, and MCP connectivity. This module is imported by your application code (see Solutions for examples).

Prerequisites

Before you begin, ensure you have:

• MCP Server credentials from your Gracenote representative:
  • Cognito username and password
  • Cognito client ID and client secret (note: client IDs differ between environments)
• MCP Server endpoint: gn-video-mcp-server.k8s-p.cloud.gracenote.com:443
• Python 3.11+ installed
• Dependencies: pip install litellm boto3 mcp python-dotenv

Authentication Flow

The Gracenote MCP Server uses AWS Cognito for authentication. The flow is:

1. Your code sends credentials to AWS Cognito (USER_PASSWORD_AUTH flow)
2. Cognito returns a JWT ID token (valid for 1 hour)
3. Your code includes this token in the Authorization header when connecting to the MCP Server
4. When the token expires, your code requests a new one

Code
 
Your App → AWS Cognito → JWT Token → MCP Server (HTTPS)

Key details:

• Tokens expire after 1 hour — long-running processes must refresh tokens
• The SECRET_HASH is required when the Cognito app client has a client secret configured
• All MCP Server connections use HTTPS on port 443

Step 1: Set Up Environment Variables

Create a .env file in your project directory:

Code
 
# LLM API Key (at least one required)
ANTHROPIC_API_KEY=sk-ant-your-key-here
# OPENAI_API_KEY=sk-your-key-here
# GEMINI_API_KEY=your-key-here

# Gracenote MCP Server
MCP_SERVER_HOST=gn-video-mcp-server.k8s-p.cloud.gracenote.com
MCP_SERVER_PORT=443

# AWS Cognito Authentication
COGNITO_USER_POOL_REGION=us-west-2
COGNITO_CLIENT_ID=your-app-client-id
COGNITO_CLIENT_SECRET=your-app-client-secret
COGNITO_USERNAME=service-account-username
COGNITO_PASSWORD=service-account-password

All examples use LiteLLM for universal model routing. Set at least one LLM API key for whichever provider you prefer — you can switch between Claude, GPT-4o, Gemini, and 100+ other models with a single parameter.

Important: Never commit credentials to version control. Use environment variables or a secrets manager in production.

Step 2: Create the Reusable MCP Utilities Module

Create mcp_utils.py — this single file handles Cognito authentication, token refresh, and MCP connectivity. All parameters fall back to environment variables, so you can instantiate with GracenoteClient() if your .env is configured.

Code
 
import os
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import boto3
import hmac
import hashlib
import base64
from typing import Optional
import time


def get_cognito_token(region: str, client_id: str, client_secret: str,
                      username: str, password: str) -> str:
    """Authenticate with AWS Cognito and return a JWT ID token."""
    client = boto3.client('cognito-idp', region_name=region)

    # Compute SECRET_HASH (required when app client has a secret)
    message = username + client_id
    secret_hash = base64.b64encode(
        hmac.new(client_secret.encode(), message.encode(), hashlib.sha256).digest()
    ).decode()

    response = client.initiate_auth(
        ClientId=client_id,
        AuthFlow='USER_PASSWORD_AUTH',
        AuthParameters={
            'USERNAME': username,
            'PASSWORD': password,
            'SECRET_HASH': secret_hash
        }
    )
    return response['AuthenticationResult']['IdToken']


class GracenoteClient:
    """Reusable MCP client with automatic Cognito token refresh.

    Usage as context manager (reads credentials from environment):
        async with GracenoteClient() as client:
            tools = await client.list_tools()

    Usage with explicit credentials:
        async with GracenoteClient(host="...", username="...", password="...") as client:
            tools = await client.list_tools()
    """

    def __init__(
        self,
        host: Optional[str] = None,
        port: Optional[str] = None,
        cognito_region: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None
    ):
        self.host = host or os.getenv("MCP_SERVER_HOST")
        self.port = port or os.getenv("MCP_SERVER_PORT", "443")
        self.cognito_region = cognito_region or os.getenv("COGNITO_USER_POOL_REGION")
        self.client_id = client_id or os.getenv("COGNITO_CLIENT_ID")
        self.client_secret = client_secret or os.getenv("COGNITO_CLIENT_SECRET")
        self.username = username or os.getenv("COGNITO_USERNAME")
        self.password = password or os.getenv("COGNITO_PASSWORD")

        if not all([self.host, self.cognito_region, self.client_id,
                   self.client_secret, self.username, self.password]):
            raise ValueError(
                "Missing required credentials. Provide them as arguments or set environment variables."
            )

        self.url = f"https://{self.host}:{self.port}/mcp"
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.token_acquired_at: Optional[float] = None
        self.token_expires_in = 3600  # Cognito default: 1 hour

    async def connect(self):
        """Authenticate with Cognito and connect to the MCP Server."""
        token = get_cognito_token(
            self.cognito_region, self.client_id,
            self.client_secret, self.username, self.password
        )
        self.token_acquired_at = time.time()

        transport = await self.exit_stack.enter_async_context(
            streamablehttp_client(self.url, headers={
                "Authorization": f"Bearer {token}",
                "Accept": "text/event-stream"
            })
        )

        read_stream, write_stream, _ = transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        await self.session.initialize()

    async def ensure_connected(self):
        """Ensure connection is established and token is valid."""
        if self.session is None:
            await self.connect()
        elif self.token_acquired_at and \
             time.time() - self.token_acquired_at > (self.token_expires_in - 300):
            # Token expires in less than 5 minutes, reconnect
            await self.close()
            await self.connect()

    async def call_tool(self, tool_name: str, arguments: dict):
        """Call an MCP tool, refreshing the token if needed."""
        await self.ensure_connected()
        return await self.session.call_tool(tool_name, arguments=arguments)

    async def list_tools(self):
        """List available MCP tools."""
        await self.ensure_connected()
        return await self.session.list_tools()

    async def tools_for_litellm(self):
        """Get MCP tools in OpenAI/LiteLLM format."""
        tools_result = await self.list_tools()
        return [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        } for tool in tools_result.tools]

    async def close(self):
        """Close the MCP connection."""
        await self.exit_stack.aclose()
        self.session = None

    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

Step 3: Test Your Connection

Create test_connection.py to verify everything works:

Code
 
import asyncio
from dotenv import load_dotenv
from mcp_utils import GracenoteClient

async def main():
    load_dotenv()

    async with GracenoteClient() as client:
        tools = await client.list_tools()
        print(f"Available tools: {[t.name for t in tools.tools]}")
        print("\n[OK] Connection successful! Your LLM can now use the MCP Server.")

if __name__ == "__main__":
    asyncio.run(main())

Run it:

Code
 
python3 test_connection.py

Expected output:

Code
 
Available tools: ['resolve_entities', 'get_availability', 'get_root', 'get_tmsid', 'get_images']

[OK] Connection successful! Your LLM can now use the MCP Server.

Common Authentication Errors

Error	Cause	Solution
NotAuthorizedException	Invalid username/password	Verify credentials from your Gracenote representative
UserNotConfirmedException	Account not activated	Contact Gracenote to confirm your user account
401 Unauthorized	Expired or invalid token	Token expired after 1 hour — GracenoteClient handles this automatically
Could not resolve host	Wrong server hostname	Verify MCP_SERVER_HOST is correct for your environment
Connection timeout	Network/firewall issue	Ensure HTTPS traffic on port 443 is allowed; test with curl -v https://your-mcp-host:443/mcp

For more troubleshooting, see Troubleshooting. For complete application examples that import mcp_utils.py, see the Solutions section.