Streamable HTTP Client Examples

Streamable HTTP is a transport mechanism in the Model Context Protocol (MCP) that enables client-server communication over HTTP. Unlike the stdio transport (used for local process communication), streamable HTTP allows you to connect to remote MCP servers over the network.

Why Streamable HTTP?

Remote access: Connect to MCP servers hosted anywhere on the internet
Stateless or stateful: Supports both session-based and stateless communication via Mcp-Session-Id header
Authentication: Works with standard HTTP auth mechanisms (Bearer tokens, OAuth)

Client Examples

TypeScript

npm install @modelcontextprotocol/sdk

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';

const ENDPOINT_URL = new URL('https://glama.ai/endpoints/xxx/mcp');
const AUTH_TOKEN = 'Bearer mcp_k1.xxx.xxx';

const client = new Client({
  name: 'example-client',
  version: '1.0.0',
});

const transport = new StreamableHTTPClientTransport(ENDPOINT_URL, {
  requestInit: {
    headers: {
      Authorization: AUTH_TOKEN,
    },
  },
});

try {
  console.log('Connecting to MCP server...');
  await client.connect(transport);
  console.log('Connected successfully!\n');

  console.log('=== Available Tools ===');
  const tools = await client.listTools();
  for (const tool of tools.tools) {
    console.log(`- ${tool.name}: ${tool.description}`);
  }
} catch (error) {
  console.error('Error:', error);
} finally {
  await client.close();
  console.log('\nConnection closed.');
}

Python

pip install mcp

import asyncio
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession

ENDPOINT_URL = "https://glama.ai/endpoints/xxx/mcp"
AUTH_TOKEN = "mcp_k1.xxx.xxx"

async def main():
    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
    
    async with streamablehttp_client(ENDPOINT_URL, headers=headers) as (
        read_stream,
        write_stream,
        _,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            
            print("=== Available Tools ===")
            tools = await session.list_tools()
            for tool in tools.tools:
                print(f"- {tool.name}: {tool.description}")

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

That's it. Connect, initialize, call tools.