Transports API Reference

Complete API documentation for all transport implementations in OmniMCP.

Transport Interface

All transports implement the base Transport interface:

interface Transport {
  // Connection management
  connect(options: any): Promise<void>;
  disconnect(): Promise<void>;
  isConnected(): boolean;
  
  // Message handling
  send(message: any): Promise<void>;
  onMessage(handler: (message: any) => void): void;
  
  // Event handling
  onError(handler: (error: Error) => void): void;
  onClose(handler: () => void): void;
  onOpen(handler: () => void): void;
  
  // Optional features
  ping?(): Promise<void>;
  setKeepAlive?(enabled: boolean, delay?: number): void;
}

StdioTransport

Transport for communication via standard input/output streams.

Constructor

new StdioTransport(options?: StdioTransportOptions)

Options

interface StdioTransportOptions {
  encoding?: BufferEncoding;      // Default: 'utf8'
  lineDelimiter?: string;         // Default: '\n'
  messageDelimiter?: string;      // Default: '\n\n'
  maxMessageSize?: number;        // Default: 1MB
  processCleanup?: boolean;       // Default: true
}

Connection Options

interface StdioConnectionOptions {
  command: string;                // Command to execute
  args?: string[];                // Command arguments
  env?: Record<string, string>;   // Environment variables
  cwd?: string;                   // Working directory
  shell?: boolean | string;       // Use shell execution
  timeout?: number;               // Connection timeout (ms)
  windowsHide?: boolean;          // Hide window on Windows
}

Methods

// Get the underlying child process
getProcess(): ChildProcess | undefined;

// Send signal to process
signal(signal: NodeJS.Signals): void;

// Check if process is running
isProcessRunning(): boolean;

Events

transport.on('processExit', (code: number, signal: string) => {
  console.log(`Process exited with code ${code}`);
});

transport.on('processError', (error: Error) => {
  console.error('Process error:', error);
});

Example

const transport = new StdioTransport({
  encoding: 'utf8',
  maxMessageSize: 5 * 1024 * 1024 // 5MB
});

await transport.connect({
  command: 'node',
  args: ['server.js', '--verbose'],
  env: {
    ...process.env,
    DEBUG: 'true'
  }
});

transport.onMessage((message) => {
  console.log('Received:', message);
});

await transport.send({ 
  jsonrpc: '2.0', 
  method: 'initialize',
  id: 1 
});

HTTPTransport

Transport for HTTP-based communication with MCP servers.

Constructor

new HTTPTransport(options?: HTTPTransportOptions)

Options

interface HTTPTransportOptions {
  timeout?: number;               // Request timeout (ms)
  retries?: number;               // Number of retries
  retryDelay?: number;            // Delay between retries (ms)
  keepAlive?: boolean;            // Use keep-alive
  maxSockets?: number;            // Max concurrent sockets
  proxy?: string | URL;           // HTTP proxy URL
  rejectUnauthorized?: boolean;   // Verify SSL certificates
  agent?: http.Agent;             // Custom HTTP agent
}

Connection Options

interface HTTPConnectionOptions {
  url: string | URL;              // Server URL
  headers?: Record<string, string>; // Custom headers
  auth?: {                        // Authentication
    username: string;
    password: string;
  } | {
    bearer: string;
  };
  query?: Record<string, string>; // Query parameters
}

Methods

// Set custom headers
setHeaders(headers: Record<string, string>): void;

// Get current headers
getHeaders(): Record<string, string>;

// Make raw HTTP request
request(options: RequestOptions): Promise<any>;

Example

const transport = new HTTPTransport({
  timeout: 30000,
  retries: 3,
  keepAlive: true
});

await transport.connect({
  url: 'https://api.example.com/mcp',
  headers: {
    'X-API-Version': '2024-11-05'
  },
  auth: {
    bearer: process.env.API_TOKEN
  }
});

// Add runtime headers
transport.setHeaders({
  'X-Request-ID': generateRequestId()
});

SSETransport

Transport for Server-Sent Events communication.

Constructor

