From 0a55fd82939cc52db048f28c5e3524d960ea8348 Mon Sep 17 00:00:00 2001 From: Ming Wen Date: Mon, 29 Jun 2026 15:59:39 +0800 Subject: [PATCH 1/2] feat(mcp): add McpServer admin resource (registration + CRUD) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `McpServer` as a first-class Admin resource, parallel to ProviderKey: an upstream MCP server registration (name, url, Streamable HTTP transport, gateway-held auth, timeout, enabled) that the MCP gateway endpoint will source upstreams from. - aisix-core: `McpServer` struct + `McpTransport`/`McpAuthType` enums + Resource impl (`kind = "mcp_servers"`); schema validator wired through the same `struct_root_schema` producer as the other resources (published == enforced), registered in dump-schema; new `mcp_servers` snapshot table. - aisix-etcd: loader decode arm + supervisor merge / present / delete / clone dispatch for the new kind. - aisix-admin: `ConfigStore` get/put/list/delete + InMemoryStore + EtcdConfigStore (subkey `mcp_servers`); `/admin/v1/mcp_servers[/:id]` handlers (validate, dup-name 409, reject the reserved `__` separator in display_name, uuid on POST, revision bump on PUT) + routes. - tests: resource unit tests, an etcd CRUD round-trip, and the new kind added to `loader_picks_up_every_admin_write` (asserts the Admin โ†’ EtcdConfigStore โ†’ loader path agrees on the subkey constant). The committed `schemas/resources/mcp_server.schema.json` is regenerated via `cargo run -p aisix-core --bin dump-schema`. OpenAPI reference for the new routes is deferred to #663 (no functionality depends on it; the routes work, only the generated OpenAPI omits them). Refs AISIX-Cloud#894 --- crates/aisix-admin/src/etcd_store.rs | 34 +++- crates/aisix-admin/src/lib.rs | 12 ++ .../aisix-admin/src/mcp_servers_handlers.rs | 119 +++++++++++ crates/aisix-admin/src/store.rs | 33 +++- crates/aisix-admin/tests/etcd_integration.rs | 28 ++- crates/aisix-core/src/bin/dump-schema.rs | 1 + crates/aisix-core/src/lib.rs | 9 +- crates/aisix-core/src/models/mcp_server.rs | 185 ++++++++++++++++++ crates/aisix-core/src/models/mod.rs | 4 +- crates/aisix-core/src/models/schema.rs | 19 ++ crates/aisix-core/src/models/snapshot.rs | 6 + crates/aisix-etcd/src/loader.rs | 18 +- crates/aisix-etcd/src/supervisor.rs | 10 + schemas/resources/mcp_server.schema.json | 94 +++++++++ 14 files changed, 561 insertions(+), 11 deletions(-) create mode 100644 crates/aisix-admin/src/mcp_servers_handlers.rs create mode 100644 crates/aisix-core/src/models/mcp_server.rs create mode 100644 schemas/resources/mcp_server.schema.json diff --git a/crates/aisix-admin/src/etcd_store.rs b/crates/aisix-admin/src/etcd_store.rs index b35ad1c6..7c65b1bb 100644 --- a/crates/aisix-admin/src/etcd_store.rs +++ b/crates/aisix-admin/src/etcd_store.rs @@ -19,7 +19,9 @@ //! deterministic behaviour continue to use [`crate::InMemoryStore`]. use aisix_core::resource::ResourceEntry; -use aisix_core::{ApiKey, CachePolicy, Guardrail, Model, ObservabilityExporter, ProviderKey}; +use aisix_core::{ + ApiKey, CachePolicy, Guardrail, McpServer, Model, ObservabilityExporter, ProviderKey, +}; use etcd_client::{Client, DeleteOptions, GetOptions}; use serde::de::DeserializeOwned; use serde::Serialize; @@ -35,6 +37,7 @@ pub const PROVIDER_KEYS_SUBKEY: &str = "provider_keys"; pub const GUARDRAILS_SUBKEY: &str = "guardrails"; pub const CACHE_POLICIES_SUBKEY: &str = "cache_policies"; pub const OBSERVABILITY_EXPORTERS_SUBKEY: &str = "observability_exporters"; +pub const MCP_SERVERS_SUBKEY: &str = "mcp_servers"; pub struct EtcdConfigStore { client: Mutex, @@ -333,6 +336,35 @@ impl ConfigStore for EtcdConfigStore { self.delete_one(&self.key_for(OBSERVABILITY_EXPORTERS_SUBKEY, id)) .await } + + async fn put_mcp_server(&self, entry: ResourceEntry) -> Result<(), StoreError> { + let key = self.key_for(MCP_SERVERS_SUBKEY, &entry.id); + self.put_json(&key, &entry.value).await + } + + async fn get_mcp_server( + &self, + id: &str, + ) -> Result>, StoreError> { + let key = self.key_for(MCP_SERVERS_SUBKEY, id); + Ok(self + .get_one::(&key) + .await? + .map(|(v, rev)| ResourceEntry::new(id, v, rev))) + } + + async fn list_mcp_servers(&self) -> Result>, StoreError> { + Ok(self + .list_range::(MCP_SERVERS_SUBKEY) + .await? + .into_iter() + .map(|(id, v, rev)| ResourceEntry::new(id, v, rev)) + .collect()) + } + + async fn delete_mcp_server(&self, id: &str) -> Result { + self.delete_one(&self.key_for(MCP_SERVERS_SUBKEY, id)).await + } } #[cfg(test)] diff --git a/crates/aisix-admin/src/lib.rs b/crates/aisix-admin/src/lib.rs index 875ed0aa..24f8cf60 100644 --- a/crates/aisix-admin/src/lib.rs +++ b/crates/aisix-admin/src/lib.rs @@ -41,6 +41,7 @@ mod error; pub mod etcd_store; mod guardrails_handlers; mod health_handler; +mod mcp_servers_handlers; mod models_handlers; mod models_status_handler; mod observability_exporters_handlers; @@ -119,6 +120,17 @@ pub fn build_router(state: AdminState) -> Router { .put(provider_keys_handlers::update_provider_key) .delete(provider_keys_handlers::delete_provider_key), ) + .route( + "/admin/v1/mcp_servers", + get(mcp_servers_handlers::list_mcp_servers) + .post(mcp_servers_handlers::create_mcp_server), + ) + .route( + "/admin/v1/mcp_servers/:id", + get(mcp_servers_handlers::get_mcp_server) + .put(mcp_servers_handlers::update_mcp_server) + .delete(mcp_servers_handlers::delete_mcp_server), + ) .route( "/admin/v1/guardrails", get(guardrails_handlers::list_guardrails) diff --git a/crates/aisix-admin/src/mcp_servers_handlers.rs b/crates/aisix-admin/src/mcp_servers_handlers.rs new file mode 100644 index 00000000..1dda5cab --- /dev/null +++ b/crates/aisix-admin/src/mcp_servers_handlers.rs @@ -0,0 +1,119 @@ +//! CRUD handlers for `/admin/v1/mcp_servers`. +//! +//! Same shape as the ProviderKeys handlers: validate against the JSON schema, +//! reject duplicate display_names (409), generate a uuid v4 on POST, bump +//! revision on PUT. Additionally rejects a display_name containing the reserved +//! tool-namespace separator `__`, since the name prefixes the server's tools. + +use aisix_core::models::validate_mcp_server; +use aisix_core::resource::ResourceEntry; +use aisix_core::McpServer; +use axum::extract::{Path, State}; +use axum::Json; +use serde_json::Value; +use uuid::Uuid; + +use crate::auth::AdminAuth; +use crate::error::AdminError; +use crate::state::AdminState; + +const STARTING_REVISION: i64 = 1; + +/// Reserved separator between a server's name and a tool name in the gateway's +/// aggregated namespace (`__`). A server name must not +/// contain it. +const TOOL_NAMESPACE_SEPARATOR: &str = "__"; + +pub async fn list_mcp_servers( + _auth: AdminAuth, + State(state): State, +) -> Result>>, AdminError> { + let entries = state.store.list_mcp_servers().await?; + Ok(Json(entries)) +} + +pub async fn get_mcp_server( + _auth: AdminAuth, + Path(id): Path, + State(state): State, +) -> Result>, AdminError> { + let entry = state + .store + .get_mcp_server(&id) + .await? + .ok_or(AdminError::NotFound)?; + Ok(Json(entry)) +} + +pub async fn create_mcp_server( + _auth: AdminAuth, + State(state): State, + Json(raw): Json, +) -> Result>, AdminError> { + let mcp_server = decode(&raw)?; + let all = state.store.list_mcp_servers().await?; + assert_unique_display_name(&all, &mcp_server.display_name, None)?; + + let id = Uuid::new_v4().to_string(); + let entry = ResourceEntry::new(&id, mcp_server, STARTING_REVISION); + state.store.put_mcp_server(entry.clone()).await?; + Ok(Json(entry)) +} + +pub async fn update_mcp_server( + _auth: AdminAuth, + Path(id): Path, + State(state): State, + Json(raw): Json, +) -> Result>, AdminError> { + let existing = state + .store + .get_mcp_server(&id) + .await? + .ok_or(AdminError::NotFound)?; + let mcp_server = decode(&raw)?; + + let all = state.store.list_mcp_servers().await?; + assert_unique_display_name(&all, &mcp_server.display_name, Some(&id))?; + + let entry = ResourceEntry::new(&id, mcp_server, existing.revision + 1); + state.store.put_mcp_server(entry.clone()).await?; + Ok(Json(entry)) +} + +pub async fn delete_mcp_server( + _auth: AdminAuth, + Path(id): Path, + State(state): State, +) -> Result, AdminError> { + let removed = state.store.delete_mcp_server(&id).await?; + if !removed { + return Err(AdminError::NotFound); + } + Ok(Json(serde_json::json!({"deleted": true, "id": id}))) +} + +fn decode(raw: &Value) -> Result { + validate_mcp_server(raw)?; + let server: McpServer = serde_json::from_value(raw.clone()) + .map_err(|e| AdminError::BadRequest(format!("malformed McpServer payload: {e}")))?; + if server.display_name.contains(TOOL_NAMESPACE_SEPARATOR) { + return Err(AdminError::BadRequest(format!( + "display_name must not contain the reserved separator `{TOOL_NAMESPACE_SEPARATOR}`" + ))); + } + Ok(server) +} + +fn assert_unique_display_name( + existing: &[ResourceEntry], + display_name: &str, + self_id: Option<&str>, +) -> Result<(), AdminError> { + for e in existing { + if e.value.display_name == display_name && self_id.is_none_or(|sid| sid != e.id) { + return Err(AdminError::Conflict(display_name.to_string())); + } + } + Ok(()) +} diff --git a/crates/aisix-admin/src/store.rs b/crates/aisix-admin/src/store.rs index 3f198ab3..16c2f872 100644 --- a/crates/aisix-admin/src/store.rs +++ b/crates/aisix-admin/src/store.rs @@ -7,7 +7,9 @@ //! in the handler layer so the store stays dumb and fast. use aisix_core::resource::ResourceEntry; -use aisix_core::{ApiKey, CachePolicy, Guardrail, Model, ObservabilityExporter, ProviderKey}; +use aisix_core::{ + ApiKey, CachePolicy, Guardrail, McpServer, Model, ObservabilityExporter, ProviderKey, +}; use dashmap::DashMap; use std::sync::Arc; @@ -65,6 +67,14 @@ pub trait ConfigStore: Send + Sync + 'static { &self, ) -> Result>, StoreError>; async fn delete_observability_exporter(&self, id: &str) -> Result; + + async fn put_mcp_server(&self, entry: ResourceEntry) -> Result<(), StoreError>; + async fn get_mcp_server( + &self, + id: &str, + ) -> Result>, StoreError>; + async fn list_mcp_servers(&self) -> Result>, StoreError>; + async fn delete_mcp_server(&self, id: &str) -> Result; } /// In-memory store. Thread-safe via DashMap; mainly used by tests, but @@ -77,6 +87,7 @@ pub struct InMemoryStore { guardrails: DashMap>, cache_policies: DashMap>, observability_exporters: DashMap>, + mcp_servers: DashMap>, } impl InMemoryStore { @@ -209,6 +220,26 @@ impl ConfigStore for InMemoryStore { async fn delete_observability_exporter(&self, id: &str) -> Result { Ok(self.observability_exporters.remove(id).is_some()) } + + async fn put_mcp_server(&self, entry: ResourceEntry) -> Result<(), StoreError> { + self.mcp_servers.insert(entry.id.clone(), entry); + Ok(()) + } + + async fn get_mcp_server( + &self, + id: &str, + ) -> Result>, StoreError> { + Ok(self.mcp_servers.get(id).map(|r| r.clone())) + } + + async fn list_mcp_servers(&self) -> Result>, StoreError> { + Ok(self.mcp_servers.iter().map(|r| r.clone()).collect()) + } + + async fn delete_mcp_server(&self, id: &str) -> Result { + Ok(self.mcp_servers.remove(id).is_some()) + } } #[cfg(test)] diff --git a/crates/aisix-admin/tests/etcd_integration.rs b/crates/aisix-admin/tests/etcd_integration.rs index 1cf6469a..9222e93d 100644 --- a/crates/aisix-admin/tests/etcd_integration.rs +++ b/crates/aisix-admin/tests/etcd_integration.rs @@ -196,6 +196,27 @@ async fn provider_keys_round_trip_through_real_etcd() { .await; } +#[tokio::test] +async fn mcp_servers_round_trip_through_real_etcd() { + let Some(url) = etcd_url() else { + eprintln!("skipping: ADMIN_TEST_ETCD_URL not set"); + return; + }; + let prefix = unique_prefix(); + let state = build_state_with_real_etcd(&url, &prefix).await; + admin_crud_round_trip( + state, + "/admin/v1/mcp_servers", + json!({ + "display_name": "github-it", + "url": "https://api.example.com/mcp", + "auth_type": "bearer", + "secret": "tok-it" + }), + ) + .await; +} + #[tokio::test] async fn guardrails_round_trip_through_real_etcd() { let Some(url) = etcd_url() else { @@ -317,6 +338,10 @@ async fn loader_picks_up_every_admin_write() { "endpoint": "https://otel.example.com/v1/traces" }), ), + ( + "/admin/v1/mcp_servers", + json!({"display_name": "loader-mcp", "url": "https://api.example.com/mcp"}), + ), ]; for (uri, body) in writes { let app = build_router(state.clone()); @@ -363,7 +388,7 @@ async fn loader_picks_up_every_admin_write() { likely a subkey-constant drift between EtcdConfigStore::*_SUBKEY \ and the match arms in aisix_etcd::loader: {stats:?}" ); - assert_eq!(stats.accepted, 6, "expected 6 entries; got {stats:?}"); + assert_eq!(stats.accepted, 7, "expected 7 entries; got {stats:?}"); // Each resource table should now have exactly one entry. assert_eq!(snap.models.len(), 1); @@ -372,4 +397,5 @@ async fn loader_picks_up_every_admin_write() { assert_eq!(snap.guardrails.len(), 1); assert_eq!(snap.cache_policies.len(), 1); assert_eq!(snap.observability_exporters.len(), 1); + assert_eq!(snap.mcp_servers.len(), 1); } diff --git a/crates/aisix-core/src/bin/dump-schema.rs b/crates/aisix-core/src/bin/dump-schema.rs index 54f8bb95..cea61f6d 100644 --- a/crates/aisix-core/src/bin/dump-schema.rs +++ b/crates/aisix-core/src/bin/dump-schema.rs @@ -62,6 +62,7 @@ fn main() { "guardrail_attachment", schema::guardrail_attachment_root_schema(), ); + dump_value(&out_dir, "mcp_server", schema::mcp_server_root_schema()); dump::(&out_dir, "ensemble"); dump::(&out_dir, "rate_limit"); diff --git a/crates/aisix-core/src/lib.rs b/crates/aisix-core/src/lib.rs index 4f0dcf76..63b5b155 100644 --- a/crates/aisix-core/src/lib.rs +++ b/crates/aisix-core/src/lib.rs @@ -31,10 +31,11 @@ pub use error::{ AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope, }; pub use models::{ - validate_apikey, validate_cache_policy, validate_guardrail, validate_model, - validate_observability_exporter, validate_provider_key, validate_rate_limit_policy, Adapter, - AisixSnapshot, ApiKey, AppliedGuardrail, CachePolicy, CooldownConfig, ExporterKind, Guardrail, - GuardrailHookPoint, GuardrailKind, KeywordConfig, KeywordPattern, Model, ObservabilityExporter, + validate_apikey, validate_cache_policy, validate_guardrail, validate_mcp_server, + validate_model, validate_observability_exporter, validate_provider_key, + validate_rate_limit_policy, Adapter, AisixSnapshot, ApiKey, AppliedGuardrail, CachePolicy, + CooldownConfig, ExporterKind, Guardrail, GuardrailHookPoint, GuardrailKind, KeywordConfig, + KeywordPattern, McpAuthType, McpServer, McpTransport, Model, ObservabilityExporter, ParamConstraints, PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, RequestOverrides, ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, diff --git a/crates/aisix-core/src/models/mcp_server.rs b/crates/aisix-core/src/models/mcp_server.rs new file mode 100644 index 00000000..cd294fdb --- /dev/null +++ b/crates/aisix-core/src/models/mcp_server.rs @@ -0,0 +1,185 @@ +//! `McpServer` entity โ€” a registered upstream MCP server. +//! +//! Registers an upstream Model Context Protocol (MCP) server so the gateway can +//! front it: its tools are aggregated into the gateway's own MCP endpoint under +//! the namespace `__`, and tool calls are routed back to it. +//! The upstream credential is held by the gateway and is never exposed to the +//! calling client. +//! +//! etcd path: `{prefix}/mcp_servers/{uuid}`. Secondary index on `display_name`. + +use serde::{Deserialize, Serialize}; + +use crate::resource::Resource; + +#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct McpServer { + /// Operator-facing label, unique within the gateway. It is used as the + /// namespace prefix for this server's tools, which are exposed to clients as + /// `__`, so it must not contain the reserved separator + /// `__`. + #[schemars(length(min = 1))] + pub display_name: String, + + /// The upstream server's MCP endpoint URL, reached over the Streamable HTTP + /// transport, such as `https://api.example.com/mcp`. + #[schemars(length(min = 1))] + pub url: String, + + /// Transport used to reach the upstream server. Streamable HTTP is the only + /// supported transport. + #[serde(default)] + pub transport: McpTransport, + + /// How the gateway authenticates to the upstream server. The credential is + /// held by the gateway and is never forwarded from or exposed to the calling + /// client. + #[serde(default)] + pub auth_type: McpAuthType, + + /// Authentication credential for the upstream server. Required when + /// `auth_type` is `bearer`, where it is sent as `Authorization: Bearer + /// ` on every upstream request. Leave unset when `auth_type` is + /// `none`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub secret: Option, + + /// Maximum time, in milliseconds, to wait for a single upstream operation + /// (establishing the session, listing tools, or calling a tool). When + /// omitted, the gateway applies a built-in default. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timeout_ms: Option, + + /// Whether this server is active. When `false`, its tools are not listed and + /// cannot be called. + #[serde(default = "default_enabled")] + pub enabled: bool, + + /// Filled in by the snapshot loader from the etcd key path. + #[serde(skip)] + pub(crate) runtime_id: String, +} + +fn default_enabled() -> bool { + true +} + +/// Transport used to reach an upstream MCP server. +#[derive( + Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, +)] +#[serde(rename_all = "snake_case")] +pub enum McpTransport { + /// Streamable HTTP transport: a single endpoint that serves both POST and + /// GET. + #[default] + StreamableHttp, +} + +/// How the gateway authenticates to an upstream MCP server. +#[derive( + Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, +)] +#[serde(rename_all = "snake_case")] +pub enum McpAuthType { + /// No authentication; the server is reached as-is. + #[default] + None, + /// Bearer token authentication. The token is supplied in `secret` and sent + /// as `Authorization: Bearer `. + Bearer, +} + +impl Resource for McpServer { + fn id(&self) -> &str { + &self.runtime_id + } + + fn name(&self) -> &str { + &self.display_name + } + + fn kind() -> &'static str { + "mcp_servers" + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deserialises_minimal_mcp_server() { + let s: McpServer = serde_json::from_str( + r#"{"display_name":"github","url":"https://api.example.com/mcp"}"#, + ) + .unwrap(); + assert_eq!(s.display_name, "github"); + assert_eq!(s.url, "https://api.example.com/mcp"); + // Defaults. + assert_eq!(s.transport, McpTransport::StreamableHttp); + assert_eq!(s.auth_type, McpAuthType::None); + assert!(s.secret.is_none()); + assert!(s.timeout_ms.is_none()); + assert!(s.enabled); + } + + #[test] + fn deserialises_with_bearer_auth() { + let s: McpServer = serde_json::from_str( + r#"{"display_name":"gh","url":"https://x/mcp","auth_type":"bearer","secret":"tok","timeout_ms":5000,"enabled":false}"#, + ) + .unwrap(); + assert_eq!(s.auth_type, McpAuthType::Bearer); + assert_eq!(s.secret.as_deref(), Some("tok")); + assert_eq!(s.timeout_ms, Some(5000)); + assert!(!s.enabled); + } + + #[test] + fn rejects_unknown_fields() { + let r: Result = + serde_json::from_str(r#"{"display_name":"x","url":"u","extra":1}"#); + assert!(r.is_err()); + } + + #[test] + fn rejects_unknown_transport_and_auth_type() { + assert!(serde_json::from_str::( + r#"{"display_name":"x","url":"u","transport":"stdio"}"# + ) + .is_err()); + assert!(serde_json::from_str::( + r#"{"display_name":"x","url":"u","auth_type":"oauth"}"# + ) + .is_err()); + } + + #[test] + fn resource_trait_routes_through_display_name() { + let mut s: McpServer = + serde_json::from_str(r#"{"display_name":"github","url":"https://x/mcp"}"#).unwrap(); + s.runtime_id = "uuid-mcp-1".into(); + assert_eq!(::kind(), "mcp_servers"); + assert_eq!(s.id(), "uuid-mcp-1"); + assert_eq!(s.name(), "github"); + } + + #[test] + fn round_trip_omits_default_optionals() { + let original = McpServer { + display_name: "github".into(), + url: "https://x/mcp".into(), + transport: McpTransport::StreamableHttp, + auth_type: McpAuthType::None, + secret: None, + timeout_ms: None, + enabled: true, + runtime_id: String::new(), + }; + let s = serde_json::to_string(&original).unwrap(); + let back: McpServer = serde_json::from_str(&s).unwrap(); + assert_eq!(original, back); + } +} diff --git a/crates/aisix-core/src/models/mod.rs b/crates/aisix-core/src/models/mod.rs index 6b0cc6c4..acc7a47f 100644 --- a/crates/aisix-core/src/models/mod.rs +++ b/crates/aisix-core/src/models/mod.rs @@ -20,6 +20,7 @@ pub mod cache_policy; pub mod embedding; pub mod ensemble; pub mod guardrail; +pub mod mcp_server; pub mod model; pub mod observability_exporter; pub mod provider_key; @@ -40,6 +41,7 @@ pub use guardrail::{ BedrockLatencyMode, Guardrail, GuardrailAttachment, GuardrailHookPoint, GuardrailKind, GuardrailScopeType, KeywordConfig, KeywordPattern, }; +pub use mcp_server::{McpAuthType, McpServer, McpTransport}; pub use model::{ Adapter, BackgroundModelCheck, CooldownConfig, Model, DEFAULT_COOLDOWN_TRIGGER_STATUSES, }; @@ -56,7 +58,7 @@ pub use rate_limit_policy::{PolicyScope, PolicyWindow, RateLimitPolicy}; pub use routing::{Routing, RoutingStrategy, RoutingTarget, WhenAllUnavailablePolicy}; pub use schema::{ validate_apikey, validate_cache_policy, validate_guardrail, validate_guardrail_attachment, - validate_model, validate_observability_exporter, validate_provider_key, + validate_mcp_server, validate_model, validate_observability_exporter, validate_provider_key, validate_rate_limit_policy, SchemaError, }; pub use semantic::{ diff --git a/crates/aisix-core/src/models/schema.rs b/crates/aisix-core/src/models/schema.rs index eea1f7c1..1d5fda37 100644 --- a/crates/aisix-core/src/models/schema.rs +++ b/crates/aisix-core/src/models/schema.rs @@ -30,6 +30,7 @@ pub struct Schemas { pub cache_policy: Validator, pub observability_exporter: Validator, pub rate_limit_policy: Validator, + pub mcp_server: Validator, } pub static SCHEMAS: Lazy> = Lazy::new(|| Arc::new(Schemas::compile())); @@ -61,6 +62,9 @@ impl Schemas { rate_limit_policy: jsonschema::options() .build(&rate_limit_policy_root_schema()) .expect("rate_limit_policy schema is well-formed"), + mcp_server: jsonschema::options() + .build(&mcp_server_root_schema()) + .expect("mcp_server schema is well-formed"), } } } @@ -117,6 +121,10 @@ pub fn validate_guardrail_attachment(value: &Value) -> Result<(), SchemaError> { validate(&SCHEMAS.guardrail_attachment, value) } +pub fn validate_mcp_server(value: &Value) -> Result<(), SchemaError> { + validate(&SCHEMAS.mcp_server, value) +} + /// Build a resource's canonical JSON Schema from its struct via `schemars`, /// the single source of field shapes and per-field constraints. /// @@ -173,6 +181,17 @@ pub fn provider_key_root_schema() -> Value { struct_root_schema::(true) } +/// Canonical JSON Schema for the `mcp_server` resource, derived from the +/// [`McpServer`](crate::models::McpServer) struct. Uses the nullable `Option` +/// representation (`true`) so the optional `secret` / `timeout_ms` fields accept +/// an explicit `null` as well as being absent, matching the resource's wire +/// contract. The `transport` / `auth_type` closed sets come from the +/// [`McpTransport`](crate::models::McpTransport) / +/// [`McpAuthType`](crate::models::McpAuthType) enums. +pub fn mcp_server_root_schema() -> Value { + struct_root_schema::(true) +} + /// Canonical JSON Schema for the `guardrail` resource, derived from the /// [`Guardrail`](crate::models::Guardrail) struct. `schemars` renders the /// internally-tagged `GuardrailKind` as a native top-level `oneOf`; the diff --git a/crates/aisix-core/src/models/snapshot.rs b/crates/aisix-core/src/models/snapshot.rs index 692bc22f..b5083e15 100644 --- a/crates/aisix-core/src/models/snapshot.rs +++ b/crates/aisix-core/src/models/snapshot.rs @@ -7,6 +7,7 @@ use super::apikey::ApiKey; use super::cache_policy::CachePolicy; use super::guardrail::{Guardrail, GuardrailAttachment}; +use super::mcp_server::McpServer; use super::model::Model; use super::observability_exporter::ObservabilityExporter; use super::provider_key::ProviderKey; @@ -34,6 +35,10 @@ pub struct AisixSnapshot { /// fan-out POST per chat completion (see `aisix-obs::OtlpHttpFanOut`). pub observability_exporters: ResourceTable, pub rate_limit_policies: ResourceTable, + /// Registered upstream MCP servers: `/aisix//mcp_servers/`. The + /// MCP gateway endpoint aggregates each enabled server's tools and routes + /// tool calls back to the owning server. + pub mcp_servers: ResourceTable, } impl AisixSnapshot { @@ -52,6 +57,7 @@ impl AisixSnapshot { + self.cache_policies.len() + self.observability_exporters.len() + self.rate_limit_policies.len() + + self.mcp_servers.len() } } diff --git a/crates/aisix-etcd/src/loader.rs b/crates/aisix-etcd/src/loader.rs index b1c0d938..a9c4fae3 100644 --- a/crates/aisix-etcd/src/loader.rs +++ b/crates/aisix-etcd/src/loader.rs @@ -13,9 +13,9 @@ use aisix_core::models::{ validate_apikey, validate_cache_policy, validate_guardrail, validate_guardrail_attachment, - validate_model, validate_observability_exporter, validate_provider_key, - validate_rate_limit_policy, ApiKey, CachePolicy, Guardrail, GuardrailAttachment, Model, - ObservabilityExporter, ProviderKey, RateLimitPolicy, SchemaError, + validate_mcp_server, validate_model, validate_observability_exporter, validate_provider_key, + validate_rate_limit_policy, ApiKey, CachePolicy, Guardrail, GuardrailAttachment, McpServer, + Model, ObservabilityExporter, ProviderKey, RateLimitPolicy, SchemaError, }; use aisix_core::resource::ResourceEntry; use aisix_core::AisixSnapshot; @@ -244,6 +244,18 @@ pub fn build_snapshot(prefix: &str, entries: &[RawEntry]) -> (AisixSnapshot, Bui snapshot.rate_limit_policies.insert(entry); } } + "mcp_servers" => { + if let Some(entry) = validate_and_parse::( + &raw.key, + raw.revision, + parsed, + &value, + validate_mcp_server, + &mut stats, + ) { + snapshot.mcp_servers.insert(entry); + } + } other => { tracing::debug!(key = %raw.key, kind = %other, "unknown etcd kind; skipping"); stats.unknown_kind += 1; diff --git a/crates/aisix-etcd/src/supervisor.rs b/crates/aisix-etcd/src/supervisor.rs index 4f4eb735..baeeea68 100644 --- a/crates/aisix-etcd/src/supervisor.rs +++ b/crates/aisix-etcd/src/supervisor.rs @@ -376,6 +376,9 @@ impl Supervisor

