Skip to content

Commit f26a71e

Browse files
authored
Merge pull request agentclientprotocol#92 from nikomatsakis/main
feat(sacp): add cookbook module and concurrent tool support
2 parents 27d1f66 + 4273d08 commit f26a71e

27 files changed

Lines changed: 981 additions & 303 deletions

src/elizacp/tests/mcp_tool_invocation.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ async fn recv<T: sacp::JrResponsePayload + Send>(
1616
response: sacp::JrResponse<T>,
1717
) -> Result<T, sacp::Error> {
1818
let (tx, rx) = tokio::sync::oneshot::channel();
19-
response.await_when_result_received(async move |result| {
19+
response.on_receiving_result(async move |result| {
2020
tx.send(result).map_err(|_| sacp::Error::internal_error())
2121
})?;
2222
rx.await.map_err(|_| sacp::Error::internal_error())?

src/sacp-conductor/src/conductor.rs

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -612,7 +612,7 @@ impl ConductorResponder {
612612
meta: None,
613613
},
614614
)
615-
.await_when_result_received({
615+
.on_receiving_result({
616616
let mut conductor_tx = conductor_tx.clone();
617617
async move |result| {
618618
match result {
@@ -979,7 +979,7 @@ impl ConductorResponder {
979979
.as_ref()
980980
.expect("we have an agent component")
981981
.send_request(initialize_req)
982-
.await_when_result_received(async move |response| {
982+
.on_receiving_result(async move |response| {
983983
tracing::debug!(?response, "got initialize response from agent");
984984
request_cx
985985
.respond_with_result_via(conductor_tx, response)
@@ -999,7 +999,7 @@ impl ConductorResponder {
999999
// Forward initialize request to our successor
10001000
connection_cx
10011001
.send_request_to(Agent, initialize_req)
1002-
.await_when_result_received(async move |result| {
1002+
.on_receiving_result(async move |result| {
10031003
tracing::trace!(
10041004
?result,
10051005
"received response to initialize_proxy from empty conductor"
@@ -1016,7 +1016,7 @@ impl ConductorResponder {
10161016
let proxy_req = InitializeProxyRequest::from(initialize_req);
10171017
self.proxies[target_component_index]
10181018
.send_request(proxy_req)
1019-
.await_when_result_received(async move |result| {
1019+
.on_receiving_result(async move |result| {
10201020
tracing::debug!(?result, "got initialize_proxy response from proxy");
10211021
// Convert InitializeProxyResponse back to InitializeResponse
10221022
request_cx
@@ -1550,7 +1550,7 @@ impl<T: JrResponsePayload> JrResponseExt<T> for JrResponse<T> {
15501550
request_cx: JrRequestCx<T>,
15511551
) -> Result<(), sacp::Error> {
15521552
let conductor_tx = conductor_tx.clone();
1553-
self.await_when_result_received(async move |result| {
1553+
self.on_receiving_result(async move |result| {
15541554
request_cx
15551555
.respond_with_result_via(conductor_tx, result)
15561556
.await

src/sacp-conductor/tests/initialization_sequence.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ async fn recv<T: sacp::JrResponsePayload + Send>(
2121
response: sacp::JrResponse<T>,
2222
) -> Result<T, sacp::Error> {
2323
let (tx, rx) = tokio::sync::oneshot::channel();
24-
response.await_when_result_received(async move |result| {
24+
response.on_receiving_result(async move |result| {
2525
tx.send(result).map_err(|_| sacp::Error::internal_error())
2626
})?;
2727
rx.await.map_err(|_| sacp::Error::internal_error())?
@@ -83,7 +83,7 @@ impl Component for InitComponent {
8383

8484
// Forward to successor and respond
8585
cx.send_request_to(sacp::Agent, request)
86-
.await_when_result_received(async move |response| {
86+
.on_receiving_result(async move |response| {
8787
let response: InitializeResponse = response?;
8888
request_cx.respond(response)
8989
})

src/sacp-conductor/tests/mcp-integration.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ async fn recv<T: sacp::JrResponsePayload + Send>(
2525
response: sacp::JrResponse<T>,
2626
) -> Result<T, sacp::Error> {
2727
let (tx, rx) = tokio::sync::oneshot::channel();
28-
response.await_when_result_received(async move |result| {
28+
response.on_receiving_result(async move |result| {
2929
tx.send(result).map_err(|_| sacp::Error::internal_error())
3030
})?;
3131
rx.await.map_err(|_| sacp::Error::internal_error())?

src/sacp-conductor/tests/mcp_server_handler_chain.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ async fn recv<T: sacp::JrResponsePayload + Send>(
3838
response: sacp::JrResponse<T>,
3939
) -> Result<T, sacp::Error> {
4040
let (tx, rx) = tokio::sync::oneshot::channel();
41-
response.await_when_result_received(async move |result| {
41+
response.on_receiving_result(async move |result| {
4242
tx.send(result).map_err(|_| sacp::Error::internal_error())
4343
})?;
4444
rx.await.map_err(|_| sacp::Error::internal_error())?
@@ -101,7 +101,7 @@ impl Component for ProxyWithMcpAndHandler {
101101

102102
// Forward to agent and relay response
103103
cx.send_request_to(Agent, request)
104-
.await_when_result_received(async move |result| {
104+
.on_receiving_result(async move |result| {
105105
let response: NewSessionResponse = result?;
106106
request_cx.respond(response)
107107
})
Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
//! Integration test for `tool_fn` - stateless concurrent tools
2+
//!
3+
//! This test verifies that `tool_fn` works correctly for stateless tools
4+
//! that don't need mutable state.
5+
6+
use sacp::Component;
7+
use sacp::ProxyToConductor;
8+
use sacp::mcp_server::McpServer;
9+
use sacp_conductor::Conductor;
10+
use schemars::JsonSchema;
11+
use serde::{Deserialize, Serialize};
12+
use tokio::io::duplex;
13+
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
14+
15+
/// Input for the greet tool
16+
#[derive(Debug, Serialize, Deserialize, JsonSchema)]
17+
struct GreetInput {
18+
name: String,
19+
}
20+
21+
/// Create a proxy that provides an MCP server with a stateless greet tool
22+
fn create_greet_proxy() -> Result<sacp::DynComponent, sacp::Error> {
23+
// Create MCP server with a stateless greet tool using tool_fn
24+
let mcp_server = McpServer::builder("greet_server".to_string())
25+
.instructions("Test MCP server with stateless greet tool")
26+
.tool_fn(
27+
"greet",
28+
"Greet someone by name",
29+
async |input: GreetInput, _context| Ok(format!("Hello, {}!", input.name)),
30+
sacp::tool_fn!(),
31+
)
32+
.build();
33+
34+
// Create proxy component
35+
Ok(sacp::DynComponent::new(ProxyWithGreetServer { mcp_server }))
36+
}
37+
38+
struct ProxyWithGreetServer<R: sacp::JrResponder<ProxyToConductor>> {
39+
mcp_server: McpServer<ProxyToConductor, R>,
40+
}
41+
42+
impl<R: sacp::JrResponder<ProxyToConductor> + 'static + Send> Component
43+
for ProxyWithGreetServer<R>
44+
{
45+
async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
46+
ProxyToConductor::builder()
47+
.name("greet-proxy")
48+
.with_mcp_server(self.mcp_server)
49+
.serve(client)
50+
.await
51+
}
52+
}
53+
54+
/// Elizacp agent component wrapper for testing
55+
struct ElizacpAgentComponent;
56+
57+
impl Component for ElizacpAgentComponent {
58+
async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
59+
// Create duplex channels for bidirectional communication
60+
let (elizacp_write, client_read) = duplex(8192);
61+
let (client_write, elizacp_read) = duplex(8192);
62+
63+
let elizacp_transport =
64+
sacp::ByteStreams::new(elizacp_write.compat_write(), elizacp_read.compat());
65+
66+
let client_transport =
67+
sacp::ByteStreams::new(client_write.compat_write(), client_read.compat());
68+
69+
// Spawn elizacp in a background task
70+
tokio::spawn(async move {
71+
if let Err(e) = elizacp::ElizaAgent::new().serve(elizacp_transport).await {
72+
tracing::error!("Elizacp error: {}", e);
73+
}
74+
});
75+
76+
// Serve the client with the transport connected to elizacp
77+
client_transport.serve(client).await
78+
}
79+
}
80+
81+
#[tokio::test]
82+
async fn test_tool_fn_greet() -> Result<(), sacp::Error> {
83+
let result = yopo::prompt(
84+
Conductor::new(
85+
"test-conductor".to_string(),
86+
vec![
87+
create_greet_proxy()?,
88+
sacp::DynComponent::new(ElizacpAgentComponent),
89+
],
90+
Default::default(),
91+
),
92+
r#"Use tool greet_server::greet with {"name": "World"}"#,
93+
)
94+
.await?;
95+
96+
expect_test::expect![[r#"
97+
"OK: CallToolResult { content: [Annotated { raw: Text(RawTextContent { text: \"\\\"Hello, World!\\\"\", meta: None }), annotations: None }], structured_content: Some(String(\"Hello, World!\")), is_error: Some(false), meta: None }"
98+
"#]].assert_debug_eq(&result);
99+
100+
Ok(())
101+
}

src/sacp-tokio/tests/debug_logging.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ async fn recv<T: sacp::JrResponsePayload + Send>(
1212
response: sacp::JrResponse<T>,
1313
) -> Result<T, sacp::Error> {
1414
let (tx, rx) = tokio::sync::oneshot::channel();
15-
response.await_when_result_received(async move |result| {
15+
response.on_receiving_result(async move |result| {
1616
tx.send(result).map_err(|_| sacp::Error::internal_error())
1717
})?;
1818
rx.await.map_err(|_| sacp::Error::internal_error())?

0 commit comments

Comments
 (0)