new SSETransport(options?: SSETransportOptions)

Options

interface SSETransportOptions {
  reconnect?: boolean;            // Auto-reconnect on disconnect
  reconnectDelay?: number;        // Initial reconnect delay (ms)
  maxReconnectDelay?: number;     // Max reconnect delay (ms)
  reconnectMultiplier?: number;   // Backoff multiplier
  timeout?: number;               // Connection timeout (ms)
  withCredentials?: boolean;      // Send cookies
  heartbeatInterval?: number;     // Heartbeat check interval (ms)
}

Connection Options

interface SSEConnectionOptions {
  url: string | URL;              // SSE endpoint URL
  headers?: Record<string, string>; // Custom headers
  lastEventId?: string;           // Resume from event ID
  retry?: number;                 // Retry interval (ms)
}

Methods

// Get current connection state
getState(): 'connecting' | 'open' | 'closed';

// Get last event ID
getLastEventId(): string | undefined;

// Force reconnection
reconnect(): Promise<void>;

Events

transport.on('retry', (attempt: number) => {
  console.log(`Reconnection attempt ${attempt}`);
});

transport.on('eventType', (data: any) => {
  console.log('Custom event:', data);
});

Example

const transport = new SSETransport({
  reconnect: true,
  reconnectDelay: 1000,
  maxReconnectDelay: 30000
});

await transport.connect({
  url: 'https://api.example.com/mcp/events',
  headers: {
    'Authorization': 'Bearer ' + token
  }
});

// Handle specific event types
transport.on('toolUpdate', (data) => {
  console.log('Tool updated:', data);
});

// Send messages via companion HTTP endpoint
await transport.send({
  endpoint: '/mcp/commands',
  data: { method: 'tools/call', params: {...} }
});

Custom Transport Development

Base Transport Class

import { Transport, TransportEvents } from '@omnimcp/core';
import { EventEmitter } from 'events';

abstract class BaseTransport extends EventEmitter implements Transport {
  protected connected = false;
  
  abstract connect(options: any): Promise<void>;
  abstract disconnect(): Promise<void>;
  abstract send(message: any): Promise<void>;
  
  isConnected(): boolean {
    return this.connected;
  }
  
  onMessage(handler: (message: any) => void): void {
    this.on('message', handler);
  }
  
  onError(handler: (error: Error) => void): void {
    this.on('error', handler);
  }
  
  onClose(handler: () => void): void {
    this.on('close', handler);
  }
  
  onOpen(handler: () => void): void {
    this.on('open', handler);
  }
  
  protected emitMessage(message: any): void {
    this.emit('message', message);
  }
  
  protected emitError(error: Error): void {
    this.emit('error', error);
  }
  
  protected setConnected(connected: boolean): void {
    this.connected = connected;
    this.emit(connected ? 'open' : 'close');
  }
}

Transport Registration

import { TransportRegistry } from '@omnimcp/core';

// Register custom transport
TransportRegistry.register('custom', CustomTransport);

// Use in client
const client = new MCPClient('app', '1.0.0');
await client.connect({
  type: 'custom',
  options: { /* custom options */ }
});

Transport Utilities

Message Framing

import { MessageFramer } from '@omnimcp/core';

const framer = new MessageFramer({
  delimiter: '\n\n',
  maxMessageSize: 1024 * 1024
});

// Frame a message
const framed = framer.frame({ 
  jsonrpc: '2.0', 
  method: 'test' 
});

// Parse incoming data
framer.on('message', (message) => {
  console.log('Parsed message:', message);
});

framer.push(incomingData);

Connection Pooling

import { TransportPool } from '@omnimcp/core';

const pool = new TransportPool({
  transport: HTTPTransport,
  size: 5,
  connectionOptions: {
    url: 'https://api.example.com/mcp'
  }
});

await pool.initialize();

const transport = await pool.acquire();
try {
  await transport.send(message);
} finally {
  pool.release(transport);
}

Next Steps

• Learn about type definitions
• See transport concepts
• Explore advanced patterns