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
use std::sync::Arc;
use asyncai::mcp::{McpToolExecutor, StdioServerConfig};
use asyncai::{CodeModeToolExecutor, ToolExecutor};
use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
const BODY: &str = "hello-mcp";
async fn loopback_server() -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let (mut socket, _) = listener.accept().await.unwrap();
tokio::spawn(async move {
let mut buffer = [0u8; 4096];
let _ = socket.read(&mut buffer).await.unwrap();
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
BODY.len(),
BODY,
);
socket.write_all(response.as_bytes()).await.unwrap();
socket.shutdown().await.ok();
});
}
});
format!("http://{address}/probe")
}
async fn simple_executor() -> McpToolExecutor {
McpToolExecutor::connect_stdio(vec![StdioServerConfig {
command: env!("CARGO_BIN_EXE_asyncai-mcp-simple").to_string(),
args: Vec::new(),
}])
.await
.unwrap()
}
#[tokio::test]
async fn stdio_mcp_server_lists_and_calls_http_get() {
let url = loopback_server().await;
let executor = simple_executor().await;
let tools = executor.list_tools().await.unwrap();
let http_get = tools
.iter()
.find(|tool| tool.name == "http_get")
.expect("http_get tool missing");
assert_eq!(http_get.input_schema["type"], "object");
assert_eq!(http_get.input_schema["properties"]["url"]["type"], "string");
let output_schema = http_get.output_schema.as_ref().expect("output schema");
assert_eq!(output_schema["properties"]["status"]["type"], "integer");
assert_eq!(http_get.read_only, Some(true));
assert!(!http_get.is_mutating());
let result = executor
.call_tool("http_get", json!({ "url": url }))
.await
.unwrap();
assert!(!result.is_error, "tool returned error: {}", result.content);
let parsed: Value = serde_json::from_str(&result.content).unwrap();
assert_eq!(parsed["status"], 200);
assert_eq!(parsed["content_type"], "text/plain");
assert_eq!(parsed["body"], BODY);
}
#[tokio::test]
async fn code_mode_drives_the_stdio_server_from_python() {
let url = loopback_server().await;
let code_mode = CodeModeToolExecutor::new(Arc::new(simple_executor().await))
.await
.unwrap();
let tools = code_mode.list_tools().await.unwrap();
assert_eq!(tools.len(), 1);
assert_eq!(tools[0].name, "run_code");
assert!(tools[0]
.description
.as_deref()
.unwrap()
.contains("async def http_get(*, url: str) -> dict[str, Any]:"));
let code = format!(
"import asyncio\n\
a, b = await asyncio.gather(http_get(url={url:?}), http_get(url={url:?}))\n\
{{'statuses': [a['status'], b['status']], 'same': a['body'] == b['body'], 'body': a['body']}}"
);
let output = code_mode
.call_tool("run_code", json!({ "code": code }))
.await
.unwrap();
assert!(!output.is_error, "{}", output.content);
let parsed: Value = serde_json::from_str(&output.content).unwrap();
assert_eq!(
parsed,
json!({"statuses": [200, 200], "same": true, "body": BODY})
);
}
#[tokio::test]
async fn duplicate_tool_names_route_to_the_first_server() {
let url = loopback_server().await;
let server = || StdioServerConfig {
command: env!("CARGO_BIN_EXE_asyncai-mcp-simple").to_string(),
args: Vec::new(),
};
let executor = McpToolExecutor::connect_stdio(vec![server(), server()])
.await
.unwrap();
let duplicates = executor.duplicates();
let names = duplicates
.iter()
.map(|duplicate| duplicate.name.as_str())
.collect::<Vec<_>>();
assert_eq!(names, ["http_get", "memo_set"]);
assert_eq!(executor.list_tools().await.unwrap().len(), 2);
let result = executor
.call_tool("http_get", json!({ "url": url }))
.await
.unwrap();
assert!(!result.is_error, "{}", result.content);
let unknown = executor.call_tool("nope", json!({})).await.unwrap_err();
assert!(unknown.to_string().contains("unknown tool \"nope\""));
}
#[tokio::test]
async fn http_get_connection_failure_is_an_error_result() {
let executor = simple_executor().await;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let closed = format!("http://{}/", listener.local_addr().unwrap());
drop(listener);
let result = executor
.call_tool("http_get", json!({ "url": closed }))
.await
.unwrap();
assert!(result.is_error, "{}", result.content);
assert!(result.content.contains("http_get error"));
}
