1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
//! D-14 live-HTTP `_meta`-at-top-level acceptance gate (Phase 104, TOUT-04).
//!
//! This is the phase's wire-shape regression lock: a REAL HTTP round-trip (no
//! in-process shim) that drives a `tools/call` against a high-level, store-less
//! `pmcp::Server` whose tool returns
//! [`ToolOutput::Result`](pmcp::ToolOutput::Result) carrying a `CallToolResult` — the
//! SEP-1686 task-augmented result path (Plan 02/04). It consumes the REAL
//! dispatch output over `StreamableHttpServer` + `StreamableHttpTransport` and
//! asserts on the RAW wire JSON that:
//!
//! 1. `result._meta` is present at the result TOP LEVEL (the `_meta[related-task]`
//! envelope survives dispatch — a `_meta`-sniffing client detects the task); AND
//! 2. `result.content[0].text`, if present, does NOT contain a stringified `_meta`
//! (the agent-lake double-wrap bug — the whole `CallToolResult` serialized into
//! `content[0].text` — must NEVER reappear).
//!
//! We read the raw JSON via the transport's `TransportMessage::Response` payload,
//! which carries the `tools/call` result as an untyped `serde_json::Value` — NOT
//! deserialized into `CallToolResult` — so the wire shape is observed faithfully
//! and never from a hand-authored fixture (the note's ask #4).
//!
//! Test reliability (carried from the Phase 102 HTTP harness):
//! - EPHEMERAL PORT — binds `127.0.0.1:0`, reads the bound address back from
//! `StreamableHttpServer::start()` (no hardcoded port).
//! - READINESS — `start()` binds the listener before returning (no fixed sleep).
//! - SHUTDOWN — the spawned server `JoinHandle` is `abort()`ed after the round-trip.
#![cfg(all(feature = "streamable-http", not(target_arch = "wasm32")))]
use std::net::SocketAddr;
use std::sync::Arc;
use async_trait::async_trait;
use pmcp::server::streamable_http_server::StreamableHttpServer;
use pmcp::shared::streamable_http::{StreamableHttpTransport, StreamableHttpTransportConfig};
use pmcp::shared::{Transport, TransportMessage};
use pmcp::types::jsonrpc::ResponsePayload;
use pmcp::types::tasks::TaskMetadata;
use pmcp::types::{
CallToolRequest, CallToolResult, ClientCapabilities, ClientRequest, Content, Implementation,
InitializeRequest, Request,
};
use pmcp::{RequestHandlerExtra, Server, ToolHandler, ToolOutput};
use tokio::sync::Mutex;
use url::Url;
/// The task id the tool stamps into the related-task `_meta` envelope.
const RELATED_TASK_ID: &str = "t-http";
/// A tool whose `handle_output` returns [`ToolOutput::Result`] — a fully-formed
/// `CallToolResult` carrying a top-level `_meta[related-task]` envelope that must
/// reach the wire VERBATIM.
struct AugmentedResultTool;
impl AugmentedResultTool {
/// The verbatim envelope: `content = [text("done")]` + related-task `_meta`.
fn envelope() -> CallToolResult {
CallToolResult::new(vec![Content::text("done")])
.with_related_task(TaskMetadata::new(RELATED_TASK_ID))
}
}
#[async_trait]
impl ToolHandler for AugmentedResultTool {
async fn handle(
&self,
_args: serde_json::Value,
_extra: RequestHandlerExtra,
) -> pmcp::Result<serde_json::Value> {
// Serialize fallback for non-dispatch callers; the dispatch path uses
// `handle_output` below.
Ok(serde_json::to_value(Self::envelope())?)
}
async fn handle_output(
&self,
_args: serde_json::Value,
_extra: RequestHandlerExtra,
) -> pmcp::Result<ToolOutput> {
// Verbatim: the handler owns the full envelope (Plan 02/04, D-04a).
Ok(ToolOutput::Result(Self::envelope()))
}
}
/// Build a high-level, store-LESS `Server` exposing the augmented-result tool.
///
/// No `task_store` here: this is a plain synchronous tool result that merely
/// POINTS at a related task via `_meta` — exactly the SEP-1686 shape whose wire
/// survival this gate locks.
fn build_server() -> pmcp::Result<Server> {
Server::builder()
.name("tool-output-result-http")
.version("1.0.0")
.tool("augmented", AugmentedResultTool)
.build()
}
/// Stand the server up over REAL HTTP; return the read-back bound address + handle.
///
/// EPHEMERAL PORT + READINESS: bind `127.0.0.1:0` and read the assigned address
/// back from `start()`, which binds the listener before returning (no sleep).
async fn spawn_http_server() -> pmcp::Result<(SocketAddr, tokio::task::JoinHandle<()>)> {
let server = Arc::new(Mutex::new(build_server()?));
let bind_addr: SocketAddr = "127.0.0.1:0".parse().expect("valid loopback addr");
let http_server = StreamableHttpServer::new(bind_addr, server);
let (bound, handle) = http_server.start().await?;
Ok((bound, handle))
}
/// Build the pmcp HTTP transport pointed at `bound` (single-response JSON path).
fn http_transport(bound: SocketAddr) -> pmcp::Result<StreamableHttpTransport> {
let config = StreamableHttpTransportConfig {
url: Url::parse(&format!("http://{bound}"))
.map_err(|e| pmcp::Error::Internal(e.to_string()))?,
extra_headers: vec![],
auth_provider: None,
session_id: None,
enable_json_response: true,
on_resumption_token: None,
http_middleware_chain: None,
};
Ok(StreamableHttpTransport::new(config))
}
/// Extract the raw JSON-RPC `result` Value from a received response message.
fn expect_result_value(msg: TransportMessage) -> pmcp::Result<serde_json::Value> {
match msg {
TransportMessage::Response(resp) => match resp.payload {
ResponsePayload::Result(value) => Ok(value),
ResponsePayload::Error(err) => Err(pmcp::Error::internal(format!(
"expected a result over HTTP, got JSON-RPC error: {}",
err.message
))),
},
other => Err(pmcp::Error::internal(format!(
"expected a Response message, got: {other:?}"
))),
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn tool_output_result_carries_meta_at_top_level_over_http() -> pmcp::Result<()> {
let (bound, server_handle) = spawn_http_server().await?;
// Guard the round-trip so the server handle is ALWAYS aborted, even on failure.
let outcome = run_round_trip(bound).await;
// SHUTDOWN: abort the spawned server task (no lingering listener, no hang).
server_handle.abort();
match server_handle.await {
Ok(()) => {},
Err(e) if e.is_cancelled() => {},
Err(e) => panic!("HTTP server task ended unexpectedly: {e}"),
}
outcome
}
async fn run_round_trip(bound: SocketAddr) -> pmcp::Result<()> {
let mut transport = http_transport(bound)?;
// 1) initialize over HTTP (establishes the session + protocol version that the
// transport reuses on subsequent sends).
transport
.send(TransportMessage::Request {
id: 1i64.into(),
request: Request::Client(Box::new(ClientRequest::Initialize(InitializeRequest::new(
Implementation::new("tout-04-gate", "1.0.0"),
ClientCapabilities::default(),
)))),
})
.await?;
let _init = transport.receive().await?;
// 2) tools/call over the REAL transport -> consume the RAW dispatch output.
transport
.send(TransportMessage::Request {
id: 2i64.into(),
request: Request::Client(Box::new(ClientRequest::CallTool(CallToolRequest::new(
"augmented",
serde_json::json!({}),
)))),
})
.await?;
let result = expect_result_value(transport.receive().await?)?;
// ASSERTION 1 — `_meta` survives at the result TOP LEVEL.
let meta = result.get("_meta").ok_or_else(|| {
pmcp::Error::internal(format!(
"result._meta must be present at top level over HTTP; got: {result}"
))
})?;
assert!(
meta.get("io.modelcontextprotocol/related-task").is_some(),
"result._meta must carry the related-task envelope, got: {meta}"
);
assert_eq!(
meta["io.modelcontextprotocol/related-task"]["taskId"].as_str(),
Some(RELATED_TASK_ID),
"related-task taskId must equal the tool's minted id over HTTP"
);
// ASSERTION 2 — content[0].text (if any) must NOT be a stringified envelope.
if let Some(text) = result
.get("content")
.and_then(|c| c.get(0))
.and_then(|c0| c0.get("text"))
.and_then(serde_json::Value::as_str)
{
assert_eq!(
text, "done",
"content[0].text must be the real text, not a wrapped envelope"
);
assert!(
!text.contains("_meta"),
"content[0].text must NOT contain a stringified `_meta` (the double-wrap bug), got: {text}"
);
}
Ok(())
}