{ for e in tiny.rate_limit_policies.entries() { new.rate_limit_policies.insert(clone_entry(&e)); } + for e in tiny.mcp_servers.entries() { + new.mcp_servers.insert(clone_entry(&e)); + } new }); self.remove_rejection_for_key(&entry.key); @@ -430,6 +433,7 @@ impl Supervisor

{ snap.observability_exporters.get_by_id(parsed.id).is_some() } "rate_limit_policies" => snap.rate_limit_policies.get_by_id(parsed.id).is_some(), + "mcp_servers" => snap.mcp_servers.get_by_id(parsed.id).is_some(), _ => false, }; let removed_rejection = self.remove_rejection_for_key(key_str); @@ -475,6 +479,9 @@ impl Supervisor

{ "rate_limit_policies" => { new.rate_limit_policies.remove(parsed.id); } + "mcp_servers" => { + new.mcp_servers.remove(parsed.id); + } _ => {} } new @@ -708,6 +715,9 @@ fn clone_snapshot(src: &AisixSnapshot) -> AisixSnapshot { for e in src.rate_limit_policies.entries() { out.rate_limit_policies.insert(clone_entry(&e)); } + for e in src.mcp_servers.entries() { + out.mcp_servers.insert(clone_entry(&e)); + } out } diff --git a/schemas/resources/mcp_server.schema.json b/schemas/resources/mcp_server.schema.json new file mode 100644 index 00000000..eb128ed7 --- /dev/null +++ b/schemas/resources/mcp_server.schema.json @@ -0,0 +1,94 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "additionalProperties": false, + "definitions": { + "McpAuthType": { + "description": "How the gateway authenticates to an upstream MCP server.", + "oneOf": [ + { + "description": "No authentication; the server is reached as-is.", + "enum": [ + "none" + ], + "type": "string" + }, + { + "description": "Bearer token authentication. The token is supplied in `secret` and sent as `Authorization: Bearer `.", + "enum": [ + "bearer" + ], + "type": "string" + } + ] + }, + "McpTransport": { + "description": "Transport used to reach an upstream MCP server.", + "oneOf": [ + { + "description": "Streamable HTTP transport: a single endpoint that serves both POST and GET.", + "enum": [ + "streamable_http" + ], + "type": "string" + } + ] + } + }, + "properties": { + "auth_type": { + "allOf": [ + { + "$ref": "#/definitions/McpAuthType" + } + ], + "default": "none", + "description": "How the gateway authenticates to the upstream server. The credential is held by the gateway and is never forwarded from or exposed to the calling client." + }, + "display_name": { + "description": "Operator-facing label, unique within the gateway. It is used as the namespace prefix for this server's tools, which are exposed to clients as `__`, so it must not contain the reserved separator `__`.", + "minLength": 1, + "type": "string" + }, + "enabled": { + "default": true, + "description": "Whether this server is active. When `false`, its tools are not listed and cannot be called.", + "type": "boolean" + }, + "secret": { + "description": "Authentication credential for the upstream server. Required when `auth_type` is `bearer`, where it is sent as `Authorization: Bearer ` on every upstream request. Leave unset when `auth_type` is `none`.", + "type": [ + "string", + "null" + ] + }, + "timeout_ms": { + "description": "Maximum time, in milliseconds, to wait for a single upstream operation (establishing the session, listing tools, or calling a tool). When omitted, the gateway applies a built-in default.", + "format": "uint64", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "transport": { + "allOf": [ + { + "$ref": "#/definitions/McpTransport" + } + ], + "default": "streamable_http", + "description": "Transport used to reach the upstream server. Streamable HTTP is the only supported transport." + }, + "url": { + "description": "The upstream server's MCP endpoint URL, reached over the Streamable HTTP transport, such as `https://api.example.com/mcp`.", + "minLength": 1, + "type": "string" + } + }, + "required": [ + "display_name", + "url" + ], + "title": "McpServer", + "type": "object" +} From cb81cab7ea9a5ae31d4d0a25ab755b01f201ca70 Mon Sep 17 00:00:00 2001 From: Ming Wen Date: Mon, 29 Jun 2026 16:07:52 +0800 Subject: [PATCH 2/2] fix(mcp): enforce bearer secret + test McpServer decode guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit (CLAUDE.md ยง8) on #664 returned APPROVE with two LOW suggestions, both folded in: - LOW: `auth_type=bearer` with no/empty `secret` was accepted and would send an empty `Authorization: Bearer ` upstream. `decode()` now rejects it (400), alongside the existing reserved-`__` check. - LOW: the `__`-rejection (net-new logic) had no coverage. Added unit tests for the decode guards: rejects `__` in display_name, rejects bearer without secret, accepts a valid server. Refs AISIX-Cloud#894 --- .../aisix-admin/src/mcp_servers_handlers.rs | 46 ++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/crates/aisix-admin/src/mcp_servers_handlers.rs b/crates/aisix-admin/src/mcp_servers_handlers.rs index 1dda5cab..23a5e725 100644 --- a/crates/aisix-admin/src/mcp_servers_handlers.rs +++ b/crates/aisix-admin/src/mcp_servers_handlers.rs @@ -7,7 +7,7 @@ use aisix_core::models::validate_mcp_server; use aisix_core::resource::ResourceEntry; -use aisix_core::McpServer; +use aisix_core::{McpAuthType, McpServer}; use axum::extract::{Path, State}; use axum::Json; use serde_json::Value; @@ -102,6 +102,13 @@ fn decode(raw: &Value) -> Result { "display_name must not contain the reserved separator `{TOOL_NAMESPACE_SEPARATOR}`" ))); } + if matches!(server.auth_type, McpAuthType::Bearer) + && server.secret.as_deref().unwrap_or_default().is_empty() + { + return Err(AdminError::BadRequest( + "secret is required and must be non-empty when auth_type is `bearer`".to_string(), + )); + } Ok(server) } @@ -117,3 +124,40 @@ fn assert_unique_display_name( } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn decode_rejects_separator_in_display_name() { + let err = decode(&json!({"display_name": "a__b", "url": "https://x/mcp"})) + .expect_err("`__` in display_name must be rejected"); + assert!(matches!(err, AdminError::BadRequest(_))); + } + + #[test] + fn decode_rejects_bearer_without_secret() { + let err = decode(&json!({ + "display_name": "gh", + "url": "https://x/mcp", + "auth_type": "bearer" + })) + .expect_err("bearer auth without a secret must be rejected"); + assert!(matches!(err, AdminError::BadRequest(_))); + } + + #[test] + fn decode_accepts_valid_server() { + let server = decode(&json!({ + "display_name": "github", + "url": "https://api.example.com/mcp", + "auth_type": "bearer", + "secret": "tok" + })) + .expect("valid server should decode"); + assert_eq!(server.display_name, "github"); + assert_eq!(server.secret.as_deref(), Some("tok")); + } +}