डायनेमिक MCP सर्वर प्रबंधन

McpServerManager स्थानीय MCP सर्वर चाइल्ड प्रोसेसों की एक रनटाइम रजिस्ट्री का स्वामी है। इसे तब उपयोग करें जब इंटीग्रेशन workspace, tenant, administrator selection, या deployment configuration के अनुसार बदलते हों।

यह remote HTTP सेवाओं के लिए connection pool नहीं है। उन्हें McpHttpClientBuilder के साथ बनाएँ और उनका lifecycle उस application में रखें जो remote configuration का स्वामी है।

जीवनचक्र

Rendering architecture…

मॉनिटर यह जाँचता है कि MCP connection बंद हुआ है या नहीं। यह crashed या failed starts को केवल तभी retry करता है जब configured RestartPolicy के पास अभी भी attempts शेष हों।

कॉन्फ़िगरेशन

{
  "mcpServers": {
    "workspace-tools": {
      "command": "/opt/company/bin/workspace-mcp",
      "args": ["--stdio", "--root", "/srv/workspace"],
      "env": {
        "RUST_LOG": "info"
      },
      "disabled": false,
      "autoApprove": [],
      "restartPolicy": {
        "initialDelayMs": 500,
        "maxDelayMs": 15000,
        "backoffMultiplier": 2.0,
        "maxRestartAttempts": 5
      }
    }
  }
}

Server IDs में ASCII letters, numbers, hyphens, और underscores स्वीकार किए जाते हैं। एक स्थिर ID का उपयोग करें क्योंकि यह collision-prefixed tool name का हिस्सा बन जाती है।

autoApprove को configuration compatibility के लिए read और write किया जाता है। manager इस field से approval नहीं देता।

टूल्स शुरू करें और aggregate करें

use adk_tool::mcp::manager::McpServerManager;
use std::sync::Arc;
use std::time::Duration;

let manager = Arc::new(McpServerManager::from_json_file("mcp.json")?
    .with_name("workspace_mcp")
    .with_health_check_interval(Duration::from_secs(15))
    .with_grace_period(Duration::from_secs(2)));

let outcomes = manager.start_all().await;
for (server_id, outcome) in outcomes {
    if let Err(error) = outcome {
        eprintln!("{server_id}: {error}");
    }
}

manager.start_monitoring();

let agent = LlmAgentBuilder::new("operator")
    .model(model)
    .toolset(manager.clone())
    .build()?;

Registry mutations को तब serial किया जाता है जब कोई child अपना MCP handshake पूरा कर रहा होता है। start_all हर enabled server के लिए एक स्वतंत्र result लौटाता है, लेकिन startup इस समय parallel-handshake path नहीं है।

रनटाइम पर registry बदलें

manager.add_server("billing".into(), billing_config).await?;
manager.start_server("billing").await?;

manager.update_server("billing", replacement_config).await?;
manager.disable_server("billing").await?;
manager.enable_server("billing").await?;

let snapshot = manager.all_configs().await;
manager.save_json_file("mcp.json").await?;

manager.remove_server("billing").await?;
manager.shutdown().await?;

चल रहे server को update करने पर वह रुकता है और replacement शुरू होता है। यदि replacement fail हो जाता है, तो manager replacement error लौटाने से पहले पिछली definition को restore और restart करता है।

save_json_file destination directory में एक temporary file लिखता है और फिर उसे destination पर rename कर देता है।

Resources, prompts, और notifications

एक managed server tools के अलावा resources और prompts भी publish कर सकता है। manager हर server की resource और prompt surface को server ID के आधार पर expose करता है, और resources/updated / resources/list_changed notifications को एक ऐसे handler तक पहुँचाता है जो हर managed connection में साझा होता है।

Handler को एक बार register करें; यह manual और automatic restarts के दौरान बना रहता है:

use adk_tool::{ResourceNotificationHandler, mcp::manager::McpServerManager};
use std::sync::Arc;

struct ReloadOnChange;

#[async_trait::async_trait]
impl ResourceNotificationHandler for ReloadOnChange {
    async fn handle_resource_updated(
        &self,
        uri: &str,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        tracing::info!(%uri, "resource changed; re-read it to refresh cached state");
        Ok(())
    }

    async fn handle_resource_list_changed(
        &self,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        Ok(())
    }
}

let manager = Arc::new(
    McpServerManager::from_json_file("mcp.json")?
        .with_resource_notification_handler(Arc::new(ReloadOnChange)),
);
manager.start_server("workspace-tools").await?;

फिर हर server के लिए read और subscribe करें:

let resources = manager.list_server_resources("workspace-tools").await?;
let templates = manager.list_server_resource_templates("workspace-tools").await?;
let contents = manager.read_server_resource("workspace-tools", "config://policy").await?;

let prompts = manager.list_server_prompts("workspace-tools").await?;
let review = manager
    .get_server_prompt("workspace-tools", "review_pr", None)
    .await?;

// Subscribe / unsubscribe. Subscriptions are restored automatically if the
// managed process reconnects.
manager.subscribe_server_resource("workspace-tools", "config://policy").await?;
manager.unsubscribe_server_resource("workspace-tools", "config://policy").await?;

हर *_server_* method ID के आधार पर एक running server को target करती है और यदि server unknown है या वर्तमान में running नहीं है, तो AdkError::Tool लौटाती है।

एक single connection के लिए (manager registry के बजाय), वही surface सीधे McpToolset पर McpToolset::with_handlers, list_resources, read_resource, list_prompts, get_prompt, और subscribe_resource के माध्यम से उपलब्ध है। runnable agentic example देखें:

cargo run --manifest-path examples/mcp_resources/Cargo.toml --bin resources-client

टूल नाम टकराव

यदि दो running servers search publish करते हैं, तो aggregated names इस प्रकार बनते हैं:

crm__search
knowledge__search

केवल एक server द्वारा publish किया गया tool name अपरिवर्तित रहता है।

शटडाउन व्यवहार

किसी server को रोकने पर उसका MCP session cancel हो जाता है, connection बंद होने के लिए configured grace period तक प्रतीक्षा की जाती है, फिर transport drop कर दिया जाता है। child process को अलग से retained process handle के बजाय TokioChildProcess transport owns करता है।

manager को drop करने से पहले shutdown() को call करें। चल रहे servers वाले manager को drop करने पर warning emit होती है क्योंकि Drop asynchronous cleanup की प्रतीक्षा नहीं कर सकता।

सत्यापित उदाहरण

cargo run --manifest-path examples/mcp_manager/Cargo.toml

यह example एक real Rust MCP child process शुरू करता है, एक tool खोजता और call करता है, एक दूसरा server जोड़ता और enable करता है, उसे update करता है, registry save करता है, उसे disable और remove करता है, और सभी sessions बंद करता है। इसे किसी model, API key, package download, या network access की आवश्यकता नहीं होती।