feat(mcp): source gateway upstreams from the snapshot

crates/aisix-mcp/Cargo.toml

@@ -9,6 +9,7 @@ authors.workspace = true
 description = "aisix: MCP gateway — upstream client + downstream-facing aggregating endpoint (Streamable HTTP)"
 
 [dependencies]
+aisix-core = { path = "../aisix-core" }
 tokio.workspace = true
 async-trait.workspace = true
 futures.workspace = true
@@ -44,6 +45,7 @@ features = [
 ]
 
 [dev-dependencies]
+aisix-core = { path = "../aisix-core" }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 axum.workspace = true
 async-trait.workspace = true

crates/aisix-mcp/src/bridge.rs

@@ -13,6 +13,7 @@
 
 use std::time::Duration;
 
+use aisix_core::{McpAuthType, McpServer};
 use async_trait::async_trait;
 use rmcp::model::CallToolRequestParams;
 use rmcp::service::{RoleClient, RunningService};
@@ -226,3 +227,62 @@ fn into_mcp_tool(tool: rmcp::model::Tool) -> McpTool {
         input_schema: serde_json::Value::Object((*tool.input_schema).clone()),
     }
 }
+
+/// Build the connection parameters for an upstream from its registered
+/// [`McpServer`] resource: maps `auth_type`/`secret` to [`McpAuth`] and
+/// `timeout_ms` to the per-operation deadline.
+pub fn upstream_from_mcp_server(server: &McpServer) -> McpUpstream {
+    let auth = match server.auth_type {
+        McpAuthType::None => McpAuth::None,
+        McpAuthType::Bearer => McpAuth::Bearer(server.secret.clone().unwrap_or_default()),
+    };
+    let timeout = server
+        .timeout_ms
+        .map(Duration::from_millis)
+        .unwrap_or(DEFAULT_UPSTREAM_TIMEOUT);
+    McpUpstream {
+        url: server.url.clone(),
+        auth,
+        timeout,
+    }
+}
+
+/// An [`McpBridge`] that opens a fresh upstream session for each operation and
+/// drops it when done.
+///
+/// The downstream `/mcp` endpoint is stateless, so the gateway holds no
+/// long-lived upstream connections: every `tools/list` / `tools/call` connects,
+/// runs, and disconnects. Connection pooling is a later optimization; this keeps
+/// the snapshot-sourced gateway free of connection-lifecycle state, so a
+/// configuration change is picked up on the next request with nothing to
+/// reconcile.
+pub struct EphemeralBridge {
+    upstream: McpUpstream,
+}
+
+impl EphemeralBridge {
+    pub fn new(upstream: McpUpstream) -> Self {
+        Self { upstream }
+    }
+}
+
+#[async_trait]
+impl McpBridge for EphemeralBridge {
+    async fn list_tools(&self) -> Result<Vec<McpTool>, McpError> {
+        RmcpBridge::connect(&self.upstream)
+            .await?
+            .list_tools()
+            .await
+    }
+
+    async fn call_tool(
+        &self,
+        name: &str,
+        arguments: serde_json::Value,
+    ) -> Result<McpToolResult, McpError> {
+        RmcpBridge::connect(&self.upstream)
+            .await?
+            .call_tool(name, arguments)
+            .await
+    }
+}

crates/aisix-mcp/src/gateway.rs

@@ -35,7 +35,9 @@ use rmcp::transport::streamable_http_server::session::local::LocalSessionManager
 use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
 use rmcp::{RoleServer, ServerHandler};
 
-use crate::bridge::McpBridge;
+use aisix_core::AisixSnapshot;
+
+use crate::bridge::{upstream_from_mcp_server, EphemeralBridge, McpBridge};
 
 /// Separator between an upstream server's registered name and a tool name in
 /// the aggregated namespace, e.g. `github__create_issue`. Server names must
@@ -87,6 +89,26 @@ impl McpGateway {
         }
     }
 
