1+ """
2+ HTTP-Streaming client implementation for MCP (Model Context Protocol).
3+ Provides HTTP chunked streaming transport as an alternative to SSE.
4+ """
5+
6+ import asyncio
7+ import logging
8+ import threading
9+ import queue
10+ import json
11+ from typing import Any , Dict , List , Optional
12+ from mcp import ClientSession
13+ from mcp .client .session import Transport
14+ from mcp .shared .memory import get_session_from_context
15+
16+ logger = logging .getLogger (__name__ )
17+
18+
19+ class HTTPStreamingTransport (Transport ):
20+ """HTTP chunked streaming transport for MCP."""
21+
22+ def __init__ (self , url : str , headers : Optional [Dict [str , str ]] = None ):
23+ self .url = url
24+ self .headers = headers or {}
25+ self ._closed = False
26+
27+ async def start (self ) -> None :
28+ """Initialize the transport."""
29+ # TODO: Implement actual HTTP streaming connection
30+ # For now, this is a placeholder that follows the Transport interface
31+ pass
32+
33+ async def close (self ) -> None :
34+ """Close the transport."""
35+ self ._closed = True
36+
37+ async def send (self , message : Dict [str , Any ]) -> None :
38+ """Send a message through the transport."""
39+ if self ._closed :
40+ raise RuntimeError ("Transport is closed" )
41+ # TODO: Implement actual HTTP streaming send
42+ # This would send the message as a chunked HTTP request
43+
44+ async def receive (self ) -> Dict [str , Any ]:
45+ """Receive a message from the transport."""
46+ if self ._closed :
47+ raise RuntimeError ("Transport is closed" )
48+ # TODO: Implement actual HTTP streaming receive
49+ # This would read from the chunked HTTP response stream
50+ raise NotImplementedError ("HTTP streaming receive not yet implemented" )
51+
52+
53+ class HTTPStreamingMCPTool :
54+ """Wrapper for MCP tools accessed via HTTP streaming."""
55+
56+ def __init__ (self , tool_def : Dict [str , Any ], call_func ):
57+ self .name = tool_def ["name" ]
58+ self .description = tool_def .get ("description" , "" )
59+ self .inputSchema = tool_def .get ("inputSchema" , {})
60+ self ._call_func = call_func
61+
62+ def __call__ (self , ** kwargs ):
63+ """Synchronous wrapper for calling the tool."""
64+ result_queue = queue .Queue ()
65+
66+ async def _async_call ():
67+ try :
68+ result = await self ._call_func (self .name , kwargs )
69+ result_queue .put (("success" , result ))
70+ except Exception as e :
71+ result_queue .put (("error" , e ))
72+
73+ # Run in event loop
74+ loop = asyncio .new_event_loop ()
75+ asyncio .set_event_loop (loop )
76+
77+ try :
78+ loop .run_until_complete (_async_call ())
79+ finally :
80+ loop .close ()
81+
82+ status , result = result_queue .get ()
83+ if status == "error" :
84+ raise result
85+ return result
86+
87+ async def _async_call (self , ** kwargs ):
88+ """Async version of tool call."""
89+ return await self ._call_func (self .name , kwargs )
90+
91+ def to_openai_tool (self ):
92+ """Convert to OpenAI tool format."""
93+ schema = self .inputSchema .copy ()
94+ self ._fix_array_schemas (schema )
95+
96+ return {
97+ "type" : "function" ,
98+ "function" : {
99+ "name" : self .name ,
100+ "description" : self .description ,
101+ "parameters" : schema
102+ }
103+ }
104+
105+ def _fix_array_schemas (self , schema ):
106+ """Fix array schemas for OpenAI compatibility."""
107+ if isinstance (schema , dict ):
108+ if schema .get ("type" ) == "array" and "items" not in schema :
109+ schema ["items" ] = {"type" : "string" }
110+ for value in schema .values ():
111+ if isinstance (value , dict ):
112+ self ._fix_array_schemas (value )
113+
114+
115+ class HTTPStreamingMCPClient :
116+ """HTTP-Streaming MCP client with same interface as SSEMCPClient."""
117+
118+ def __init__ (self , server_url : str , debug : bool = False , timeout : int = 60 ):
119+ self .server_url = server_url
120+ self .debug = debug
121+ self .timeout = timeout
122+ self .tools = []
123+ self ._client = None
124+ self ._session = None
125+ self ._transport = None
126+ self ._thread = None
127+ self ._loop = None
128+
129+ # Initialize in background thread
130+ self ._initialize ()
131+
132+ def _initialize (self ):
133+ """Initialize the HTTP streaming connection in a background thread."""
134+ init_done = threading .Event ()
135+
136+ def _thread_init ():
137+ self ._loop = asyncio .new_event_loop ()
138+ asyncio .set_event_loop (self ._loop )
139+
140+ async def _async_init ():
141+ try :
142+ # Create transport
143+ self ._transport = HTTPStreamingTransport (self .server_url )
144+
145+ # Create MCP client
146+ self ._client = ClientSession ()
147+
148+ # Initialize session with transport
149+ await self ._client .initialize (self ._transport )
150+
151+ # Store session in context
152+ self ._session = self ._client
153+
154+ # List available tools
155+ tools_result = await self ._client .call_tool ("list-tools" , {})
156+ if tools_result and hasattr (tools_result , 'tools' ):
157+ for tool_def in tools_result .tools :
158+ tool = HTTPStreamingMCPTool (
159+ tool_def .model_dump (),
160+ self ._call_tool_async
161+ )
162+ self .tools .append (tool )
163+
164+ if self .debug :
165+ logger .info (f"HTTP Streaming MCP client initialized with { len (self .tools )} )
166+
167+ except Exception as e :
168+ logger .error (f"Failed to initialize HTTP Streaming MCP client: { e } )
169+ raise
170+
171+ self ._loop .run_until_complete (_async_init ())
172+ init_done .set ()
173+
174+ # Keep the loop running
175+ self ._loop .run_forever ()
176+
177+ self ._thread = threading .Thread (target = _thread_init , daemon = True )
178+ self ._thread .start ()
179+
180+ # Wait for initialization
181+ init_done .wait (timeout = self .timeout )
182+
183+ async def _call_tool_async (self , tool_name : str , arguments : Dict [str , Any ]):
184+ """Call a tool asynchronously."""
185+ if not self ._session :
186+ raise RuntimeError ("HTTP Streaming MCP client not initialized" )
187+
188+ result = await self ._session .call_tool (tool_name , arguments )
189+
190+ # Extract content from result
191+ if hasattr (result , 'content' ):
192+ content = result .content
193+ if len (content ) == 1 and hasattr (content [0 ], 'text' ):
194+ return content [0 ].text
195+ return [c .text if hasattr (c , 'text' ) else str (c ) for c in content ]
196+ return result
197+
198+ def __iter__ (self ):
199+ """Make client iterable to return tools."""
200+ return iter (self .tools )
201+
202+ def to_openai_tools (self ):
203+ """Convert all tools to OpenAI format."""
204+ return [tool .to_openai_tool () for tool in self .tools ]
205+
206+ def shutdown (self ):
207+ """Shutdown the client."""
208+ if self ._loop and self ._thread :
209+ self ._loop .call_soon_threadsafe (self ._loop .stop )
210+ self ._thread .join (timeout = 5 )
211+
212+ if self ._transport and not self ._transport ._closed :
213+ async def _close ():
214+ await self ._transport .close ()
215+
216+ if self ._loop :
217+ asyncio .run_coroutine_threadsafe (_close (), self ._loop )
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?v=4&size=40){kind=avatar}
0 commit comments