mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-26 16:02:36 +00:00
fix: run rmcp's reqwest client without a log subscriber
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
d7428f7568
commit
31f230af21
@@ -1,4 +1,4 @@
|
||||
use std::{collections::HashMap, sync::Arc};
|
||||
use std::{collections::HashMap, future::Future, sync::Arc};
|
||||
|
||||
use futures::stream::BoxStream;
|
||||
use reqwest::header::{HeaderName, HeaderValue};
|
||||
@@ -9,19 +9,21 @@ use rmcp::{
|
||||
},
|
||||
};
|
||||
use sse_stream::Sse;
|
||||
use tracing::{instrument::WithSubscriber, subscriber::NoSubscriber};
|
||||
|
||||
type Error = StreamableHttpError<reqwest::Error>;
|
||||
|
||||
/// reqwest names the URL it failed on in its errors, and an MCP server URL can
|
||||
/// carry the server's key. rmcp logs those errors itself as well as returning
|
||||
/// them, so the URL is dropped here, before either can happen.
|
||||
/// carry the server's key. rmcp logs those errors as well as returning them, so
|
||||
/// the URL is dropped before the transport sees one, and rmcp's reqwest client,
|
||||
/// which logs the errors it swallows, runs without a subscriber.
|
||||
#[derive(Clone)]
|
||||
pub(super) struct UrlRedactingClient(pub(super) reqwest::Client);
|
||||
|
||||
fn without_url(e: Error) -> Error {
|
||||
match e {
|
||||
StreamableHttpError::Client(e) => StreamableHttpError::Client(e.without_url()),
|
||||
e => e,
|
||||
async fn redacted<T>(request: impl Future<Output = Result<T, Error>>) -> Result<T, Error> {
|
||||
match request.with_subscriber(NoSubscriber::default()).await {
|
||||
Err(StreamableHttpError::Client(e)) => Err(StreamableHttpError::Client(e.without_url())),
|
||||
r => r,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,10 +38,11 @@ impl StreamableHttpClient for UrlRedactingClient {
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> Result<StreamableHttpPostResponse, Error> {
|
||||
self.0
|
||||
.post_message(uri, message, session_id, auth_header, custom_headers)
|
||||
.await
|
||||
.map_err(without_url)
|
||||
redacted(
|
||||
self.0
|
||||
.post_message(uri, message, session_id, auth_header, custom_headers),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn post_message_with_max_sse_event_size(
|
||||
@@ -51,17 +54,15 @@ impl StreamableHttpClient for UrlRedactingClient {
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
max_sse_event_size: usize,
|
||||
) -> Result<StreamableHttpPostResponse, Error> {
|
||||
self.0
|
||||
.post_message_with_max_sse_event_size(
|
||||
uri,
|
||||
message,
|
||||
session_id,
|
||||
auth_header,
|
||||
custom_headers,
|
||||
max_sse_event_size,
|
||||
)
|
||||
.await
|
||||
.map_err(without_url)
|
||||
redacted(self.0.post_message_with_max_sse_event_size(
|
||||
uri,
|
||||
message,
|
||||
session_id,
|
||||
auth_header,
|
||||
custom_headers,
|
||||
max_sse_event_size,
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
async fn delete_session(
|
||||
@@ -71,10 +72,11 @@ impl StreamableHttpClient for UrlRedactingClient {
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> Result<(), Error> {
|
||||
self.0
|
||||
.delete_session(uri, session_id, auth_header, custom_headers)
|
||||
.await
|
||||
.map_err(without_url)
|
||||
redacted(
|
||||
self.0
|
||||
.delete_session(uri, session_id, auth_header, custom_headers),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_stream(
|
||||
@@ -85,10 +87,11 @@ impl StreamableHttpClient for UrlRedactingClient {
|
||||
auth_header: Option<String>,
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
) -> Result<BoxStream<'static, Result<Sse, SseError>>, Error> {
|
||||
self.0
|
||||
.get_stream(uri, session_id, last_event_id, auth_header, custom_headers)
|
||||
.await
|
||||
.map_err(without_url)
|
||||
redacted(
|
||||
self.0
|
||||
.get_stream(uri, session_id, last_event_id, auth_header, custom_headers),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_stream_with_max_sse_event_size(
|
||||
@@ -100,16 +103,14 @@ impl StreamableHttpClient for UrlRedactingClient {
|
||||
custom_headers: HashMap<HeaderName, HeaderValue>,
|
||||
max_sse_event_size: usize,
|
||||
) -> Result<BoxStream<'static, Result<Sse, SseError>>, Error> {
|
||||
self.0
|
||||
.get_stream_with_max_sse_event_size(
|
||||
uri,
|
||||
session_id,
|
||||
last_event_id,
|
||||
auth_header,
|
||||
custom_headers,
|
||||
max_sse_event_size,
|
||||
)
|
||||
.await
|
||||
.map_err(without_url)
|
||||
redacted(self.0.get_stream_with_max_sse_event_size(
|
||||
uri,
|
||||
session_id,
|
||||
last_event_id,
|
||||
auth_header,
|
||||
custom_headers,
|
||||
max_sse_event_size,
|
||||
))
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user