Struct McpService

pub struct McpService<T: TransportHandle> { /* private fields */ }

A wrapper service that implements Tower’s Service trait for MCP transport

Implementations

impl<T: TransportHandle> McpService<T>

pub fn new(transport: T) -> Self

impl<T> McpService<T>

where T: TransportHandle,

pub fn with_timeout(transport: T, timeout: Duration) -> Timeout<McpService<T>>

Examples found in repository

examples/stdio.rs (line 29)

async fn main() -> Result<(), ClientError> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env()
                .add_directive("mcp_client=debug".parse().unwrap())
                .add_directive("eventsource_client=debug".parse().unwrap()),
        )
        .init();

    // 1) Create the transport
    let transport = StdioTransport::new("uvx", vec!["mcp-server-git".to_string()], HashMap::new());

    // 2) Start the transport to get a handle
    let transport_handle = transport.start().await?;

    // 3) Create the service with timeout middleware
    let service = McpService::with_timeout(transport_handle, Duration::from_secs(10));

    // 4) Create the client with the middleware-wrapped service
    let mut client = McpClient::new(service);

    // Initialize
    let server_info = client
        .initialize(
            ClientInfo {
                name: "test-client".into(),
                version: "1.0.0".into(),
            },
            ClientCapabilities::default(),
        )
        .await?;
    println!("Connected to server: {server_info:?}\n");

    // List tools
    let tools = client.list_tools(None).await?;
    println!("Available tools: {tools:?}\n");

    // Call tool 'git_status' with arguments = {"repo_path": "."}
    let tool_result = client
        .call_tool("git_status", serde_json::json!({ "repo_path": "." }))
        .await?;
    println!("Tool result: {tool_result:?}\n");

    // List resources
    let resources = client.list_resources(None).await?;
    println!("Available resources: {resources:?}\n");

    Ok(())
}

examples/sse.rs (line 27)

async fn main() -> Result<()> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env()
                .add_directive("mcp_client=debug".parse().unwrap())
                .add_directive("eventsource_client=info".parse().unwrap()),
        )
        .init();

    // Create the base transport
    let transport = SseTransport::new("http://localhost:8000/sse", HashMap::new());

    // Start transport
    let handle = transport.start().await?;

    // Create the service with timeout middleware
    let service = McpService::with_timeout(handle, Duration::from_secs(3));

    // Create client
    let mut client = McpClient::new(service);
    println!("Client created\n");

    // Initialize
    let server_info = client
        .initialize(
            ClientInfo {
                name: "test-client".into(),
                version: "1.0.0".into(),
            },
            ClientCapabilities::default(),
        )
        .await?;
    println!("Connected to server: {server_info:?}\n");

    // Sleep for 100ms to allow the server to start - surprisingly this is required!
    tokio::time::sleep(Duration::from_millis(500)).await;

    // List tools
    let tools = client.list_tools(None).await?;
    println!("Available tools: {tools:?}\n");

    // Call tool
    let tool_result = client
        .call_tool(
            "echo_tool",
            serde_json::json!({ "message": "Client with SSE transport - calling a tool" }),
        )
        .await?;
    println!("Tool result: {tool_result:?}\n");

    // List resources
    let resources = client.list_resources(None).await?;
    println!("Resources: {resources:?}\n");

    // Read resource
    let resource = client.read_resource("echo://fixedresource").await?;
    println!("Resource: {resource:?}\n");

    Ok(())
}

examples/stdio_integration.rs (line 38)

async fn main() -> Result<(), ClientError> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env()
                .add_directive("mcp_client=debug".parse().unwrap())
                .add_directive("eventsource_client=debug".parse().unwrap()),
        )
        .init();

    // Create the transport
    let transport = StdioTransport::new(
        "cargo",
        vec!["run", "-p", "mcp-server"]
            .into_iter()
            .map(|s| s.to_string())
            .collect(),
        HashMap::new(),
    );

    // Start the transport to get a handle
    let transport_handle = transport.start().await.unwrap();

    // Create the service with timeout middleware
    let service = McpService::with_timeout(transport_handle, Duration::from_secs(10));

    // Create client
    let mut client = McpClient::new(service);

    // Initialize
    let server_info = client
        .initialize(
            ClientInfo {
                name: "test-client".into(),
                version: "1.0.0".into(),
            },
            ClientCapabilities::default(),
        )
        .await?;
    println!("Connected to server: {server_info:?}\n");

    // List tools
    let tools = client.list_tools(None).await?;
    println!("Available tools: {tools:?}\n");

    // Call tool 'increment' tool 3 times
    for _ in 0..3 {
        let increment_result = client.call_tool("increment", serde_json::json!({})).await?;
        println!("Tool result for 'increment': {increment_result:?}\n");
    }

    // Call tool 'get_value'
    let get_value_result = client.call_tool("get_value", serde_json::json!({})).await?;
    println!("Tool result for 'get_value': {get_value_result:?}\n");

    // Call tool 'decrement' once
    let decrement_result = client.call_tool("decrement", serde_json::json!({})).await?;
    println!("Tool result for 'decrement': {decrement_result:?}\n");

    // Call tool 'get_value'
    let get_value_result = client.call_tool("get_value", serde_json::json!({})).await?;
    println!("Tool result for 'get_value': {get_value_result:?}\n");

    // List resources
    let resources = client.list_resources(None).await?;
    println!("Resources: {resources:?}\n");

    // Read resource
    let resource = client.read_resource("memo://insights").await?;
    println!("Resource: {resource:?}\n");

    let prompts = client.list_prompts(None).await?;
    println!("Prompts: {prompts:?}\n");

    let prompt = client
        .get_prompt(
            "example_prompt",
            serde_json::json!({"message": "hello there!"}),
        )
        .await?;
    println!("Prompt: {prompt:?}\n");

    Ok(())
}

