Streaming results options

[schwifty-brows]

Are there any examples of returning a stream (chunked) like sse does? i currently have an SSE setup and i don't see where i have any option to return results as i get them, but i have to do it in one giant return, which is causing issues.

Any help is super appreciated, thanks!

[schwifty-brows]

an example is streaming results from a db back to the caller.

as query results are coming in -> sse send chunk

[schwifty-brows]

i tried

adding to the tool

ctx: RequestContext

then

let progress_notificationParam = ProgressNotificationParam {
progress_token: ProgressToken(NumberOrString::Number(1)),
progress: 0,
total: None,
message: Some("Querying starting".to_string()),
};

ctx.peer
            .notify_progress(progress_notificationParam)
            .await
            .unwrap();

// DB call

i dont see the "querying" message in postman events received for some reason

[jokemanfire]

Sorry, I may misunderstand your question.
if you means create a tool in mcp server, but you want this tool return a stream but not a JsonResult derictly?

[schwifty-brows]

Hello!

So what I want to do is as rows are being received I want to stream them back to via progress updates, instead of in batch (return the entire response) like I have to do now

[jokemanfire]

Let me confirm again. Do you want the MCP service to retrieve the query row results when calling the tool call on the client side (which is querying the database rows on your end), instead of returning the entire result as it is now, but returning each row result of the database in a streaming manner?

[browsified]

Yes, returning each row in a streaming manner.

Call tool, inside the method call is made to query the db or whatever and get stream back. While the stream is read produce the json/string back to the client via progress update or if there is a better way that’s cool as well

[jokemanfire]

I have similar idea about mcp spec modelcontextprotocol/modelcontextprotocol#510.
Perhaps it is difficult to solve from the perspective of the protocol itself now,
In fact, we can use another method to solve this problem. You can create an HTTP service in the MCP service. When you want to query a row in the database and return it in a streaming format, you can directly return a URL. The client can then establish this connection in a streaming format (which can be a WebSocket), which should temporarily meet this requirement

[browsified]

Do you have an example of how you do this?

[jokemanfire]

I will add async exec in mcp-server ,it also need stream, But I cannot guarantee that I will have time to do it soon. And I think you can first refer to the implementation of containerd exec async. containerd

[schwifty-brows]

so this kinda works but something is going on with the buffer

            Ok(mut data) => {
                loop {
                    let record = data.next().await;
                    if record.is_none() {
                        tokio::time::sleep(Duration::from_millis(1000)).await;
                        break Ok(CallToolResult::success(vec![Content::text("done")]));
                    }
                    let record_to_send = record.unwrap().unwrap();
                    debug!("record: {}", record_to_send);
                    let progress_param = ProgressNotificationParam {
                        progress_token: ProgressToken(NumberOrString::Number(counter)),
                        progress: counter,
                        total: None,
                        message: Some(record_to_send.clone()),
                    };
                    match ctx.peer
                        .notify_progress(progress_param.clone())
                        .await {
                        Ok(_) => {
                            counter += 1;
                            debug!("sent progress: {record_to_send}")
                        }
                        Err(e) => {
                            return Err(rmcp::Error::internal_error(format!("failed to notify progress: {}", e), Some(e.to_string().parse().unwrap())))
                        }
                    }
                }

i get various amount of the records back, not consistent at all

[schwifty-brows]

@jokemanfire confirmed, there seems to be an issue with the notify_progress. if i have 50 items lets say, sometimes i get x or nothing at all. is this a known issue?

[jokemanfire]

@jokemanfire confirmed, there seems to be an issue with the notify_progress. if i have 50 items lets say, sometimes i get x or nothing at all. is this a known issue?

I will confirm that, looks like there's no example , I wil ladd it first.

[schwifty-brows]

try this as an example

sample from a reqwest i have getting data

    let byte_stream = response.bytes_stream();
    let stream_reader = StreamReader::new(byte_stream.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
    let lines = BufReader::new(stream_reader).lines();
    Ok(LinesStream::new(lines))

then this is called with the return from above

        Ok(mut data) => {
            loop {
                let record = data.next().await;
                if record.is_none() {
                    tokio::time::sleep(Duration::from_millis(1000)).await;
                    break Ok(CallToolResult::success(vec![Content::text("done")]));
                }
                let record_to_send = record.unwrap().unwrap();
                debug!("record: {}", record_to_send);
                let progress_param = ProgressNotificationParam {
                    progress_token: ProgressToken(NumberOrString::Number(counter)),
                    progress: counter,
                    total: None,
                    message: Some(record_to_send.clone()),
                };
                match ctx.peer
                    .notify_progress(progress_param.clone())
                    .await {
                    Ok(_) => {
                        counter += 1;
                        debug!("sent progress: {record_to_send}")
                    }
                    Err(e) => {
                        return Err(rmcp::Error::internal_error(format!("failed to notify progress: {}", e), Some(e.to_string().parse().unwrap())))
                    }
                }
            }

where data is of type impl Stream<Item=Result<String, Error>>

[schwifty-brows]

this

tokio::time::sleep(Duration::from_millis(1000)).await;

was added to test if there was some delay by chance i was missing, didn't make much of a difference

[jokemanfire]

#271 @schwifty-brows please check my example ,