+    /// Build a gateway whose upstreams are the **enabled** `mcp_servers` in the
+    /// snapshot, each reached through an [`EphemeralBridge`] (connect per
+    /// request). Disabled servers are skipped. Registration order follows the
+    /// snapshot's iteration order; duplicate display_names are deduped (first
+    /// wins) by [`McpGateway::new`], though the Admin API already enforces
+    /// uniqueness.
+    pub fn from_snapshot(snapshot: &AisixSnapshot) -> Self {
+        let upstreams = snapshot
+            .mcp_servers
+            .entries()
+            .into_iter()
+            .filter(|entry| entry.value.enabled)
+            .map(|entry| {
+                let upstream = upstream_from_mcp_server(&entry.value);
+                let bridge: Arc<dyn McpBridge> = Arc::new(EphemeralBridge::new(upstream));
+                (entry.value.display_name.clone(), bridge)
+            });
+        McpGateway::new(upstreams)
+    }
+
     fn find(&self, server: &str) -> Option<&Arc<dyn McpBridge>> {
         self.upstreams
             .iter()

crates/aisix-mcp/src/lib.rs

@@ -17,6 +17,9 @@ pub mod bridge;
 pub mod error;
 pub mod gateway;
 
-pub use bridge::{McpAuth, McpBridge, McpTool, McpToolResult, McpUpstream, RmcpBridge};
+pub use bridge::{
+    upstream_from_mcp_server, EphemeralBridge, McpAuth, McpBridge, McpTool, McpToolResult,
+    McpUpstream, RmcpBridge,
+};
 pub use error::McpError;
 pub use gateway::{streamable_http_service, McpGateway, TOOL_NAMESPACE_SEPARATOR};

crates/aisix-mcp/tests/gateway_aggregation.rs

@@ -13,9 +13,10 @@
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+use aisix_core::{AisixSnapshot, McpServer, ResourceEntry};
 use aisix_mcp::{
-    streamable_http_service, McpBridge, McpError, McpGateway, McpTool, McpToolResult, McpUpstream,
-    RmcpBridge,
+    streamable_http_service, upstream_from_mcp_server, McpAuth, McpBridge, McpError, McpGateway,
+    McpTool, McpToolResult, McpUpstream, RmcpBridge,
 };
 use rmcp::model::{
     CallToolRequestParams, CallToolResult, Content, ErrorData, ListToolsResult,
@@ -288,6 +289,82 @@ async fn duplicate_upstream_name_keeps_first() {
     );
 }
 
+#[test]
+fn upstream_from_mcp_server_maps_auth_and_timeout() {
+    let server: McpServer = serde_json::from_value(serde_json::json!({
+        "display_name": "gh",
+        "url": "https://api.example.com/mcp",
+        "auth_type": "bearer",
+        "secret": "tok",
+        "timeout_ms": 1234
+    }))
+    .unwrap();
+    let upstream = upstream_from_mcp_server(&server);
+    assert_eq!(upstream.url, "https://api.example.com/mcp");
+    assert_eq!(upstream.timeout, std::time::Duration::from_millis(1234));
+    assert!(matches!(upstream.auth, McpAuth::Bearer(ref t) if t == "tok"));
+
+    // `none` auth and absent timeout fall back to defaults.
+    let plain: McpServer = serde_json::from_value(serde_json::json!({
+        "display_name": "x", "url": "https://x/mcp"
+    }))
+    .unwrap();
+    assert!(matches!(
+        upstream_from_mcp_server(&plain).auth,
+        McpAuth::None
+    ));
+}
+
+/// Build a snapshot resource entry for an upstream at `addr`.
+fn mcp_entry(id: &str, name: &str, addr: &SocketAddr, enabled: bool) -> ResourceEntry<McpServer> {
+    let server: McpServer = serde_json::from_value(serde_json::json!({
+        "display_name": name,
+        "url": format!("http://{addr}/mcp"),
+        "enabled": enabled
+    }))
+    .unwrap();
+    ResourceEntry::new(id, server, 1)
+}
+
+#[tokio::test]
+async fn from_snapshot_sources_only_enabled_upstreams() {
+    let alpha = spawn_upstream("alpha").await;
+    let beta = spawn_upstream("beta").await;
+
+    let snapshot = AisixSnapshot::new();
+    snapshot
+        .mcp_servers
+        .insert(mcp_entry("e1", "alpha", &alpha, true));
+    snapshot
+        .mcp_servers
+        .insert(mcp_entry("e2", "beta", &beta, false)); // disabled
+
+    let gw_addr = spawn_gateway(McpGateway::from_snapshot(&snapshot)).await;
+    let client = ()
+        .serve(StreamableHttpClientTransport::from_uri(format!(
+            "http://{gw_addr}/mcp"
+        )))
+        .await
+        .expect("connect");
+
+    // Only the enabled server's tool is aggregated.
+    let tools = client.list_all_tools().await.expect("list tools");
+    let names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect();
+    assert_eq!(
+        tools.len(),
+        1,
+        "disabled upstream must be skipped: {names:?}"
+    );
+    assert_eq!(names[0], "alpha__echo");
+
+    // And it routes correctly (ephemeral connect per call).
+    let result = client
+        .call_tool(call("alpha__echo", "hi"))
+        .await
+        .expect("call");
+    assert_eq!(first_text(&result), "alpha:hi");
+}
+
 /// Build a `tools/call` for `name` with a single `text` argument.
 fn call(name: &'static str, text: &str) -> CallToolRequestParams {
     let args = serde_json::json!({ "text": text });