agent_client_protocol_rmcp/
lib.rs1use agent_client_protocol::mcp_server::{McpConnectionTo, McpServer, McpServerConnect};
43use agent_client_protocol::role;
44use agent_client_protocol::{ByteStreams, ConnectTo, DynConnectTo, NullRun, Role};
45use futures_concurrency::future::TryJoin as _;
46use rmcp::ServiceExt;
47use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
48
49mod builder;
50
51pub use agent_client_protocol::mcp_server::{EnabledTools, McpTool};
52pub use agent_client_protocol::{tool_fn, tool_fn_mut};
53pub use builder::McpServerBuilder;
54
55pub trait McpServerExt<Counterpart: Role> {
57 fn builder(name: impl ToString) -> McpServerBuilder<Counterpart, NullRun> {
59 McpServerBuilder::new(name.to_string())
60 }
61
62 fn from_rmcp<S>(
68 name: impl ToString,
69 new_fn: impl Fn() -> S + Send + Sync + 'static,
70 ) -> McpServer<Counterpart, NullRun>
71 where
72 S: rmcp::Service<rmcp::RoleServer>,
73 {
74 struct RmcpServer<F> {
75 name: String,
76 new_fn: F,
77 }
78
79 impl<Counterpart, F, S> McpServerConnect<Counterpart> for RmcpServer<F>
80 where
81 Counterpart: Role,
82 F: Fn() -> S + Send + Sync + 'static,
83 S: rmcp::Service<rmcp::RoleServer>,
84 {
85 fn name(&self) -> String {
86 self.name.clone()
87 }
88
89 fn connect(
90 &self,
91 _cx: McpConnectionTo<Counterpart>,
92 ) -> DynConnectTo<role::mcp::Client> {
93 let service = (self.new_fn)();
94 DynConnectTo::new(RmcpServerComponent { service })
95 }
96 }
97
98 McpServer::new(
99 RmcpServer {
100 name: name.to_string(),
101 new_fn,
102 },
103 NullRun,
104 )
105 }
106}
107
108impl<Counterpart: Role> McpServerExt<Counterpart> for McpServer<Counterpart> {}
109
110struct RmcpServerComponent<S> {
112 service: S,
113}
114
115impl<S> ConnectTo<role::mcp::Client> for RmcpServerComponent<S>
116where
117 S: rmcp::Service<rmcp::RoleServer>,
118{
119 async fn connect_to(
120 self,
121 client: impl ConnectTo<role::mcp::Server>,
122 ) -> Result<(), agent_client_protocol::Error> {
123 let (mcp_server_stream, mcp_client_stream) = tokio::io::duplex(8192);
125 let (mcp_server_read, mcp_server_write) = tokio::io::split(mcp_server_stream);
126 let (mcp_client_read, mcp_client_write) = tokio::io::split(mcp_client_stream);
127
128 let bytes_to_acp = async {
129 let byte_streams =
131 ByteStreams::new(mcp_client_write.compat_write(), mcp_client_read.compat());
132
133 drop(ConnectTo::<role::mcp::Client>::connect_to(byte_streams, client).await);
135
136 Ok(())
137 };
138
139 let bytes_to_rmcp = async {
140 let running_server = self
142 .service
143 .serve((mcp_server_read, mcp_server_write))
144 .await
145 .map_err(agent_client_protocol::Error::into_internal_error)?;
146
147 running_server
149 .waiting()
150 .await
151 .map(|_quit_reason| ())
152 .map_err(agent_client_protocol::Error::into_internal_error)
153 };
154
155 (bytes_to_acp, bytes_to_rmcp).try_join().await?;
156 Ok(())
157 }
158}
159
160#[cfg(test)]
161mod tests {
162 use agent_client_protocol::{mcp_server::McpServer, role};
163
164 use crate::McpServerExt as _;
165
166 #[test]
167 fn builds_standalone_server_without_acp_transport_feature() {
168 let _server: McpServer<role::mcp::Client, _> = McpServer::builder("standalone").build();
169 }
170}