Paste one of the following code blocks into the custom Python code based on your authentication method.Tip:
You can copy the code block from the Configuration tab of the MCP server details page.
- - If you use the OAuth Authorization Code flow, paste the following code block and replace the <client ID> with the client ID, the authorization, token, and resource URLs with actual endpoint URLs, and <server endpoint URL> with the MCP server endpoint URL:
- - If you use JWT authentication or the OAuth Client Credentials flow, paste the following code block and replace <endpoint URL> with the endpoint URL and <access token> with the bearer token:
- - If you use session-based authentication, paste the following code block and replace <endpoint URL> with the endpoint URL and <session ID> with the session ID:
# Requires MCP Python SDK v2: pip install "mcp>=2"
import base64
import hashlib
import secrets
import urllib.parse
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# OAuth Authorization Code grant with PKCE -- verified working example (no MCP SDK helper
# performs this automatically today, unlike the mcp-remote npx wrapper used by other
# clients in this file). PKCE (code_verifier/code_challenge) stands in for a client
# secret for this public-client flow -- no such secret is used anywhere below.
CLIENT_ID = "<client ID>"
CALLBACK_PORT = 3000
REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}/callback"
AUTHORIZE_ENDPOINT = "<authorization endpoint URL>"
TOKEN_ENDPOINT = "<token endpoint URL>"
RESOURCE = "<resource endpoint URL>"
auth_code = None
class OAuthCallbackHandler(BaseHTTPRequestHandler):
"""Local HTTP listener that captures the authorization code from the redirect."""
def do_GET(self):
global auth_code
query_params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
if "code" in query_params:
auth_code = query_params["code"][0]
self.send_response(200)
self.end_headers()
self.wfile.write(b"Authentication successful! You can close this window.")
else:
self.send_response(400)
self.end_headers()
def generate_pkce_pair():
verifier = secrets.token_urlsafe(64)
digest = hashlib.sha256(verifier.encode("utf-8")).digest()
challenge = base64.urlsafe_b64encode(digest).decode("utf-8").replace("=", "")
return verifier, challenge
def obtain_oauth21_token() -> str:
global auth_code
verifier, challenge = generate_pkce_pair()
server = HTTPServer(("127.0.0.1", CALLBACK_PORT), OAuthCallbackHandler)
server_thread = Thread(target=server.handle_request)
server_thread.start()
params = {
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"code_challenge": challenge,
"code_challenge_method": "S256",
"resource": RESOURCE,
}
webbrowser.open(f"{AUTHORIZE_ENDPOINT}?{urllib.parse.urlencode(params)}")
server_thread.join()
if not auth_code:
raise RuntimeError("OAuth authentication failed: no authorization code received.")
with httpx2.Client() as client:
response = client.post(
TOKEN_ENDPOINT,
# A confidential client would additionally send a "client_secret" key here; this
# public/PKCE client relies solely on code_verifier below instead.
data={
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": auth_code,
"redirect_uri": REDIRECT_URI,
"code_verifier": verifier,
"resource": RESOURCE,
},
)
response.raise_for_status()
return response.json()["access_token"]
# Create an MCP client instance, presenting the token fetched above as a Bearer header
access_token = obtain_oauth21_token()
http_client = httpx2.AsyncClient(
headers={"Authorization": f"Bearer {access_token}"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<server endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()
import base64
import hashlib
import secrets
import urllib.parse
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# OAuth Authorization Code grant with PKCE -- verified working example (no MCP SDK helper
# performs this automatically today, unlike the mcp-remote npx wrapper used by other
# clients in this file). PKCE (code_verifier/code_challenge) stands in for a client
# secret for this public-client flow -- no such secret is used anywhere below.
CLIENT_ID = "<client ID>"
CALLBACK_PORT = 3000
REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}/callback"
AUTHORIZE_ENDPOINT = "<authorization endpoint URL>"
TOKEN_ENDPOINT = "<token endpoint URL>"
RESOURCE = "<resource endpoint URL>"
auth_code = None
class OAuthCallbackHandler(BaseHTTPRequestHandler):
"""Local HTTP listener that captures the authorization code from the redirect."""
def do_GET(self):
global auth_code
query_params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
if "code" in query_params:
auth_code = query_params["code"][0]
self.send_response(200)
self.end_headers()
self.wfile.write(b"Authentication successful! You can close this window.")
else:
self.send_response(400)
self.end_headers()
def generate_pkce_pair():
verifier = secrets.token_urlsafe(64)
digest = hashlib.sha256(verifier.encode("utf-8")).digest()
challenge = base64.urlsafe_b64encode(digest).decode("utf-8").replace("=", "")
return verifier, challenge
def obtain_oauth21_token() -> str:
global auth_code
verifier, challenge = generate_pkce_pair()
server = HTTPServer(("127.0.0.1", CALLBACK_PORT), OAuthCallbackHandler)
server_thread = Thread(target=server.handle_request)
server_thread.start()
params = {
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"code_challenge": challenge,
"code_challenge_method": "S256",
"resource": RESOURCE,
}
webbrowser.open(f"{AUTHORIZE_ENDPOINT}?{urllib.parse.urlencode(params)}")
server_thread.join()
if not auth_code:
raise RuntimeError("OAuth authentication failed: no authorization code received.")
with httpx2.Client() as client:
response = client.post(
TOKEN_ENDPOINT,
# A confidential client would additionally send a "client_secret" key here; this
# public/PKCE client relies solely on code_verifier below instead.
data={
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": auth_code,
"redirect_uri": REDIRECT_URI,
"code_verifier": verifier,
"resource": RESOURCE,
},
)
response.raise_for_status()
return response.json()["access_token"]
# Create an MCP client instance, presenting the token fetched above as a Bearer header
access_token = obtain_oauth21_token()
http_client = httpx2.AsyncClient(
headers={"Authorization": f"Bearer {access_token}"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<server endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()
# Requires MCP Python SDK v2: pip install "mcp>=2"
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# Create an MCP client instance, pointing to your deployed server URL
http_client = httpx2.AsyncClient(
headers={"Authorization":"Bearer <access token>"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# Create an MCP client instance, pointing to your deployed server URL
http_client = httpx2.AsyncClient(
headers={"Authorization":"Bearer <access token>"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()
# Requires MCP Python SDK v2: pip install "mcp>=2"
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# Create an MCP client instance, pointing to your deployed server URL
http_client = httpx2.AsyncClient(
headers={"IDS-SESSION-ID":"<session ID>"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
# Create an MCP client instance, pointing to your deployed server URL
http_client = httpx2.AsyncClient(
headers={"IDS-SESSION-ID":"<session ID>"},
timeout=httpx2.Timeout(30.0, read=300.0),
)
transport = streamable_http_client(
url="<endpoint URL>",
http_client=http_client,
)
# By default, Client negotiates the latest protocol version (mode="auto"): it probes
# server/discover and falls back to the initialize handshake on legacy servers.
# If your workflow requires legacy mode (pre-2026 / initialize handshake),
# pass mode="legacy" to force it: Client(transport, mode="legacy")
mcp_client = Client(transport)
# Usage (inside an async function): close http_client together with the client
# async with http_client, mcp_client:
# result = await mcp_client.list_tools()