examples/clients.rs (line 23)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env().add_directive("mcp_client=debug".parse().unwrap()),
        )
        .init();

    let transport1 = StdioTransport::new("uvx", vec!["mcp-server-git".to_string()], HashMap::new());
    let handle1 = transport1.start().await?;
    let service1 = McpService::with_timeout(handle1, Duration::from_secs(30));
    let client1 = McpClient::new(service1);

    let transport2 = StdioTransport::new("uvx", vec!["mcp-server-git".to_string()], HashMap::new());
    let handle2 = transport2.start().await?;
    let service2 = McpService::with_timeout(handle2, Duration::from_secs(30));
    let client2 = McpClient::new(service2);

    let transport3 = SseTransport::new("http://localhost:8000/sse", HashMap::new());
    let handle3 = transport3.start().await?;
    let service3 = McpService::with_timeout(handle3, Duration::from_secs(10));
    let client3 = McpClient::new(service3);

    // Initialize both clients
    let mut clients: Vec<Box<dyn McpClientTrait>> =
        vec![Box::new(client1), Box::new(client2), Box::new(client3)];

    // Initialize all clients
    for (i, client) in clients.iter_mut().enumerate() {
        let info = ClientInfo {
            name: format!("example-client-{}", i + 1),
            version: "1.0.0".to_string(),
        };
        let capabilities = ClientCapabilities::default();

        println!("\nInitializing client {}", i + 1);
        let init_result = client.initialize(info, capabilities).await?;
        println!("Client {} initialized: {:?}", i + 1, init_result);
    }

    // List tools for all clients
    for (i, client) in clients.iter_mut().enumerate() {
        let tools = client.list_tools(None).await?;
        println!("\nClient {} tools: {:?}", i + 1, tools);
    }

    println!("\n\n----------------------------------\n\n");

    // Wrap clients in Arc before spawning tasks
    let clients = Arc::new(clients);
    let mut handles = vec![];

    for i in 0..20 {
        let clients = Arc::clone(&clients);
        let handle = tokio::spawn(async move {
            // let mut rng = rand::thread_rng();
            let mut rng = rand::rngs::StdRng::from_entropy();
            tokio::time::sleep(Duration::from_millis(rng.gen_range(5..50))).await;

            // Randomly select an operation
            match rng.gen_range(0..4) {
                0 => {
                    println!("\n{i}: Listing tools for client 1 (stdio)");
                    match clients[0].list_tools(None).await {
                        Ok(tools) => {
                            println!("  {i}: -> Got tools, first one: {:?}", tools.tools.first())
                        }
                        Err(e) => println!("  {i}: -> Error: {}", e),
                    }
                }
                1 => {
                    println!("\n{i}: Calling tool for client 2 (stdio)");
                    match clients[1]
                        .call_tool("git_status", serde_json::json!({ "repo_path": "." }))
                        .await
                    {
                        Ok(result) => println!(
                            "  {i}: -> Tool execution result, is_error: {:?}",
                            result.is_error
                        ),
                        Err(e) => println!("  {i}: -> Error: {}", e),
                    }
                }
                2 => {
                    println!("\n{i}: Listing tools for client 3 (sse)");
                    match clients[2].list_tools(None).await {
                        Ok(tools) => {
                            println!("  {i}: -> Got tools, first one: {:?}", tools.tools.first())
                        }
                        Err(e) => println!("  {i}: -> Error: {}", e),
                    }
                }
                3 => {
                    println!("\n{i}: Calling tool for client 3 (sse)");
                    match clients[2]
                            .call_tool(
                                "echo_tool",
                                serde_json::json!({ "message": "Client with SSE transport - calling a tool" }),
                            )
                            .await
                        {
                        Ok(result) => println!("  {i}: -> Tool execution result, is_error: {:?}", result.is_error),
                        Err(e) => println!("  {i}: -> Error: {}", e),
                    }
                }
                _ => unreachable!(),
            }
            Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
        });
        handles.push(handle);
    }

    // Wait for all tasks to complete
    for handle in handles {
        handle.await.unwrap().unwrap();
    }

    Ok(())
}

Trait Implementations

impl<T: Clone + TransportHandle> Clone for McpService<T>

fn clone(&self) -> McpService<T>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T> Service<JsonRpcMessage> for McpService<T>

where T: TransportHandle + Send + Sync + 'static,

type Response = JsonRpcMessage

Responses given by the service.

type Error = Error

Errors produced by the service.

type Future = Pin<Box<dyn Future<Output = Result<<McpService<T> as Service<JsonRpcMessage>>::Response, <McpService<T> as Service<JsonRpcMessage>>::Error>> + Send>>

The future response value.

fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, request: JsonRpcMessage) -> Self::Future

Process the request and return the response asynchronously.