إدارة خادم MCP الديناميكية

McpServerManager يمتلك سجلًا وقت التشغيل لعمليات خادم MCP المحلية الفرعية. استخدمه عندما تتغير عمليات التكامل حسب مساحة العمل، أو المستأجر، أو اختيار المسؤول، أو إعدادات النشر.

إنه ليس مجموعة الاتصالات لخدمات HTTP البعيدة. ابنِ تلك باستخدام McpHttpClientBuilder وأبقِ دورة حياتها في التطبيق الذي يملك الإعداد البعيد.

دورة الحياة

Rendering architecture…

يتحقق المراقب مما إذا كان اتصال MCP قد أُغلق. ويعيد المحاولة عند تعطل البدء أو فشله فقط ما دام أن RestartPolicy المُكوَّن لا يزال لديه محاولات متبقية.

الإعداد

{
  "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
      }
    }
  }
}

تقبل معرّفات الخوادم أحرف ASCII والأرقام والشرطات والشرطات السفلية. استخدم معرّفًا ثابتًا لأنه يصبح جزءًا من اسم أداة مسبوق بتعارض.

autoApprove يُقرأ ويُكتب من أجل توافق الإعداد. لا يمنح المدير الموافقة من هذا الحقل.

بدء الأدوات وتجميعها

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()?;

تُسلسَل تعديلات السجل بينما يُكمل أحد العناصر الفرعية مصافحة MCP الخاصة به. start_all تُرجع نتيجة مستقلة لكل خادم مُمكَّن، لكن البدء ليس حاليًا مسار مصافحة متوازٍ.

تغيير السجل وقت التشغيل

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?;

تحديث خادم يعمل يوقفه ثم يبدأ البديل. إذا فشل البديل، يستعيد المدير التعريف السابق ويعيد تشغيله قبل إرجاع خطأ البديل.

save_json_file يكتب ملفًا مؤقتًا في دليل الوجهة ثم يعيد تسميته فوق الوجهة.

الموارد والمطالبات والإشعارات

يمكن للخادم المُدار أن ينشر موارد ومطالبات بالإضافة إلى الأدوات. يعرض المدير واجهة الموارد والمطالبات لكل خادم بواسطة معرّف الخادم، ويمرر إشعارات resources/updated / resources/list_changed إلى معالج مشترك عبر كل اتصال مُدار.

سجّل المعالج مرة واحدة؛ ويظل محتفظًا به عبر عمليات إعادة التشغيل اليدوية والتلقائية:

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?;

ثم اقرأ واشترك لكل خادم:

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_* تستهدف خادمًا واحدًا يعمل بواسطة المعرّف وتُرجع AdkError::Tool إذا كان الخادم غير معروف أو غير قيد التشغيل حاليًا.

لاتصال واحد (بدلًا من سجل المدير)، تتوفر الواجهة نفسها مباشرة على McpToolset عبر McpToolset::with_handlers، list_resources، read_resource، list_prompts، get_prompt، و subscribe_resource. انظر المثال العملي القابل للتشغيل:

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

تعارضات أسماء الأدوات

إذا نشر خادمان يعملان search، تصبح الأسماء المجمعة:

crm__search
knowledge__search

اسم الأداة الذي ينشره خادم واحد فقط يظل دون تغيير.

سلوك الإيقاف

إيقاف خادم يلغي جلسة MCP الخاصة به، وينتظر حتى فترة السماح المُكوَّنة لإغلاق الاتصال، ثم يفصل وسيط النقل. عملية الطفل مملوكة لوسيط نقل TokioChildProcess بدلًا من مقبض عملية محتفظ به بشكل منفصل.

استدعِ shutdown() قبل التخلص من المدير. التخلص من مدير لديه خوادم تعمل يطلق تحذيرًا لأن Drop لا يمكنه انتظار الإنهاء غير المتزامن.

مثال مُتحقق منه

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

يبدأ المثال عملية فرعية فعلية لخادم Rust MCP، ويكتشف أداة ويستدعيها، ويضيف خادمًا ثانيًا ويمكّنه، ويحدّثه، ويحفظ السجل، ويعطّلها ويزيله، ويغلق جميع الجلسات. لا يتطلب أي نموذج، أو مفتاح API، أو تنزيل حزمة، أو وصولًا إلى الشبكة.