How to stream reponse back from MCP server to Client? (MCP Tool Streaming) #1124

Activity

1. ashish-2412

   added

   questionFurther information is requestedFurther information is requested

   on Jul 11, 2025
2. ashish-2412

   changed the title [-]How to stream reponse back from MCP server to Client?[/-] [+]How to stream reponse back from MCP server to Client? (MCP Tool Streaming)[/+] on Jul 11, 2025
3. 1810912213qqcom commented on Jul 25, 2025

   

   1810912213qqcom

   on Jul 25, 2025

   More actions

   Hi friend, I recently implemented this feature using a somewhat hacky approach while making as few changes as possible to the original SDK. Later, I tried integrating the feature as an extension of client.call_tool directly into the client class. I’ll now present both approaches.
   Let's take a look at the results now. I test the function with postman.

   This is the result response in chunks.
   
   And here is the mcp server tool code.

   

   Of cource you can only yield one chunk , which is the tool finally return.

   

   What I did was provide a custom implementation of the ProgressHandler used inside client.call_tool. I had first tried to yield chunks directly from the handler, but that’s impossible when call_tool itself is awaited. So I extended call_tool so that the custom progress states can also be yielded out.

   I didn’t change the original @server.tool at all, so you can still use it exactly as before. Whenever you want to return additional states, just send your custom protocol—say, a JSON chunk—via ctx.report_progress().

   Now let’s look at the code, starting with the interface layer.

   @app.post("/v1/mcp/tinker")
   async def call_mcp_tools(request: Request):
       body = await request.json()
       return StreamingResponse(
           custom_mcp(url=body["url"], funname=body["name"], args=body["args"]),
           media_type="text/event-stream",
           headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
       )
   
   
   async def custom_mcp(url: str, funname: str, args: dict):
   
       try:
           try:
               async with Client(url) as client:
   
                   async for chunk in client.call_tool_yield_chunks(
                       name=funname, arguments=args
                   ):
                       yield f"data: {json.dumps({"chunk":chunk},ensure_ascii=False)}\n\n"
   
           except Exception as e:
               print("task failed", e)
   
       except Exception as e:
           logger.error(f"Error: {e}")
           async for sendchunk in stream_completions_create(
               content=str(e), type="mcp_tool_error"
           ):
               yield f"data: {json.dumps(sendchunk,ensure_ascii=False)}\n\n"
   

   Next, we created an enhanced version of call_tool as a custom method inside the Client class.

           async def fetch_mcp_progress():
               try:
                    
                   result = await self.call_tool_mcp(
                       name=name,
                       arguments=arguments or {},
                       timeout=timeout,
                       progress_handler=state_update,
                   )
                   data = None
                   if result.isError and raise_on_error:
                       msg = cast(mcp.types.TextContent, result.content[0]).text
                       raise ToolError(msg)
                   elif result.structuredContent:
                       try:
                           if name not in self.session._tool_output_schemas:
                               await self.session.list_tools()
                           if name in self.session._tool_output_schemas:
                               output_schema = self.session._tool_output_schemas.get(name)
                               if output_schema:
                                   if output_schema.get("x-fastmcp-wrap-result"):
                                       output_schema = output_schema.get("properties", {}).get(
                                           "result"
                                       )
                                       structured_content = result.structuredContent.get("result")
                                   else:
                                       structured_content = result.structuredContent
                                   output_type = json_schema_to_type(output_schema)
                                   type_adapter = get_cached_typeadapter(output_type)
                                   data = type_adapter.validate_python(structured_content)
                               else:
                                   data = result.structuredContent
                       except Exception as e:
                           logger.error(f"Error parsing structured content: {e}")
   
                   tempresult= CallToolResult(
                       content=result.content,
                       structured_content=result.structuredContent,
                       data=data,
                       is_error=result.isError,
                   )
                   logger.info(f"result:{tempresult} {type(tempresult)}")
                   await mcp_progress_queue.put(tempresult)
                   logger.info(f"run success, result:{tempresult}")
                   await mcp_progress_queue.put(None)
               except Exception as e:
                   logger.error(f"task failed: {str(e)}")
           mcp_task = asyncio.create_task(fetch_mcp_progress())
           mcp_done = False
           while not mcp_done:
               try:
                   mcp_data = await mcp_progress_queue.get()
                   if mcp_data is None:
                       mcp_done = True
                       break
                   else:
                       if type(mcp_data) == str:
                           yield mcp_data
                       elif type(mcp_data) == CallToolResult:
   
                           data = {
                               "content": vars(mcp_data.content[0]),
                               "structured_content": mcp_data.structured_content,
                               "data": mcp_data.data,
                               "is_error": mcp_data.is_error
                           }
                           yield data
                       await asyncio.sleep(0.0001)
   
               except asyncio.QueueEmpty:
                   await asyncio.sleep(0.01)  
           await asyncio.gather(mcp_task)
   

   If you prefer not to touch the Client class, you can also wrap client.call_tool externally, like this:

   @app.post("/v1/mcp/callToolstest/{server_id}")
   async def call_mcp_tools(server_id: str, request: Request):
       async def mcp_tools_data(server_id: str, body: dict):
           mcp_progress_queue = asyncio.Queue()
   
           # 
           async def state_update(
               progress: float, total: float = None, message: str = None
           ):
               try:
                   if message:  # 错信号
                       logger.success(f"catch {message}")
                       await mcp_progress_queue.put(message)
   
               except ValueError as e:
                   logger.error(f"no message {str(e)}")
   
           # mcp-process
           async def fetch_mcp_progress():
               db = DB(type="pb")
   
               try:
                   mcpInfo = db.retrieve_mcp_config(server_id)
               except Exception as e:
                   logger.error(f"Error: serverid is error {e}")
               try:
                   if mcpInfo["args"]:
   
                       if type(mcpInfo["args"]) == list:
                           url = mcpInfo["args"][0]
                       elif type(mcpInfo["args"]) == str:
                           url = mcpInfo["args"]
                       else:
                           logger.error(
                               f"mcpInfo['args'] is not a list or a string.---{mcpInfo['args']}"
                           )
                   else:
                       url = mcpInfo["baseUrl"]
               except Exception as e:
                   logger.error(f"Error: extract url error {e}")
               try:
                   async with Client(url) as client:
   
                       result = await client.call_tool(
                           name=body["name"],
                           arguments=body["args"],
                           progress_handler=state_update,
                       )
                       await mcp_progress_queue.put(result)
                       # await mcp_progress_queue.put(["DONE"])
                       await mcp_progress_queue.put(None)
               except Exception as e:
                   print("task failed", e)
   
           mcp_task = asyncio.create_task(fetch_mcp_progress())
           mcp_done = False
           while not mcp_done:
               try:
                   mcp_data = await mcp_progress_queue.get()
                   if mcp_data is None:
                       mcp_done = True
                       break
                   elif mcp_data == ["DONE"]:
                       logger.error(f"mcp done?:{mcp_data}")
                       yield f"data [DONE]\n\n"
                   else:
                       logger.error(f"yield mcp_data:{mcp_data} {type(mcp_data)}")
                       if type(mcp_data) == str:
                           async for send_chunk in stream_completions_create(
                               content=mcp_data,
                               type="mcp_tool_in_progress",
                               message_id=str(uuid.uuid4()),
                           ):
                               yield f"data: {json.dumps(send_chunk,ensure_ascii=False)} \n\n"
                       else:
                           data = {
                               "content": [
                                   {
                                       "type": mcp_data.content[0].type,
                                       "text": mcp_data.content[0].text,
                                   }
                               ]
                           }
                           async for send_chunk in stream_completions_create(
                               content=json.dumps(data["responses"], ensure_ascii=False),
                               type="mcp_tool_complete",
                               message_id=str(uuid.uuid4()),
                           ):
                               yield f"data: {json.dumps(send_chunk,ensure_ascii=False)} \n\n"
                   
                       yield f""
                       await asyncio.sleep(0.0001)  #
   
               except asyncio.QueueEmpty:
                   await asyncio.sleep(0.01)  # 
           await asyncio.gather(mcp_task)
           yield "data: [DONE]\n\n"
   
       logger.info(f"Request server_id: {server_id}")
       body = await request.json()
       try:
           return StreamingResponse(
               mcp_tools_data(server_id=server_id, body=body),
               media_type="text/event-stream",
               headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
           )
   
       except Exception as e:
           logger.error(e)
   
           return HTTPException(status_code=500, detail=str(e))
   

   There are some miscellaneous bits in here—like database reads and logging—just ignore those; the overall idea is the same.Hope it helps!

   👍React with 👍1Reacted by Felix Weinberger
4. felixweinberger

   closed this as a duplicateof can we call tools and return steaming output? #472 on Oct 7, 2025