Claude finished @MervinPraison's task —— View job

Task List

Search codebase for MCP implementation
Analyze current SSE implementation
Research HTTP-Streaming requirements
Design backward-compatible solution
Provide implementation recommendations

Analysis Results

Current MCP Implementation

The codebase currently supports two transport methods for MCP:

SSE (Server-Sent Events) - For HTTP-based connections
Stdio - For process-based connections

Key Files

src/praisonai-agents/praisonaiagents/mcp/mcp_sse.py - Python SSE implementation
src/praisonai-agents/praisonaiagents/mcp/mcp.py - Main MCP module
src/praisonai-ts/src/tools/mcpSse.ts - TypeScript SSE implementation

Proposed Solution

1. Add HTTP-Streaming Support with Backward Compatibility

Create a new transport module mcp_http_stream.py alongside the existing mcp_sse.py:

# src/praisonai-agents/praisonaiagents/mcp/mcp_http_stream.py
"""
HTTP-Streaming client implementation for MCP (Model Context Protocol).
Provides HTTP chunked streaming transport as an alternative to SSE.
"""

import asyncio
import json
import logging
from typing import List, Dict, Any, Optional, AsyncIterator
import httpx
from mcp import ClientSession
from mcp.client.session import BaseSession

class HTTPStreamTransport:
    """HTTP chunked streaming transport for MCP."""
    
    def __init__(self, url: str, headers: Optional[Dict] = None):
        self.url = url
        self.headers = headers or {}
        self.client = httpx.AsyncClient()
        
    async def read_stream(self) -> AsyncIterator[bytes]:
        """Read from HTTP stream using chunked transfer encoding."""
        async with self.client.stream('GET', self.url, headers=self.headers) as response:
            async for chunk in response.aiter_bytes():
                yield chunk
                
    async def write_stream(self, data: bytes):
        """Write to HTTP stream."""
        await self.client.post(self.url, content=data, headers=self.headers)

class HTTPStreamMCPClient:
    """HTTP-Streaming MCP client with same interface as SSEMCPClient."""
    
    def __init__(self, server_url: str, debug: bool = False, timeout: int = 60):
        self.server_url = server_url
        self.debug = debug
        self.timeout = timeout
        # Similar initialization as SSEMCPClient...

2. Modify Main MCP Module

Update mcp.py to support both SSE and HTTP-Streaming:

# In src/praisonai-agents/praisonaiagents/mcp/mcp.py

def __init__(self, command_or_string=None, args=None, *, command=None, 
             timeout=60, debug=False, transport="auto", **kwargs):
    """
    Args:
        transport: Transport type - "auto", "sse", "http-stream", or "stdio"
                  "auto" will detect based on URL format
    """
    # ... existing code ...
    
    # Check if this is an HTTP URL
    if isinstance(command_or_string, str) and re.match(r'^https?://', command_or_string):
        # Determine transport type
        if transport == "auto":
            # Default to SSE for /sse endpoints, HTTP-stream otherwise
            if command_or_string.endswith('/sse'):
                transport = "sse"
            else:
                transport = "http-stream"
                
        if transport == "sse":
            from .mcp_sse import SSEMCPClient
            self.client = SSEMCPClient(command_or_string, debug=debug, timeout=timeout)
        elif transport == "http-stream":
            from .mcp_http_stream import HTTPStreamMCPClient
            self.client = HTTPStreamMCPClient(command_or_string, debug=debug, timeout=timeout)
        else:
            raise ValueError(f"Unknown transport: {transport}")
            
        self._tools = list(self.client.tools)
        self.is_http = True
        return

3. TypeScript Implementation

Add HTTP-Streaming support to TypeScript:

// src/praisonai-ts/src/tools/mcpHttpStream.ts
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { MCPTool } from './mcpSse';

export class HTTPStreamTransport {
  constructor(private url: URL) {}
  
  // Implement HTTP chunked streaming transport
  // following MCP SDK transport interface
}

export class MCPHttpStream {
  tools: MCPTool[] = [];
  private client: Client | null = null;
  
  constructor(private url: string, private debug = false) {}
  
  async initialize(): Promise<void> {
    this.client = new Client({ name: 'praisonai-ts-mcp', version: '1.0.0' });
    const transport = new HTTPStreamTransport(new URL(this.url));
    await this.client.connect(transport);
    // ... similar to SSE implementation
  }
}

4. Usage Examples

# Automatic detection (backward compatible)
agent = Agent(
    tools=MCP("http://localhost:8080/sse")  # Uses SSE
)

agent = Agent(
    tools=MCP("http://localhost:8080/stream")  # Uses HTTP-stream
)

# Explicit transport selection
agent = Agent(
    tools=MCP("http://localhost:8080/api", transport="http-stream")
)

# Legacy SSE endpoint with new transport
agent = Agent(
    tools=MCP("http://localhost:8080/sse", transport="http-stream")
)

Benefits of This Approach

Zero Breaking Changes - Existing code continues to work
Automatic Detection - Smart defaults based on URL patterns
Explicit Control - Users can force a specific transport
Minimal Code Changes - New transport is isolated in its own module
Consistent Interface - Both transports expose the same API

Implementation Priority

First implement Python HTTP-Streaming support
Test with existing MCP servers
Add TypeScript support
Update documentation and examples

Summary

I've designed a solution that adds HTTP-Streaming support to PraisonAI's MCP implementation while maintaining full backward compatibility. The solution introduces a new transport parameter that defaults to "auto" for intelligent transport selection based on URL patterns. All existing code will continue to work without modifications.

Would you like me to proceed with implementing this solution?

Uh oh!

Support MCP with HTTP-Streaming #722

Description

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions