Feature - Enable Streaming Capabilities For Tool Calls That Send Real-Time Notificaitons

[Amrrx]

MCP Streaming Support

Adds real-time streaming capabilities to MCP (Model Context Protocol) tool integrations with enhanced connection management and SSE integration.

Changes

• Enhanced MCP client connection management with SSE integration
• Added streaming support for MCP tool execution with real-time feedback
• Implemented server-side event streaming for MCP tool responses
• Added connection lifecycle management with automatic cleanup

Server Capabilities Schema

Server advertises streaming support via capabilities response:

// Server capabilities detection const capabilities = client.getServerCapabilities() const hasStreaming = capabilities?.notifications?.streaming === true || capabilities?.experimental?.notifications?.streaming === true

Expected server capabilities format: { "capabilities": { "notifications": { "streaming": true } } }

Alternative experimental format: { "capabilities": { "experimental": { "notifications": { "streaming": true } } } }

Tool Capabilities Schema

MCP tools declare streaming support via annotations:

// Tool definition with streaming annotation { "name": "tool_name", "description": "Tool description", "annotations": { "streaming_enabled": true, "notification_types": ["task_completion", "progress", "logging"] }, "inputSchema": { "type": "object", "properties": { /* tool parameters */ } } }

Tool streaming annotations:

• streaming_enabled: boolean - Enables streaming for this tool
• notification_types: string[] - Completion signal types (default: ["task_completion"])

SSE Streaming Events

Tools with streaming support emit events via IServerSideEventStreamer:

interface IServerSideEventStreamer { // Core streaming methods streamTokenEvent(chatId: string, data: string): void streamToolEvent(chatId: string, data: any): void streamCustomEvent(chatId: string, eventType: string, data: any): void

// MCP connection management
addMcpConnection?(chatId: string, toolName?: string): void
removeMcpConnection?(chatId: string, toolName?: string): void
markMcpConnectionCompleting?(chatId: string, toolName?: string): void
hasMcpConnections?(chatId: string): boolean

}

Notification Handling Schema

MCP notifications follow the LoggingMessageNotificationSchema:

// Notification structure { "method": "logging/message", "params": { "data": "notification message", "logger": "task_completion" // Used for completion detection } }

Configuration

const MCP_STREAMING_CONFIG = { DEFAULT_COMPLETION_TIMEOUT: 600000, // 10 minutes fallback NOTIFICATION_DELAY: 1000, // 1 second cleanup delay SUPPORTED_NOTIFICATION_TYPES: ['logging/message', 'progress'], STREAMING_MARKER: '[MCP Streaming]' // UI indicator }

Technical Implementation

• Transport Layer: Supports StdioClientTransport, SSEClientTransport, and StreamableHTTPClientTransport
• Connection Management: Per-tool connection tracking with automatic cleanup
• Completion Detection: Based on notification logger types from tool annotations
• Fallback Support: Graceful degradation for non-streaming servers
• Progress Tokens: Unique identifiers for tracking tool execution progress

Enables real-time user feedback during MCP tool execution while maintaining backward compatibility with existing non-streaming MCP servers.