|
1 | 1 | use rmcp::{ |
2 | 2 | ClientHandler, ErrorData, RoleClient, ServiceExt, |
3 | 3 | model::*, |
4 | | - service::RequestContext, |
| 4 | + service::{RequestContext, serve_directly}, |
5 | 5 | transport::{ |
6 | 6 | AuthClient, AuthorizationManager, StreamableHttpClientTransport, |
7 | 7 | auth::{AuthorizationCallback, OAuthState}, |
@@ -824,6 +824,97 @@ async fn run_elicitation_defaults_client(server_url: &str) -> anyhow::Result<()> |
824 | 824 | Ok(()) |
825 | 825 | } |
826 | 826 |
|
| 827 | +/// A minimal stateless client transport: every outgoing message is one HTTP |
| 828 | +/// POST and the JSON response body (if any) is queued for `receive()`. |
| 829 | +/// |
| 830 | +/// The SEP-2322 client scenario's mock server speaks the stateless lifecycle |
| 831 | +/// (no `initialize` handshake, plain JSON responses), which the session-based |
| 832 | +/// `StreamableHttpClientTransport` cannot do. The transport is harness |
| 833 | +/// plumbing; the behavior under test — the SDK's MRTR retry driver — runs |
| 834 | +/// unchanged on top of it. |
| 835 | +struct StatelessHttpTransport { |
| 836 | + http: reqwest::Client, |
| 837 | + uri: std::sync::Arc<str>, |
| 838 | + tx: tokio::sync::mpsc::Sender<ServerJsonRpcMessage>, |
| 839 | + rx: tokio::sync::mpsc::Receiver<ServerJsonRpcMessage>, |
| 840 | +} |
| 841 | + |
| 842 | +impl StatelessHttpTransport { |
| 843 | + fn new(uri: &str) -> Self { |
| 844 | + let (tx, rx) = tokio::sync::mpsc::channel(16); |
| 845 | + Self { |
| 846 | + http: reqwest::Client::new(), |
| 847 | + uri: uri.into(), |
| 848 | + tx, |
| 849 | + rx, |
| 850 | + } |
| 851 | + } |
| 852 | +} |
| 853 | + |
| 854 | +impl rmcp::transport::Transport<RoleClient> for StatelessHttpTransport { |
| 855 | + type Error = std::io::Error; |
| 856 | + |
| 857 | + fn send( |
| 858 | + &mut self, |
| 859 | + item: rmcp::model::ClientJsonRpcMessage, |
| 860 | + ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send + 'static { |
| 861 | + let http = self.http.clone(); |
| 862 | + let uri = self.uri.clone(); |
| 863 | + let tx = self.tx.clone(); |
| 864 | + async move { |
| 865 | + let response = http |
| 866 | + .post(uri.as_ref()) |
| 867 | + .header("MCP-Protocol-Version", "2026-07-28") |
| 868 | + .json(&item) |
| 869 | + .send() |
| 870 | + .await |
| 871 | + .map_err(std::io::Error::other)?; |
| 872 | + match response.json::<ServerJsonRpcMessage>().await { |
| 873 | + Ok(message) => { |
| 874 | + let _ = tx.send(message).await; |
| 875 | + } |
| 876 | + Err(_) => { |
| 877 | + // No JSON-RPC body (e.g. 202/204 for notifications). |
| 878 | + } |
| 879 | + } |
| 880 | + Ok(()) |
| 881 | + } |
| 882 | + } |
| 883 | + |
| 884 | + async fn receive(&mut self) -> Option<ServerJsonRpcMessage> { |
| 885 | + self.rx.recv().await |
| 886 | + } |
| 887 | + |
| 888 | + async fn close(&mut self) -> Result<(), Self::Error> { |
| 889 | + Ok(()) |
| 890 | + } |
| 891 | +} |
| 892 | + |
| 893 | +/// A stateless-lifecycle client: the scenario's server has no `initialize` |
| 894 | +/// handler, so skip the handshake with `serve_directly`, list the tools, and |
| 895 | +/// call each one via the high-level `call_tool` helper (which drives SEP-2322 |
| 896 | +/// `input_required` retry rounds when the server requests them). Used by the |
| 897 | +/// `sep-2322-client-request-state` scenario, whose mock server verifies |
| 898 | +/// requestState echo, fresh JSON-RPC ids on retry, state omission, isolation |
| 899 | +/// between tools, and the `resultType` default. |
| 900 | +async fn run_stateless_client(server_url: &str) -> anyhow::Result<()> { |
| 901 | + let transport = StatelessHttpTransport::new(server_url); |
| 902 | + let peer_info = InitializeResult::new(ServerCapabilities::builder().enable_tools().build()) |
| 903 | + .with_protocol_version(ProtocolVersion::V_2026_07_28); |
| 904 | + let client = serve_directly(FullClientHandler, transport, Some(peer_info)); |
| 905 | + |
| 906 | + let tools = client.list_tools(Default::default()).await?; |
| 907 | + tracing::debug!("Listed {} tools", tools.tools.len()); |
| 908 | + for tool in &tools.tools { |
| 909 | + let result = client |
| 910 | + .call_tool(CallToolRequestParams::new(tool.name.clone())) |
| 911 | + .await; |
| 912 | + tracing::debug!("Called {}: {:?}", tool.name, result.is_ok()); |
| 913 | + } |
| 914 | + client.cancel().await?; |
| 915 | + Ok(()) |
| 916 | +} |
| 917 | + |
827 | 918 | async fn run_sse_retry_client(server_url: &str) -> anyhow::Result<()> { |
828 | 919 | let transport = StreamableHttpClientTransport::from_uri(server_url); |
829 | 920 | let client = BasicClientHandler.serve(transport).await?; |
@@ -869,6 +960,7 @@ async fn main() -> anyhow::Result<()> { |
869 | 960 | run_elicitation_defaults_client(&server_url).await? |
870 | 961 | } |
871 | 962 | "sse-retry" => run_sse_retry_client(&server_url).await?, |
| 963 | + "sep-2322-client-request-state" => run_stateless_client(&server_url).await?, |
872 | 964 |
|
873 | 965 | // Auth scenarios - standard OAuth flow |
874 | 966 | "auth/metadata-default" |
|
0 commit comments