From ff34d5b6eb3674d51e00a107872882dec088d4b5 Mon Sep 17 00:00:00 2001 From: Ragnor Comerford Date: Tue, 13 Oct 2026 21:43:55 -0700 Subject: [PATCH 1/2] feat(server): schema and ad-hoc read tools on /mcp, metadata at the advertised path --- .../mcp-schema-read-discovery.added.md | 15 + crates/omnigraph-server/src/handlers.rs | 1 + crates/omnigraph-server/src/http_contract.rs | 4 +- crates/omnigraph-server/src/lib.rs | 38 +- crates/omnigraph-server/src/main.rs | 15 +- crates/omnigraph-server/src/mcp.rs | 439 ++++++++++++++++-- crates/omnigraph-server/src/settings.rs | 1 + .../omnigraph-server/tests/integration/mcp.rs | 282 ++++++++++- .../tests/integration/openapi.rs | 12 +- ...identity-credentials-and-applied-policy.md | 6 +- ...026-09-26-self-contained-server-testing.md | 4 +- docs/user/index.md | 1 + docs/user/operations/mcp.md | 110 +++++ docs/user/operations/server.md | 32 +- openapi.json | 33 ++ 15 files changed, 913 insertions(+), 80 deletions(-) create mode 100644 changelog.d/mcp-schema-read-discovery.added.md create mode 100644 docs/user/operations/mcp.md diff --git a/changelog.d/mcp-schema-read-discovery.added.md b/changelog.d/mcp-schema-read-discovery.added.md new file mode 100644 index 000000000..73f8012b2 --- /dev/null +++ b/changelog.d/mcp-schema-read-discovery.added.md @@ -0,0 +1,15 @@ +- The server's `/mcp` endpoint gains a `schema` tool, which returns a graph's + `.pg` schema under the `read` action, and an opt-in `read` tool, which runs an + ad-hoc GQ read through `POST /query`, refuses mutations, and is listed only + when the server starts with `--mcp-allow-adhoc` (or + `OMNIGRAPH_MCP_ALLOW_ADHOC=1`). Tools now carry titles, fuller descriptions + and, where the result shape is fixed, an output schema, and `initialize` + returns instructions that tell a model to read the schema first. See + [MCP][mcp-schema-read-guide]. +- Protected-resource metadata is also served at the path the `/mcp` challenge + advertises (`/.well-known/oauth-protected-resource/`), so an + OAuth client such as Claude connects to a resource identifier with a path + without a proxy rewrite, and a refused bearer token now gets + `error="invalid_token"` in that challenge. + +[mcp-schema-read-guide]: ../docs/user/operations/mcp.md diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs index d76be84f3..2c46243a6 100644 --- a/crates/omnigraph-server/src/handlers.rs +++ b/crates/omnigraph-server/src/handlers.rs @@ -312,6 +312,7 @@ const ALWAYS_FLAT_PATHS: &[&str] = &[ "/cluster/deployments", "/cluster/deployments/{id}", "/.well-known/oauth-protected-resource", + "/.well-known/oauth-protected-resource/{resource_path}", ]; /// In multi-mode `server_openapi`, every protected path-item is diff --git a/crates/omnigraph-server/src/http_contract.rs b/crates/omnigraph-server/src/http_contract.rs index d36019563..fa486f833 100644 --- a/crates/omnigraph-server/src/http_contract.rs +++ b/crates/omnigraph-server/src/http_contract.rs @@ -37,7 +37,7 @@ pub(crate) async fn require_contract(request: Request, next: Next) -> Result Response { let path = request.uri().path(); - if matches!(path, "/mcp" | "/.well-known/oauth-protected-resource") { + if path == "/mcp" || crate::mcp::is_resource_metadata_path(path) { return next.run(request).await; } let discovery = path == "/healthz"; @@ -60,7 +60,7 @@ pub(crate) fn describe_contract(doc: &mut utoipa::openapi::OpenApi) { schema.enum_values = Some(vec![HTTP_API_CONTRACT.into()]); for (path, item) in &mut doc.paths.paths { // OAuth resource metadata follows its standard protocol, as does MCP. - if path == "/.well-known/oauth-protected-resource" { + if crate::mcp::is_resource_metadata_path(path) { continue; } let protected = diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index 406b22cc6..dfc583823 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -31,6 +31,7 @@ pub mod workload; pub use graph_id::GraphId; pub use identity::{AuthSource, AuthenticatedActor, GraphKey, ResolvedActor, Scope, TenantId}; +pub use mcp::McpSettings; pub use registry::{ BlockedGraph, GraphEntry, GraphHandle, GraphRegistry, InsertError, LoadingGraph, RegistryLookup, RegistrySnapshot, StartupFailure, @@ -112,6 +113,7 @@ fn hash_bearer_token(token: &str) -> BearerTokenHash { ), paths( mcp::resource_metadata, + mcp::resource_metadata_at_resource_path, deployment::status, deployment::lookup, deployment::apply, @@ -237,6 +239,7 @@ pub struct ManagedServerConfig { canonical_root: String, trust: Option, oidc_trust: Option>, + mcp: McpSettings, } impl ManagedServerConfig { @@ -253,6 +256,12 @@ impl ManagedServerConfig { self.config.shutdown_grace = grace; self } + + /// Set the MCP boot settings, preserving the validated root binding. + pub fn with_mcp_settings(mut self, mcp: McpSettings) -> Self { + self.mcp = mcp; + self + } } /// The default bound on graceful shutdown. @@ -398,6 +407,8 @@ pub struct AppState { bearer_tokens: Arc<[(BearerTokenHash, Arc)]>, data_token_trust: Option>, oidc_identity_trust: Option>, + /// MCP boot settings: which opt-in tools `/mcp` lists and admits. + mcp: Arc, /// Bounded process-wide ownership for queued served-export bytes. The /// response body and detached producer jointly retain each reservation. export_transport: export_transport::ExportTransport, @@ -767,6 +778,7 @@ impl AppState { bearer_tokens, data_token_trust: None, oidc_identity_trust: None, + mcp: Arc::new(McpSettings::default()), operations: operations::OperationRuntime::new(), export_transport: export_transport::ExportTransport::with_defaults(), witness: Arc::new(BootWitness::default()), @@ -820,6 +832,7 @@ impl AppState { bearer_tokens, data_token_trust: None, oidc_identity_trust: None, + mcp: Arc::new(McpSettings::default()), operations: operations::OperationRuntime::new(), export_transport: export_transport::ExportTransport::with_defaults(), witness: Arc::new(BootWitness::default()), @@ -911,6 +924,17 @@ impl AppState { self } + /// Attach the MCP boot settings `/mcp` reads for every request. + #[must_use] + pub fn with_mcp_settings(mut self, mcp: McpSettings) -> Self { + self.mcp = Arc::new(mcp); + self + } + + pub(crate) fn mcp_settings(&self) -> &McpSettings { + &self.mcp + } + pub(crate) fn oidc_resource_metadata(&self) -> Option { self.oidc_identity_trust .as_ref() @@ -2736,19 +2760,26 @@ pub fn build_app(state: AppState) -> Router { } pub async fn serve(config: ServerConfig) -> Result<()> { - serve_config(config, None, None).await + serve_config(config, None, None, McpSettings::default()).await +} + +/// Serve with explicit MCP boot settings (`--mcp-*` flags and their +/// environment variables, resolved by [`McpSettings::resolve`]). +pub async fn serve_with_mcp_settings(config: ServerConfig, mcp: McpSettings) -> Result<()> { + serve_config(config, None, None, mcp).await } /// Serve settings whose offline data-token trust was validated against their /// applied snapshot's canonical root before any graph engine open. pub async fn serve_with_data_token_trust(config: ManagedServerConfig) -> Result<()> { - serve_config(config.config, config.trust, config.oidc_trust).await + serve_config(config.config, config.trust, config.oidc_trust, config.mcp).await } async fn serve_config( config: ServerConfig, data_token_trust: Option, oidc_identity_trust: Option>, + mcp: McpSettings, ) -> Result<()> { // RFC 0049: the signal listener is installed before anything else, so // the shutdown bound covers startup. On the signal it sets `draining`, @@ -2866,7 +2897,8 @@ async fn serve_config( Arc::clone(&draining), shutdown_grace, ) - .with_process_defaults(process_defaults); + .with_process_defaults(process_defaults) + .with_mcp_settings(mcp); deployment::initialize_boot_activation(&state); let retained_admission = state.cluster_admission.clone(); let startup_owner = operations diff --git a/crates/omnigraph-server/src/main.rs b/crates/omnigraph-server/src/main.rs index e328963f4..0ef31cfac 100644 --- a/crates/omnigraph-server/src/main.rs +++ b/crates/omnigraph-server/src/main.rs @@ -3,8 +3,8 @@ use std::path::PathBuf; use clap::Parser; use color_eyre::eyre::Result; use omnigraph_server::{ - init_tracing, load_server_settings, load_server_settings_with_identity_trust, - resolve_shutdown_grace, serve, serve_with_data_token_trust, + McpSettings, init_tracing, load_server_settings, load_server_settings_with_identity_trust, + resolve_shutdown_grace, serve_with_data_token_trust, serve_with_mcp_settings, }; #[derive(Debug, Parser)] @@ -50,6 +50,11 @@ struct Cli { /// termination grace must be longer. #[arg(long)] shutdown_grace_seconds: Option, + /// List and admit the MCP `read` tool, which runs ad-hoc GQ reads under the + /// caller's applied `read` policy. Off by default. Equivalent to setting + /// `OMNIGRAPH_MCP_ALLOW_ADHOC=1`. + #[arg(long)] + mcp_allow_adhoc: bool, } #[tokio::main] @@ -58,6 +63,7 @@ async fn main() -> Result<()> { init_tracing(); let cli = Cli::parse(); + let mcp = McpSettings::resolve(cli.mcp_allow_adhoc); match (cli.data_token_trust, cli.oidc_identity_trust) { (data, oidc) if data.is_some() || oidc.is_some() => { let settings = load_server_settings_with_identity_trust( @@ -69,7 +75,8 @@ async fn main() -> Result<()> { oidc.as_deref(), ) .await? - .with_shutdown_grace(resolve_shutdown_grace(cli.shutdown_grace_seconds)?); + .with_shutdown_grace(resolve_shutdown_grace(cli.shutdown_grace_seconds)?) + .with_mcp_settings(mcp); serve_with_data_token_trust(settings).await } _ => { @@ -81,7 +88,7 @@ async fn main() -> Result<()> { ) .await?; settings.shutdown_grace = resolve_shutdown_grace(cli.shutdown_grace_seconds)?; - serve(settings).await + serve_with_mcp_settings(settings, mcp).await } } } diff --git a/crates/omnigraph-server/src/mcp.rs b/crates/omnigraph-server/src/mcp.rs index b6d303d86..11a6e871f 100644 --- a/crates/omnigraph-server/src/mcp.rs +++ b/crates/omnigraph-server/src/mcp.rs @@ -1,4 +1,5 @@ -//! Read-only MCP transport over the existing graph discovery and stored-query handlers. +//! Read-only MCP transport over the existing graph discovery, schema, stored-query +//! and inline-query handlers. //! The official SDK owns protocol framing, negotiation and cancellation. use std::future::Future; use std::sync::Arc; @@ -6,7 +7,7 @@ use std::time::Duration; use axum::body::{Bytes, to_bytes}; use axum::extract::{Path, Request, State}; -use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; +use axum::http::{HeaderMap, HeaderValue, StatusCode, Uri, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; use axum::routing::get; @@ -33,6 +34,61 @@ use crate::{ApiError, AppState, AuthenticatedActor, GraphId, GraphKey}; pub(crate) const REQUEST_BYTES: usize = 64 * 1024; const RESULT_BYTES: usize = 1024 * 1024; const READ_DEADLINE: Duration = Duration::from_secs(30); +const RESOURCE_METADATA_PATH: &str = "/.well-known/oauth-protected-resource"; + +/// Environment equivalent of `--mcp-allow-adhoc`. +pub const MCP_ALLOW_ADHOC_ENV: &str = "OMNIGRAPH_MCP_ALLOW_ADHOC"; + +/// Process boot settings for `/mcp`. They select which opt-in tools are listed +/// and admitted; they never grant anything, since every tool still runs only +/// what the applied policy allows its caller. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[non_exhaustive] +pub struct McpSettings { + /// Lists and admits the ad-hoc `read` tool. + pub allow_adhoc: bool, +} + +impl McpSettings { + /// The boot flags, each enabled also by its environment variable. + pub fn resolve(allow_adhoc: bool) -> Self { + Self { + allow_adhoc: allow_adhoc || env_flag(MCP_ALLOW_ADHOC_ENV), + } + } +} + +fn env_flag(name: &str) -> bool { + std::env::var(name).is_ok_and(|value| { + let value = value.trim(); + !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false") + }) +} + +/// Whether `path` is protected-resource metadata: the root document or the +/// path-suffixed location the challenge advertises. +pub(crate) fn is_resource_metadata_path(path: &str) -> bool { + path.strip_prefix(RESOURCE_METADATA_PATH) + .is_some_and(|rest| rest.is_empty() || rest.starts_with('/')) +} + +/// The metadata URL a client derives from the resource identifier (RFC 9728 +/// §3.1): the well-known prefix inserted before the resource's path. +fn advertised_metadata_url(resource: &str) -> Option { + let mut url = url::Url::parse(resource).ok()?; + let path = format!( + "{RESOURCE_METADATA_PATH}{}", + url.path().trim_end_matches('/') + ); + url.set_path(&path); + url.set_query(None); + url.set_fragment(None); + Some(url) +} + +/// The metadata path this server advertises, fixed by its boot binding. +#[derive(Clone)] +pub(crate) struct AdvertisedMetadataPath(Arc); /// Mount only when an explicitly configured OIDC resource enables this surface. /// Every HTTP request authenticates again; no MCP session retains an actor. @@ -75,13 +131,24 @@ pub(crate) fn router(state: AppState) -> Router { state.clone(), handlers::require_bearer_auth, )) - .route_layer(middleware::from_fn_with_state(state, resource_challenge)); - Router::new() - .route( - "/.well-known/oauth-protected-resource", - get(resource_metadata), - ) - .merge(protected) + .route_layer(middleware::from_fn_with_state( + state.clone(), + resource_challenge, + )); + // Computed once from the fixed binding. A wildcard route compared with the + // raw request path keeps the configured path out of the router's syntax. + let advertised = state + .oidc_resource_url() + .and_then(|resource| advertised_metadata_url(&resource)) + .map(|url| AdvertisedMetadataPath(Arc::from(url.path()))); + let mut discovery = Router::new().route(RESOURCE_METADATA_PATH, get(resource_metadata)); + if let Some(advertised) = advertised { + discovery = discovery.route( + &format!("{RESOURCE_METADATA_PATH}/{{*resource_path}}"), + get(resource_metadata_at_resource_path).layer(Extension(advertised)), + ); + } + discovery.merge(protected) } #[utoipa::path(get, path = "/.well-known/oauth-protected-resource", tag = "management", @@ -95,25 +162,57 @@ pub(crate) async fn resource_metadata(State(state): State) -> Response } } +/// The same metadata at the location the `/mcp` challenge advertises for a +/// resource identifier with a path: the well-known prefix followed by that +/// path (`/.well-known/oauth-protected-resource/mcp` for `https://host/mcp`). +/// Every other suffix is 404. +#[utoipa::path(get, path = "/.well-known/oauth-protected-resource/{resource_path}", tag = "management", + operation_id = "oauthProtectedResourceMetadataAtResourcePath", + params(("resource_path" = String, Path, description = "The configured resource identifier's path, without its leading `/`")), + responses((status = 200, description = "Public OAuth protected resource metadata", body = Value), + (status = 404, description = "Not the configured resource's path, or OAuth resource identity is not configured")))] +pub(crate) async fn resource_metadata_at_resource_path( + State(state): State, + Extension(AdvertisedMetadataPath(advertised)): Extension, + uri: Uri, +) -> Response { + if uri.path() == &*advertised { + resource_metadata(State(state)).await + } else { + StatusCode::NOT_FOUND.into_response() + } +} + async fn resource_challenge( State(state): State, request: Request, next: Next, ) -> Response { + // RFC 6750 §3.1: `invalid_token` answers a presented bearer credential + // that was refused; a request without one gets no error code. + let presented = request + .headers() + .get_all(header::AUTHORIZATION) + .iter() + .any(|value| { + value + .as_bytes() + .get(..7) + .is_some_and(|scheme| scheme.eq_ignore_ascii_case(b"bearer ")) + }); let mut response = next.run(request).await; if response.status() == StatusCode::UNAUTHORIZED { - if let Some(mut url) = state + if let Some(url) = state .oidc_resource_url() - .and_then(|url| url::Url::parse(&url).ok()) + .and_then(|resource| advertised_metadata_url(&resource)) { - let path = format!( - "/.well-known/oauth-protected-resource{}", - url.path().trim_end_matches('/') - ); - url.set_path(&path); - url.set_query(None); - url.set_fragment(None); - if let Ok(value) = HeaderValue::from_str(&format!("Bearer resource_metadata=\"{url}\"")) + let error = if presented { + "error=\"invalid_token\", " + } else { + "" + }; + if let Ok(value) = + HeaderValue::from_str(&format!("Bearer {error}resource_metadata=\"{url}\"")) { response .headers_mut() @@ -151,12 +250,29 @@ struct QueryArguments { #[serde(default)] branch: Option, } +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ReadArguments { + graph: String, + query: String, + #[serde(default)] + name: Option, + #[serde(default)] + params: Option>, + #[serde(default)] + branch: Option, + #[serde(default)] + snapshot: Option, +} impl ServerHandler for GraphTools { fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) - .with_server_info(Implementation::new("omnigraph-server", env!("CARGO_PKG_VERSION"))) - .with_instructions("Discover graph names, list readable stored queries, then invoke a named read. Applied graph policy governs every read. Mutation tools are unavailable.") + .with_server_info(Implementation::new( + "omnigraph-server", + env!("CARGO_PKG_VERSION"), + )) + .with_instructions(instructions(self.state.mcp_settings())) } async fn list_tools( @@ -164,11 +280,15 @@ impl ServerHandler for GraphTools { _: Option, _: RequestContext, ) -> Result { - Ok(ListToolsResult::with_all_items(tools())) + Ok(ListToolsResult::with_all_items(tools( + self.state.mcp_settings(), + ))) } fn get_tool(&self, name: &str) -> Option { - tools().into_iter().find(|tool| tool.name == name) + tools(self.state.mcp_settings()) + .into_iter() + .find(|tool| tool.name == name) } async fn call_tool( @@ -312,6 +432,35 @@ impl GraphTools { .0; bounded_result(&output) } + "schema" => { + let args: GraphArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_schema_get(Extension(handle), Some(Extension(actor))) + .await? + .0; + bounded_result(&output) + } + "read" if self.state.mcp_settings().allow_adhoc => { + let args: ReadArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let request: crate::api::QueryRequest = serde_json::from_value(json!({ + "query": args.query, + "name": args.name, + "params": args.params, + "branch": args.branch, + "snapshot": args.snapshot, + })) + .map_err(|_| ApiError::bad_request("invalid MCP tool arguments"))?; + let output = handlers::server_query( + State(self.state.clone()), + Extension(handle), + Some(Extension(actor)), + Ok(Json(request)), + ) + .await? + .0; + bounded_result(&output) + } _ => Err(ApiError::bad_request("unknown MCP tool")), } } @@ -339,14 +488,242 @@ fn arguments_as(value: Value) -> Result Vec { - let text = json!({"type":"string","minLength":1}); - [ - ("graphs", "List effective graph identifiers and display names.", json!({"type":"object","properties":{},"additionalProperties":false})), - ("queries", "List stored reads visible under the graph's applied read policy. Invocation is separately authorized.", json!({"type":"object","properties":{"graph":text},"required":["graph"],"additionalProperties":false})), - ("query", "Invoke a named stored read under the graph's applied policy. Defaults to the main branch; mutations are refused.", json!({"type":"object","properties":{"graph":text,"name":text,"params":{"type":"object"},"branch":text},"required":["graph","name"],"additionalProperties":false})), - ].into_iter().map(|(name, description, schema)| Tool::new(name, description, schema.as_object().expect("literal object").clone()) - .with_annotations(ToolAnnotations::new().read_only(true))).collect() +/// Which boot setting, if any, lists and admits a tool. +#[derive(Clone, Copy)] +enum Gate { + Always, + Adhoc, +} + +impl Gate { + fn open(self, settings: &McpSettings) -> bool { + match self { + Gate::Always => true, + Gate::Adhoc => settings.allow_adhoc, + } + } +} + +struct ToolSpec { + name: &'static str, + title: &'static str, + description: &'static str, + input: fn() -> Value, + /// Success shape, for tools whose result is fixed. The declared output + /// schema also admits the error envelope: clients validate any structured + /// content against it, including an `isError` result's. + output: Option Value>, + gate: Gate, +} + +const TOOLS: &[ToolSpec] = &[ + ToolSpec { + name: "graphs", + title: "List graphs", + description: "List the graphs this server serves: each graph's `graph_id` and display name. Every other tool takes one of these ids as `graph`. Listing a graph does not mean your credential may read it.", + input: no_arguments_schema, + output: Some(graphs_output_schema), + gate: Gate::Always, + }, + ToolSpec { + name: "schema", + title: "Read a graph's schema", + description: "Return a graph's schema as `.pg` source (`schema_source`): its node and edge types, properties with their types, keys, edge endpoints and declared embeddings. Call this before writing any query: GQ names types, properties and edges exactly as declared here, and edges are matched in lowerCamelCase.", + input: graph_schema, + output: Some(schema_output_schema), + gate: Gate::Always, + }, + ToolSpec { + name: "queries", + title: "List stored reads", + description: "List a graph's stored reads: each `name`, its description and instruction, and typed `params` (`name`, `kind`, `nullable`; `item_kind` for lists, `vector_dim` for vectors). A non-nullable parameter is required. Run one with `query`. Invocation is authorized separately, so a listed read can still be refused.", + input: graph_schema, + output: Some(queries_output_schema), + gate: Gate::Always, + }, + ToolSpec { + name: "query", + title: "Run a stored read", + description: "Run a stored read by `name` with `params` keyed by its declared parameter names (see `queries`). Reads `branch`, `main` when omitted. Returns `rows`, `row_count` and the read's `graph_commit_id`. Stored mutations are refused.", + input: query_schema, + output: None, + gate: Gate::Always, + }, + ToolSpec { + name: "read", + title: "Run a GQ read", + description: "Run a GQ read query you write. Call `schema` first. Declare every value as a typed parameter (`query people($min: I32) { match { $p: Person $p.age >= $min } return { $p.name } }`) and pass the values in `params`; pick one declared query with `name` when the source declares several. Reads `branch` (default `main`) or a `snapshot`, never both. Search operators (`nearest`, `bm25`, `rrf`) need a trailing `limit`. Also answers `branch list`, `branch diff [into ]`, `branch merge-base …`, `branch merge … dry run` and `explain query …`, sent without `branch`, `snapshot`, `name` or `params`. Mutations and branch writes are refused. Returns `rows`, `row_count` and `graph_commit_id`.", + input: read_schema, + output: None, + gate: Gate::Adhoc, + }, +]; + +fn instructions(settings: &McpSettings) -> String { + let mut text = String::from( + "OmniGraph serves typed property graphs. Every call runs under the applied policy for your credential; a listed tool or graph can still be refused.\n\ + 1. Call `graphs` for graph ids.\n\ + 2. Call `schema` for the graph before writing a query: it declares the types, properties, keys and edges a query may name.\n\ + 3. Prefer a stored read: `queries` lists them with typed parameters, and `query` runs one.\n", + ); + if settings.allow_adhoc { + text.push_str( + "4. Otherwise write a GQ read and run it with `read`, with every value passed as a declared, typed parameter. `read` refuses mutations.\n", + ); + } + text.push_str( + "Results carry the read's `graph_commit_id`. A result is limited to 1 MiB and a read to 30 seconds; narrow the query with filters or `limit` when a result is too large. A refused call returns `isError` with the HTTP `status` and the server's `error` body, or an MCP `code` and `message`.", + ); + text +} + +fn tools(settings: &McpSettings) -> Vec { + TOOLS + .iter() + .filter(|spec| spec.gate.open(settings)) + .map(|spec| { + let mut tool = Tool::new(spec.name, spec.description, object((spec.input)())) + .with_title(spec.title) + .with_annotations(ToolAnnotations::new().read_only(true).open_world(false)); + if let Some(output) = spec.output { + tool = tool.with_raw_output_schema(Arc::new(object(json!({ + "type": "object", + "anyOf": [output(), error_schema()], + })))); + } + tool + }) + .collect() +} + +fn object(value: Value) -> serde_json::Map { + match value { + Value::Object(map) => map, + _ => unreachable!("tool schemas are JSON object literals"), + } +} + +fn text_schema(description: &str) -> Value { + json!({"type": "string", "minLength": 1, "description": description}) +} + +fn graph_argument() -> Value { + text_schema("Graph id, from `graphs`.") +} + +fn no_arguments_schema() -> Value { + json!({"type": "object", "properties": {}, "additionalProperties": false}) +} + +fn graph_schema() -> Value { + json!({ + "type": "object", + "properties": {"graph": graph_argument()}, + "required": ["graph"], + "additionalProperties": false, + }) +} + +fn query_schema() -> Value { + json!({ + "type": "object", + "properties": { + "graph": graph_argument(), + "name": text_schema("Stored read name, from `queries`."), + "params": {"type": "object", "description": "Values keyed by the read's declared parameter names."}, + "branch": text_schema("Branch to read. Defaults to `main`."), + }, + "required": ["graph", "name"], + "additionalProperties": false, + }) +} + +fn read_schema() -> Value { + json!({ + "type": "object", + "properties": { + "graph": graph_argument(), + "query": text_schema("GQ source declaring one or more read queries, or one branch read or `explain` statement."), + "name": text_schema("Which declared query to run when `query` declares several."), + "params": {"type": "object", "description": "Values keyed by the query's declared parameter names, without `$`."}, + "branch": text_schema("Branch to read. Defaults to `main`. Not with `snapshot`."), + "snapshot": text_schema("Graph commit id to read instead of a branch head."), + }, + "required": ["graph", "query"], + "additionalProperties": false, + }) +} + +/// The `isError` envelope: an HTTP refusal (`status` and the route's `error` +/// body) or an MCP-side refusal (`code` and `message`). +fn error_schema() -> Value { + json!({ + "type": "object", + "properties": { + "status": {"type": "integer"}, + "error": {"type": "object"}, + "code": {"type": "string"}, + "message": {"type": "string"}, + }, + "additionalProperties": false, + }) +} + +fn graphs_output_schema() -> Value { + json!({ + "type": "object", + "properties": {"graphs": {"type": "array", "items": { + "type": "object", + "properties": {"graph_id": {"type": "string"}, "display_name": {"type": "string"}}, + "required": ["graph_id", "display_name"], + }}}, + "required": ["graphs"], + }) +} + +fn schema_output_schema() -> Value { + json!({ + "type": "object", + "properties": { + "schema_source": {"type": "string"}, + "system_columns": { + "type": "object", + "properties": {"id": {"type": "string"}, "src": {"type": "string"}, "dst": {"type": "string"}}, + "required": ["id", "src", "dst"], + }, + }, + "required": ["schema_source"], + }) +} + +fn queries_output_schema() -> Value { + let kind = json!({"type": "string", "enum": ["string", "bool", "int", "bigint", "float", "date", "datetime", "blob", "vector", "list"]}); + json!({ + "type": "object", + "properties": {"queries": {"type": "array", "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "tool_name": {"type": "string"}, + "description": {"type": "string"}, + "instruction": {"type": "string"}, + "mutation": {"type": "boolean"}, + "params": {"type": "array", "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "kind": kind, + "item_kind": kind, + "vector_dim": {"type": "integer"}, + "nullable": {"type": "boolean"}, + }, + "required": ["name", "kind", "nullable"], + }}, + }, + "required": ["name", "tool_name", "mutation", "params"], + }}}, + "required": ["queries"], + }) } struct LimitedBytes(Vec); diff --git a/crates/omnigraph-server/src/settings.rs b/crates/omnigraph-server/src/settings.rs index 0a688a1aa..5d20f975c 100644 --- a/crates/omnigraph-server/src/settings.rs +++ b/crates/omnigraph-server/src/settings.rs @@ -392,6 +392,7 @@ pub async fn load_server_settings_with_identity_trust( canonical_root, trust, oidc_trust, + mcp: McpSettings::default(), }) } diff --git a/crates/omnigraph-server/tests/integration/mcp.rs b/crates/omnigraph-server/tests/integration/mcp.rs index 111f7d8b4..d3db398f1 100644 --- a/crates/omnigraph-server/tests/integration/mcp.rs +++ b/crates/omnigraph-server/tests/integration/mcp.rs @@ -60,6 +60,18 @@ fn trust(root: &std::path::Path) -> Arc { } async fn fixture_app() -> (tempfile::TempDir, Router, AppState) { + fixture_app_with(omnigraph_server::McpSettings::default()).await +} + +fn adhoc() -> omnigraph_server::McpSettings { + let mut settings = omnigraph_server::McpSettings::default(); + settings.allow_adhoc = true; + settings +} + +async fn fixture_app_with( + settings: omnigraph_server::McpSettings, +) -> (tempfile::TempDir, Router, AppState) { let temp = init_loaded_graph().await; let graph = graph_path(temp.path()); let policy = temp.path().join("policy.yaml"); @@ -139,7 +151,8 @@ async fn fixture_app() -> (tempfile::TempDir, Router, AppState) { None, ) .unwrap() - .with_oidc_identity_trust(trust(temp.path())); + .with_oidc_identity_trust(trust(temp.path())) + .with_mcp_settings(settings); (temp, omnigraph_server::build_app(state.clone()), state) } @@ -195,7 +208,7 @@ async fn oidc_mcp_reuses_discovery_cedar_and_stored_read_handlers_without_mutati .iter() .map(|t| t["name"].as_str().unwrap()) .collect(); - assert_eq!(names, vec!["graphs", "queries", "query"]); + assert_eq!(names, vec!["graphs", "schema", "queries", "query"]); for tool in list["result"]["tools"].as_array().unwrap() { assert_eq!(tool["annotations"]["readOnlyHint"], true); } @@ -356,6 +369,35 @@ async fn mcp_authenticates_every_request_and_enforces_resource_and_http_bounds() serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()).unwrap(); assert_eq!(metadata["resource"], RESOURCE); assert!(!metadata.to_string().contains("stable_alice")); + // The challenge's URL itself answers the same document, so a client + // probing the path-suffixed location first needs no proxy rewrite. + let advertised = "/.well-known/oauth-protected-resource/clusters/A/incarnations/one"; + let response = app + .clone() + .oneshot(Request::get(advertised).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["cache-control"], "no-store"); + assert!(!response.headers().contains_key(HTTP_API_CONTRACT_HEADER)); + let at_path: Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()).unwrap(); + assert_eq!(at_path, metadata); + for other in [ + "/.well-known/oauth-protected-resource/clusters/A", + "/.well-known/oauth-protected-resource/clusters/A/incarnations/one/", + "/.well-known/oauth-protected-resource/clusters/A/incarnations/two", + "/.well-known/oauth-protected-resource/clusters/A/incarnations/one/extra", + "/.well-known/oauth-protected-resource/mcp", + "/.well-known/oauth-protected-resource/clusters/%41/incarnations/one", + ] { + let response = app + .clone() + .oneshot(Request::get(other).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND, "{other}"); + } let response = app .clone() .oneshot(Request::get("/no-such-route").body(Body::empty()).unwrap()) @@ -373,15 +415,29 @@ async fn mcp_authenticates_every_request_and_enforces_resource_and_http_bounds() token("unknown", RESOURCE), "ogc_retired".into(), ] { + let response = app + .clone() + .oneshot(rpc(Some(&wrong), "tools/list", json!({}))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + // RFC 6750: a presented, refused bearer is `invalid_token`. assert_eq!( - app.clone() - .oneshot(rpc(Some(&wrong), "tools/list", json!({}))) - .await - .unwrap() - .status(), - StatusCode::UNAUTHORIZED + response.headers()["www-authenticate"], + "Bearer error=\"invalid_token\", resource_metadata=\"https://data.example/.well-known/oauth-protected-resource/clusters/A/incarnations/one\"" ); } + // Another scheme is not a presented bearer credential. + let mut basic = rpc(None, "tools/list", json!({})); + basic + .headers_mut() + .insert("authorization", "Basic YWxpY2U6eA==".parse().unwrap()); + let response = app.clone().oneshot(basic).await.unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!( + response.headers()["www-authenticate"], + "Bearer resource_metadata=\"https://data.example/.well-known/oauth-protected-resource/clusters/A/incarnations/one\"" + ); let mut forged = rpc(None, "tools/list", json!({})); forged .extensions_mut() @@ -555,3 +611,213 @@ async fn mcp_requires_explicit_resource_identity_configuration() { StatusCode::NOT_FOUND ); } + +async fn tool_list(app: &Router, token: &str) -> Vec { + let (status, list) = json_response(app, rpc(Some(token), "tools/list", json!({}))).await; + assert_eq!(status, StatusCode::OK, "{list}"); + list["result"]["tools"].as_array().unwrap().clone() +} + +async fn instructions(app: &Router, token: &str) -> String { + let (status, body) = json_response( + app, + rpc( + Some(token), + "initialize", + json!({"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"fixture","version":"1"}}), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + body["result"]["instructions"].as_str().unwrap().to_owned() +} + +/// One line per tool: name, required arguments, read-only hint, whether an +/// output schema is declared. The opt-in `read` tool appears only when the +/// server enables it. +#[tokio::test] +async fn mcp_tool_list_follows_the_adhoc_setting_and_describes_every_tool() { + let alice = token("alice", RESOURCE); + for (settings, expected) in [ + ( + omnigraph_server::McpSettings::default(), + vec![ + "graphs [] read_only output", + "schema [graph] read_only output", + "queries [graph] read_only output", + "query [graph,name] read_only", + ], + ), + ( + adhoc(), + vec![ + "graphs [] read_only output", + "schema [graph] read_only output", + "queries [graph] read_only output", + "query [graph,name] read_only", + "read [graph,query] read_only", + ], + ), + ] { + let enabled = settings.allow_adhoc; + let (_temp, app, _state) = fixture_app_with(settings).await; + let tools = tool_list(&app, &alice).await; + let lines: Vec = tools + .iter() + .map(|tool| { + let required: Vec<&str> = tool["inputSchema"]["required"] + .as_array() + .map(|names| names.iter().map(|n| n.as_str().unwrap()).collect()) + .unwrap_or_default(); + assert_eq!(tool["inputSchema"]["additionalProperties"], false); + assert!(tool["title"].as_str().is_some_and(|t| !t.is_empty())); + assert!(tool["description"].as_str().unwrap().len() > 60); + assert_eq!(tool["annotations"]["openWorldHint"], false); + let mut line = format!( + "{} [{}]{}", + tool["name"].as_str().unwrap(), + required.join(","), + if tool["annotations"]["readOnlyHint"] == true { + " read_only" + } else { + "" + } + ); + if let Some(output) = tool.get("outputSchema") { + assert_eq!(output["type"], "object"); + assert_eq!(output["anyOf"].as_array().unwrap().len(), 2); + line.push_str(" output"); + } + line + }) + .collect(); + assert_eq!(lines, expected); + let text = instructions(&app, &alice).await; + assert!(text.contains("Call `schema`"), "{text}"); + assert_eq!(text.contains("`read`"), enabled, "{text}"); + if !enabled { + let refused = call(&app, &alice, "read", json!({"graph":"default","query":"query q() { match { $p: Person } return { $p.name } }"})).await; + assert!(refused.get("error").is_some(), "{refused}"); + } + } +} + +#[tokio::test] +async fn mcp_schema_and_adhoc_read_reuse_their_routes_and_policy() { + let (temp, app, _state) = fixture_app_with(adhoc()).await; + let alice = token("alice", RESOURCE); + let bob = token("bob", RESOURCE); + + // `schema` answers exactly the route's body. + let schema = call(&app, &alice, "schema", json!({"graph":"default"})).await; + assert_eq!(schema["result"]["isError"], false, "{schema}"); + let (status, native) = json_response(&app, get_request("/graphs/default/schema", &alice)).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(schema["result"]["structuredContent"], native); + assert!( + native["schema_source"] + .as_str() + .unwrap() + .contains("node Person") + ); + // `queries` carries each stored read's typed parameters. + let catalog = call(&app, &alice, "queries", json!({"graph":"default"})).await; + let person = catalog["result"]["structuredContent"]["queries"] + .as_array() + .unwrap() + .iter() + .find(|query| query["name"] == "person") + .unwrap() + .clone(); + assert_eq!( + person["params"], + json!([{"name":"name","kind":"string","nullable":false}]) + ); + + // `read` answers exactly `POST /query`'s body, including branch reads. + let source = "query adults($min: I32) { match { $p: Person $p.age >= $min } return { $p.name } order { $p.name asc } }"; + for arguments in [ + json!({"graph":"default","query":source,"params":{"min":30}}), + json!({"graph":"default","query":source,"params":{"min":30},"branch":"main"}), + json!({"graph":"default","query":"branch list"}), + ] { + let read = call(&app, &alice, "read", arguments.clone()).await; + assert_eq!(read["result"]["isError"], false, "{read}"); + let mut body = arguments.clone(); + body.as_object_mut().unwrap().remove("graph"); + let request = Request::post("/graphs/default/query") + .header(HTTP_API_CONTRACT_HEADER, HTTP_API_CONTRACT) + .header("content-type", "application/json") + .header("authorization", format!("Bearer {alice}")) + .body(Body::from(body.to_string())) + .unwrap(); + let (status, native) = json_response(&app, request).await; + assert_eq!(status, StatusCode::OK, "{native}"); + assert_eq!(read["result"]["structuredContent"], native); + } + + // A denied actor gets the route's refusal in the shared envelope. + for (tool, arguments) in [ + ("schema", json!({"graph":"default"})), + ( + "read", + json!({"graph":"default","query":source,"params":{"min":1}}), + ), + ] { + let denied = call(&app, &bob, tool, arguments).await; + assert_eq!(denied["result"]["isError"], true, "{denied}"); + assert_eq!( + denied["result"]["structuredContent"]["status"], 403, + "{denied}" + ); + assert_eq!( + denied["result"]["structuredContent"]["error"]["code"], + "forbidden" + ); + } + + // Writes never pass through `read`, and leave the graph unchanged. + let before = manifest_dataset_version(&graph_path(temp.path())).await; + for query in [ + "query add() { insert Person { name: \"Mallory\", age: 1 } }", + "query bump() { update Person set { age: 99 } where name = \"Alice\" }", + "query drop() { delete Person where name = \"Alice\" }", + "branch create feature from main", + ] { + let refused = call( + &app, + &alice, + "read", + json!({"graph":"default","query":query}), + ) + .await; + assert_eq!(refused["result"]["isError"], true, "{query}: {refused}"); + assert_eq!( + refused["result"]["structuredContent"]["status"], 400, + "{query}: {refused}" + ); + } + let unknown = call( + &app, + &alice, + "read", + json!({"graph":"default","query":"branch list","settings":{}}), + ) + .await; + assert_eq!(unknown["result"]["isError"], true, "{unknown}"); + assert_eq!( + manifest_dataset_version(&graph_path(temp.path())).await, + before + ); + + // The request bound applies to the inline source as to any MCP request. + let oversized = rpc( + Some(&alice), + "tools/call", + json!({"name":"read","arguments":{"graph":"default","query":format!("// {}\nbranch list", "x".repeat(64 * 1024))}}), + ); + assert_eq!( + app.clone().oneshot(oversized).await.unwrap().status(), + StatusCode::PAYLOAD_TOO_LARGE + ); +} diff --git a/crates/omnigraph-server/tests/integration/openapi.rs b/crates/omnigraph-server/tests/integration/openapi.rs index d8cbcbd7f..7daf73c1a 100644 --- a/crates/omnigraph-server/tests/integration/openapi.rs +++ b/crates/omnigraph-server/tests/integration/openapi.rs @@ -192,6 +192,7 @@ fn openapi_info_contains_version() { // protected route nests under `/graphs/{graph_id}/…`. const EXPECTED_PATHS: &[&str] = &[ "/.well-known/oauth-protected-resource", + "/.well-known/oauth-protected-resource/{resource_path}", "/healthz", "/readyz", "/graphs", @@ -2191,7 +2192,11 @@ async fn auth_mode_healthz_still_has_no_security() { .body(Body::empty()) .unwrap(); let (_, json) = json_response(&app, request).await; - for path in ["/healthz", "/.well-known/oauth-protected-resource"] { + for path in [ + "/healthz", + "/.well-known/oauth-protected-resource", + "/.well-known/oauth-protected-resource/{resource_path}", + ] { let operation = &json["paths"][path]["get"]; assert!( operation.get("security").is_none() || operation["security"].is_null(), @@ -2369,6 +2374,7 @@ async fn multi_mode_openapi_keeps_management_paths_flat() { "/cluster/deployments", "/cluster/deployments/{id}", "/.well-known/oauth-protected-resource", + "/.well-known/oauth-protected-resource/{resource_path}", ] { assert!( paths.contains_key(flat), @@ -2408,6 +2414,7 @@ async fn multi_mode_openapi_prefixes_operation_ids_with_cluster() { | "/cluster/deployments" | "/cluster/deployments/{id}" | "/.well-known/oauth-protected-resource" + | "/.well-known/oauth-protected-resource/{resource_path}" ) { continue; } @@ -2480,6 +2487,7 @@ async fn multi_mode_openapi_declares_graph_id_path_parameter() { "/cluster/deployments", "/cluster/deployments/{id}", "/.well-known/oauth-protected-resource", + "/.well-known/oauth-protected-resource/{resource_path}", ] { let item = paths.get(flat).unwrap(); for method in ["get", "head", "post", "put", "delete", "patch"] { @@ -2611,7 +2619,7 @@ fn openapi_describes_api_contract_admission_and_response_identity() { .iter() .filter(|parameter| parameter["name"] == HTTP_API_CONTRACT_HEADER) .collect(); - let oauth = path == "/.well-known/oauth-protected-resource"; + let oauth = path.starts_with("/.well-known/oauth-protected-resource"); let protected = path == "/graphs" || path.starts_with("/graphs/") || path.starts_with("/cluster/"); if protected { diff --git a/docs/rfcs/2026-09-09-identity-credentials-and-applied-policy.md b/docs/rfcs/2026-09-09-identity-credentials-and-applied-policy.md index e48c09ae3..d61398322 100644 --- a/docs/rfcs/2026-09-09-identity-credentials-and-applied-policy.md +++ b/docs/rfcs/2026-09-09-identity-credentials-and-applied-policy.md @@ -309,9 +309,11 @@ per-cluster native signing key. When OIDC trust is configured, the server exposes protected-resource metadata and a stateless `/mcp` transport built on the official Rust MCP SDK. Its -initial tools provide graph existence, stored-read catalog and stored-read -invocation through the same HTTP handler functions and Cedar gates. The +initial tools provided graph existence, stored-read catalog and stored-read +invocation through the same HTTP handler functions and Cedar gates; the read-kind assertion prevents invoking a mutation through the read tool. +[RFC 0003](0003-mcp-server-surface.md) owns the tool catalog, discovery paths +and mounting from then on. Protocol framing, negotiation and cancellation use the SDK. Requests are bounded to 64 KiB, returned tool data to 1 MiB, read attempts to 30 seconds and concurrent tools to 16. Tool visibility is not permission to invoke. diff --git a/docs/rfcs/2026-09-26-self-contained-server-testing.md b/docs/rfcs/2026-09-26-self-contained-server-testing.md index 87828f9c9..c70002d10 100644 --- a/docs/rfcs/2026-09-26-self-contained-server-testing.md +++ b/docs/rfcs/2026-09-26-self-contained-server-testing.md @@ -175,7 +175,7 @@ separately. |---|---| | `omnigraph-gqt` runner (`src/lib.rs`) | Case parsing, server epoch identities, comparisons, capability admission and terminal reporting | | `omnigraph-gqt` supervisor (`src/dst_runner.rs`) | Immutable input, worker/server/helper lifetime, real deadlines, replay and containment | -| `omnigraph-gqt` client adapters, one per API (`engine`, `http`, `cli`; `mcp` once an MCP tool executes an authored GQ read) | Dispatch of each step by its header's API against the one owned server; the physical read after containment | +| `omnigraph-gqt` client adapters, one per API (`engine`, `http`, `cli`, `mcp`) | Dispatch of each step by its header's API against the one owned server; the physical read after containment | | CLI/server process support (`crates/omnigraph-cli/tests/support/mod.rs`: `converged_loaded_cluster`, `spawn_server_with_cluster`, `TestServer`) | Local cluster provisioning, production connection adapter, readiness and lifecycle controls | | `omnigraph-dst` `run_universe` | One controlled scheduling, time, entropy and storage environment; its server consumer is owned by the Deterministic server execution proposal | | Engine test support (`crates/omnigraph/tests/helpers/`) | Real engine calls, storage controls and physical-state oracles | @@ -209,7 +209,7 @@ that path. | Step header | Where the step goes under the `omnigraph-server` route | |---|---| | `via http` | The runner's real HTTP client, kernel TCP on loopback, the real server process and its listener, `axum::serve`, router, authorization, handler, the embedded engine, and the response back over the same socket. | -| `via mcp` | The server's MCP surface at `/mcp` (`crates/omnigraph-server/src/mcp.rs`, streamable HTTP, mounted only when an OIDC resource is configured). Its shipped tools invoke stored reads only, so `via mcp` is refused before fixture effects as `unsupported_capability: mcp` until a tool that executes an authored GQ read exists ([RFC 0003](0003-mcp-server-surface.md) or its amendment); `via mcp` is admitted on `--- query` only. | +| `via mcp` | The server's MCP surface at `/mcp` (`crates/omnigraph-server/src/mcp.rs`, streamable HTTP, mounted only when an OIDC resource is configured). A `--- query` step runs its authored GQ read through the `read` tool ([RFC 0003](0003-mcp-server-surface.md)), so the owned server starts with `--mcp-allow-adhoc`; `via mcp` is admitted on `--- query` only. | | `--- cli` | The `omnigraph` binary spawned by the runner in served mode against the same listener, so the CLI's own HTTP client crosses the same socket path; its evidence is exit code, stdout and stderr. | | `via engine` | On a `--- query`, a read-only `Omnigraph` open of the owned root, admitted only after a lifecycle step has contained the serving process: the physical read. `--- mutate via engine` in such a file is refused as `invalid_case`. | diff --git a/docs/user/index.md b/docs/user/index.md index 9277096c8..65619cc62 100644 --- a/docs/user/index.md +++ b/docs/user/index.md @@ -34,6 +34,7 @@ deployment. - [Managed data credentials and queries](cli/managed-data.md) - [Managed creation, configuration and deletion](cli/managed-lifecycle.md) - [HTTP server](operations/server.md) +- [MCP](operations/mcp.md) - [Troubleshooting](operations/troubleshooting.md) ## Operate a deployment diff --git a/docs/user/operations/mcp.md b/docs/user/operations/mcp.md new file mode 100644 index 000000000..cf1671064 --- /dev/null +++ b/docs/user/operations/mcp.md @@ -0,0 +1,110 @@ +# MCP + +`omnigraph-server` serves the Model Context Protocol at `/mcp`, so Claude, +ChatGPT, Cursor and other MCP clients can read a graph with a URL and the +caller's own credential. It is the only OmniGraph MCP implementation. + +Every tool calls the same handler as its HTTP route, under the same actor and +the same [policy](policy.md) action. A tool appearing in the list is not +permission to use it: a call the policy refuses returns the route's refusal. + +## Enable it + +`/mcp` is mounted when the server is started with an +[OIDC resource identity](server.md#oidc-resource-identities-and-mcp) +(`--oidc-identity-trust`). The resource identifier in that file is the URL users +enter in their MCP client, for example `https://graphs.example.com/mcp`. + +The endpoint is stateless Streamable HTTP built on the official Rust MCP SDK: +each request is a `POST` that carries its own bearer credential and gets a JSON +response. There is no MCP session, and a successful `initialize` never admits a +later unauthenticated request. + +### Sign-in and discovery + +A request without a valid credential is refused with 401 before its body is +read. The response's challenge names the protected-resource metadata: + +```text +WWW-Authenticate: Bearer resource_metadata="https://graphs.example.com/.well-known/oauth-protected-resource/mcp" +``` + +When a bearer credential was presented and refused (expired, wrong audience, +unknown subject), the challenge also carries `error="invalid_token"`, which tells +a client to refresh or sign in again. + +The server answers the metadata both at `/.well-known/oauth-protected-resource` +and at the advertised path: the well-known prefix followed by the resource +identifier's path (`/mcp` above). Any other path below the prefix is 404. No +proxy rewrite is needed. + +### Host and Origin + +The SDK refuses a request whose `Host` is not allowed with 403. Allowed hosts +are `localhost`, `127.0.0.1`, `::1` and the resource identifier's authority. +A browser `Origin` must be the resource identifier's origin; native clients send +no `Origin`. + +## Settings + +| Flag | Environment | Default | Effect | +|---|---|---|---| +| `--mcp-allow-adhoc` | `OMNIGRAPH_MCP_ALLOW_ADHOC=1` | off | Lists and admits the `read` tool | + +The setting is read once at boot. It enables a tool; it grants nothing. When it +is off, `read` is absent from `tools/list` and a call to it is refused as an +unknown tool. + +## Tools + +| Tool | Arguments | Runs | Policy action | +|---|---|---|---| +| `graphs` | none | `GET /graphs/discovery` | any authenticated identity | +| `schema` | `graph` | `GET /graphs/{graph}/schema` | `read` | +| `queries` | `graph` | `GET /graphs/{graph}/queries`, stored mutations removed | `read` | +| `query` | `graph`, `name`, optional `params`, `branch` | `POST /graphs/{graph}/queries/{name}` as a read | `invoke_query`, then `read` | +| `read` (opt-in) | `graph`, `query`, optional `name`, `params`, `branch`, `snapshot` | `POST /graphs/{graph}/query` | `read` | + +All tools are read-only. The server's `initialize` instructions tell a model to +call `graphs`, then `schema`, then a stored read with `query` or, when enabled, +its own GQ read with `read`. + +- `queries` lists each stored read with its typed parameters (`name`, `kind`, + `nullable`, plus `item_kind` for lists and `vector_dim` for vectors). +- `query` refuses stored mutations. +- `read` takes GQ source with typed parameter declarations, the same body as + `POST /query`. It also answers `branch list`, `branch diff`, + `branch merge-base`, `branch merge … dry run` and `explain query …`. It refuses + `insert`, `update`, `delete` and branch writes with the route's 400. + +## Results and errors + +A successful call returns the route's JSON response body as structured content, +plus a text copy for older clients. `graphs`, `schema` and `queries` declare an +output schema. + +A refused call returns `isError: true` with one of two shapes: + +- `{"status": , "error": }` when the route + refused it, for example 403 for a policy denial, 404 for an unknown or hidden + graph, 503 for a graph that is not available; +- `{"code": "...", "message": "..."}` when the MCP layer refused it: + `capacity_exceeded`, `cancelled`, `deadline_exceeded` or `result_too_large`. + +## Limits + +| Limit | Value | +|---|---| +| Request body | 64 KiB | +| Complete tool result, including the text copy | 1 MiB | +| Wait for one read | 30 seconds | +| Concurrent tool executions | 16 | + +A cancelled or expired call stops the caller's wait. The execution keeps its +input reservation and slot until it finishes, and there is no later result +lookup. See [admission and shutdown](../deployment.md). + +## Logs + +Verbose MCP SDK logs contain query arguments and results, so the server keeps +them out of its log; see [Logs](server.md#logs). diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 2ffed9fa7..209f9a549 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -186,30 +186,10 @@ never fetch keys or call a control service, so publisher outages eventually prevent new OIDC access; accepted operations can finish. Admission refresh does not restart writers. Graph configuration retains its normal activation. -With this profile configured, the server additionally exposes: - -- `GET /.well-known/oauth-protected-resource`: public resource identifier, - authorization server and supported bearer delivery, without identity lists. -- `/mcp`: Streamable HTTP MCP using the maintained Rust SDK. Authentication - failures advertise protected-resource metadata for standard OAuth clients. - -The initial MCP tools are `graphs` (IDs and names), `queries` (permitted stored -read names for one graph), and `query` (a named stored read with parameters and -an optional branch). Mutation definitions are excluded and cannot be invoked -through a read tool. These tools use the same actor and Cedar checks as HTTP -graph requests. Limits are 30 seconds to wait, 16 concurrent executions, -64 KiB requests and 1 MiB results. Cancelled or expired callers stop waiting; -execution retains its input and slot until it finishes, with no later result -lookup. See [admission and shutdown](../deployment.md). - -Verbose MCP SDK logs contain query arguments and results, so the server keeps -them out of its log; see [Logs](#logs). - -Requests require the resource authority or a loopback host; browsers must use -the resource origin, while native clients can omit `Origin`. The deployment -must separately supply a reachable server URL and public metadata at the -advertised resource location. Direct/static deployments without this option -keep their existing routes and do not expose MCP. +With this profile configured, the server also serves public +protected-resource metadata and an OAuth challenge on `/mcp`, so standard OAuth +clients such as Claude can sign in. The [MCP guide](mcp.md) describes the +endpoint, its tools, settings and limits. ## Route families @@ -222,8 +202,8 @@ keep their existing routes and do not expose MCP. | `GET /graphs/discovery` | Graph IDs and display names only; requires an identity credential | | `POST /cluster/deployments` | Submit an exact-ID deployment: graph creation/deletion or schema, query, policy, provider and Blob-binding changes | | `GET /cluster/deployments`, `GET /cluster/deployments/{id}` | Authorized deployment status and current-process activation observation | -| `GET /.well-known/oauth-protected-resource` | Public OIDC resource metadata; only when OIDC trust is configured | -| `/mcp` | Stored reads and discovery over MCP; only when OIDC trust is configured | +| `GET /.well-known/oauth-protected-resource`, also at the resource's path suffix | Public OIDC resource metadata; only when OIDC trust is configured | +| `/mcp` | The [MCP endpoint](mcp.md); only when OIDC trust is configured | | `/graphs/{id}/query`, `/mutate` | Run inline GQ source | | `/graphs/{id}/mutate/if-graph-commit` | Run an inline conditional mutation | | `/graphs/{id}/queries` | List and invoke stored queries, including conditional mutations | diff --git a/openapi.json b/openapi.json index 27928ac54..95a735bc6 100644 --- a/openapi.json +++ b/openapi.json @@ -31,6 +31,39 @@ } } }, + "/.well-known/oauth-protected-resource/{resource_path}": { + "get": { + "tags": [ + "management" + ], + "summary": "The same metadata at the location the `/mcp` challenge advertises for a\nresource identifier with a path: the well-known prefix followed by that\npath (`/.well-known/oauth-protected-resource/mcp` for `https://host/mcp`).\nEvery other suffix is 404.", + "operationId": "oauthProtectedResourceMetadataAtResourcePath", + "parameters": [ + { + "name": "resource_path", + "in": "path", + "description": "The configured resource identifier's path, without its leading `/`", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Public OAuth protected resource metadata", + "content": { + "application/json": { + "schema": {} + } + } + }, + "404": { + "description": "Not the configured resource's path, or OAuth resource identity is not configured" + } + } + } + }, "/cluster/deployments": { "get": { "tags": [ From b816f8cc8932983aad2204eb390317deaad7d62b Mon Sep 17 00:00:00 2001 From: Ragnor Comerford Date: Tue, 13 Oct 2026 21:51:49 -0700 Subject: [PATCH 2/2] feat(server): /mcp for every credential source, read parity, resources and opt-in writes --- ...cp-every-credential-parity-writes.added.md | 14 + crates/omnigraph-server/src/handlers.rs | 4 +- crates/omnigraph-server/src/lib.rs | 12 +- crates/omnigraph-server/src/main.rs | 16 +- crates/omnigraph-server/src/mcp.rs | 1271 +++++++++++++++-- .../omnigraph-server/tests/integration/mcp.rs | 540 ++++++- docs/user/operations/mcp.md | 105 +- docs/user/operations/server.md | 6 +- 8 files changed, 1807 insertions(+), 161 deletions(-) create mode 100644 changelog.d/mcp-every-credential-parity-writes.added.md diff --git a/changelog.d/mcp-every-credential-parity-writes.added.md b/changelog.d/mcp-every-credential-parity-writes.added.md new file mode 100644 index 000000000..4991eeaac --- /dev/null +++ b/changelog.d/mcp-every-credential-parity-writes.added.md @@ -0,0 +1,14 @@ +- `/mcp` is now mounted whenever the server has a bearer credential source + (static tokens, signed data tokens or OIDC), so a self-hosted server reaches + Claude Code, Cursor and other MCP clients with its existing token; OAuth + discovery stays OIDC-only, `--mcp-allowed-host` admits extra `Host` names, and + without OIDC any request carrying an `Origin` header is refused. + New read tools `health`, `snapshot`, `branches_list`, `commits_list`, + `commits_get`, `commits_changes` and `changes_poll`, and resources for the + graph list and each graph's schema and branches, call the same handlers and + policy actions as their HTTP routes. Write tools `mutate`, `load`, + `branches_create`, `branches_delete` and `branches_merge` are listed only with + `--mcp-allow-writes`; they never retry, honour `if_graph_commit`, and report + `outcome_unknown` instead of guessing. See [MCP][mcp-every-credential-guide]. + +[mcp-every-credential-guide]: ../docs/user/operations/mcp.md diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs index 2c46243a6..2ce860570 100644 --- a/crates/omnigraph-server/src/handlers.rs +++ b/crates/omnigraph-server/src/handlers.rs @@ -2789,7 +2789,7 @@ async fn branch_create_body( /// when route nesting changes" class. #[derive(Deserialize)] pub(crate) struct BranchPath { - branch: String, + pub(crate) branch: String, } #[utoipa::path( @@ -3165,7 +3165,7 @@ pub(crate) async fn server_commit_list( /// for the design rationale — same pattern, different field name. #[derive(Deserialize)] pub(crate) struct CommitPath { - commit_id: String, + pub(crate) commit_id: String, } #[utoipa::path( diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index dfc583823..941e16ac2 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -948,6 +948,14 @@ impl AppState { .map(str::to_owned) } + /// Whether any bearer credential can authenticate a request: static + /// tokens, signed data tokens or OIDC. Fixed at construction. + pub(crate) fn has_credential_source(&self) -> bool { + !self.bearer_tokens.is_empty() + || self.data_token_trust.is_some() + || self.oidc_identity_trust.is_some() + } + fn requires_bearer_auth(&self) -> bool { if !self.bearer_tokens.is_empty() || self.data_token_trust.is_some() @@ -2750,7 +2758,9 @@ pub fn build_app(state: AppState) -> Router { .route("/readyz", get(server_ready)) .route("/openapi.json", get(server_openapi)) .merge(protected); - if state.oidc_identity_trust.is_some() { + // MCP needs an authenticated caller; without a credential source there is + // no actor for its policy checks, so the endpoint does not exist. + if state.has_credential_source() { app = app.merge(mcp::router(state.clone())); } app.layer(middleware::from_fn(http_contract::identify_response)) diff --git a/crates/omnigraph-server/src/main.rs b/crates/omnigraph-server/src/main.rs index 0ef31cfac..4fbe49d88 100644 --- a/crates/omnigraph-server/src/main.rs +++ b/crates/omnigraph-server/src/main.rs @@ -55,6 +55,16 @@ struct Cli { /// `OMNIGRAPH_MCP_ALLOW_ADHOC=1`. #[arg(long)] mcp_allow_adhoc: bool, + /// List and admit the MCP write tools (`mutate`, `load`, `branches_create`, + /// `branches_delete`, `branches_merge`), each under its route's policy + /// action. Off by default. Equivalent to setting `OMNIGRAPH_MCP_ALLOW_WRITES=1`. + #[arg(long)] + mcp_allow_writes: bool, + /// A `Host` value `/mcp` admits, as `host` or `host:port`, besides the + /// loopback names and the OIDC resource authority. Repeatable; also read + /// comma-separated from `OMNIGRAPH_MCP_ALLOWED_HOSTS`. + #[arg(long = "mcp-allowed-host", value_name = "HOST")] + mcp_allowed_hosts: Vec, } #[tokio::main] @@ -63,7 +73,11 @@ async fn main() -> Result<()> { init_tracing(); let cli = Cli::parse(); - let mcp = McpSettings::resolve(cli.mcp_allow_adhoc); + let mcp = McpSettings::resolve( + cli.mcp_allow_adhoc, + cli.mcp_allow_writes, + cli.mcp_allowed_hosts, + )?; match (cli.data_token_trust, cli.oidc_identity_trust) { (data, oidc) if data.is_some() || oidc.is_some() => { let settings = load_server_settings_with_identity_trust( diff --git a/crates/omnigraph-server/src/mcp.rs b/crates/omnigraph-server/src/mcp.rs index 11a6e871f..6877b88a3 100644 --- a/crates/omnigraph-server/src/mcp.rs +++ b/crates/omnigraph-server/src/mcp.rs @@ -1,12 +1,12 @@ -//! Read-only MCP transport over the existing graph discovery, schema, stored-query -//! and inline-query handlers. +//! MCP transport over the existing HTTP handlers: every tool and resource calls +//! the handler its route calls, under the same actor and policy action. //! The official SDK owns protocol framing, negotiation and cancellation. use std::future::Future; use std::sync::Arc; use std::time::Duration; use axum::body::{Bytes, to_bytes}; -use axum::extract::{Path, Request, State}; +use axum::extract::{Path, Query, RawQuery, Request, State}; use axum::http::{HeaderMap, HeaderValue, StatusCode, Uri, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; @@ -14,7 +14,9 @@ use axum::routing::get; use axum::{Extension, Json, Router}; use rmcp::model::{ CallToolRequestParams, CallToolResponse, CallToolResult, ErrorData, Implementation, - ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, ToolAnnotations, + ListResourceTemplatesResult, ListResourcesResult, ListToolsResult, PaginatedRequestParams, + ReadResourceRequestParams, ReadResourceResponse, ReadResourceResult, Resource, + ResourceContents, ResourceTemplate, ServerCapabilities, ServerInfo, Tool, ToolAnnotations, }; use rmcp::service::RequestContext; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; @@ -34,10 +36,18 @@ use crate::{ApiError, AppState, AuthenticatedActor, GraphId, GraphKey}; pub(crate) const REQUEST_BYTES: usize = 64 * 1024; const RESULT_BYTES: usize = 1024 * 1024; const READ_DEADLINE: Duration = Duration::from_secs(30); +/// How long a caller waits for a write. The write itself is owned by the +/// server and continues past this; the caller then learns its outcome is +/// unknown. +const WRITE_DEADLINE: Duration = Duration::from_secs(60); const RESOURCE_METADATA_PATH: &str = "/.well-known/oauth-protected-resource"; /// Environment equivalent of `--mcp-allow-adhoc`. pub const MCP_ALLOW_ADHOC_ENV: &str = "OMNIGRAPH_MCP_ALLOW_ADHOC"; +/// Environment equivalent of `--mcp-allow-writes`. +pub const MCP_ALLOW_WRITES_ENV: &str = "OMNIGRAPH_MCP_ALLOW_WRITES"; +/// Comma-separated environment equivalent of repeated `--mcp-allowed-host`. +pub const MCP_ALLOWED_HOSTS_ENV: &str = "OMNIGRAPH_MCP_ALLOWED_HOSTS"; /// Process boot settings for `/mcp`. They select which opt-in tools are listed /// and admitted; they never grant anything, since every tool still runs only @@ -47,14 +57,64 @@ pub const MCP_ALLOW_ADHOC_ENV: &str = "OMNIGRAPH_MCP_ALLOW_ADHOC"; pub struct McpSettings { /// Lists and admits the ad-hoc `read` tool. pub allow_adhoc: bool, + /// Lists and admits the write tools. + pub allow_writes: bool, + /// `Host` values admitted besides the SDK's loopback names and the OIDC + /// resource authority, as `host` or `host:port`. + pub allowed_hosts: Vec, } impl McpSettings { - /// The boot flags, each enabled also by its environment variable. - pub fn resolve(allow_adhoc: bool) -> Self { - Self { - allow_adhoc: allow_adhoc || env_flag(MCP_ALLOW_ADHOC_ENV), + /// The boot flags, each also set by its environment variable; allowed + /// hosts from both sources are combined. + /// + /// # Errors + /// + /// An allowed host that is not a bare `host` or `host:port`. + pub fn resolve( + allow_adhoc: bool, + allow_writes: bool, + allowed_hosts: Vec, + ) -> color_eyre::eyre::Result { + let mut hosts = allowed_hosts; + if let Ok(value) = std::env::var(MCP_ALLOWED_HOSTS_ENV) { + hosts.extend( + value + .split(',') + .map(str::trim) + .filter(|host| !host.is_empty()) + .map(str::to_owned), + ); } + for host in &hosts { + validate_allowed_host(host)?; + } + Ok(Self { + allow_adhoc: allow_adhoc || env_flag(MCP_ALLOW_ADHOC_ENV), + allow_writes: allow_writes || env_flag(MCP_ALLOW_WRITES_ENV), + allowed_hosts: hosts, + }) + } +} + +/// Accept exactly an authority the SDK compares with `Host`: a host name or +/// address with an optional port, and nothing else. +fn validate_allowed_host(host: &str) -> color_eyre::eyre::Result<()> { + let parsed = url::Url::parse(&format!("http://{host}")); + let valid = parsed.as_ref().is_ok_and(|url| { + url.username().is_empty() + && url.password().is_none() + && url.path() == "/" + && url.query().is_none() + && url.fragment().is_none() + && !host.contains(['/', '?', '#', '@']) + }); + if valid { + Ok(()) + } else { + Err(color_eyre::eyre::eyre!( + "--mcp-allowed-host must be a host or host:port, got `{host}`" + )) } } @@ -90,8 +150,9 @@ fn advertised_metadata_url(resource: &str) -> Option { #[derive(Clone)] pub(crate) struct AdvertisedMetadataPath(Arc); -/// Mount only when an explicitly configured OIDC resource enables this surface. -/// Every HTTP request authenticates again; no MCP session retains an actor. +/// Mounted when the server has a bearer credential source. Discovery metadata +/// and the OAuth challenge exist only with an OIDC resource. Every HTTP request +/// authenticates again; no MCP session retains an actor. pub(crate) fn router(state: AppState) -> Router { let handler = GraphTools { state: state.clone(), @@ -101,6 +162,10 @@ pub(crate) fn router(state: AppState) -> Router { config.legacy_session_mode = false; config.json_response = true; config.max_request_body_bytes = REQUEST_BYTES; + config + .allowed_hosts + .extend(state.mcp_settings().allowed_hosts.iter().cloned()); + let oidc = state.oidc_resource_url().is_some(); if let Some(resource) = state .oidc_resource_url() .and_then(|url| url::Url::parse(&url).ok()) @@ -127,6 +192,21 @@ pub(crate) fn router(state: AppState) -> Router { crate::ingress::admit(&state, request, next).await }, )) + .route_layer(middleware::from_fn( + move |request: Request, next: Next| async move { + // Without an OIDC resource there is no origin to allow. The + // SDK treats an empty origin list as unrestricted, so refuse + // browser requests here; native clients send no `Origin`. + if !oidc && request.headers().contains_key(header::ORIGIN) { + return ( + StatusCode::FORBIDDEN, + "Forbidden: Origin header is not allowed", + ) + .into_response(); + } + next.run(request).await + }, + )) .route_layer(middleware::from_fn_with_state( state.clone(), handlers::require_bearer_auth, @@ -232,6 +312,119 @@ struct GraphTools { slots: Arc, } +/// Why a tool call or resource read produced no handler result. +enum Refusal { + Unauthenticated, + MissingContext, + Capacity, + Cancelled, + Deadline, + Api(ApiError), +} + +impl Refusal { + /// The `isError` tool result. A write's caller learns when its wait ended + /// before the owned write did, or when the server could not tell whether + /// the write completed: the outcome is unknown and must be reconciled, never + /// replayed blindly. + async fn tool_result(self, write: bool) -> CallToolResult { + let unknown = "The write keeps running and its outcome is unknown. Read the target and reconcile before replaying it."; + match self { + Refusal::Unauthenticated => failure( + "unauthenticated", + "A verified bearer credential is required.", + ), + Refusal::MissingContext => { + failure("internal_error", "MCP admission context is missing.") + } + Refusal::Capacity => failure( + "capacity_exceeded", + "The server is handling its maximum concurrent MCP tool calls.", + ), + Refusal::Cancelled if write => outcome_unknown("cancelled", unknown), + Refusal::Cancelled => failure("cancelled", "The read was cancelled."), + Refusal::Deadline if write => outcome_unknown("deadline_exceeded", unknown), + Refusal::Deadline => failure( + "deadline_exceeded", + "The read exceeded its 30 second deadline.", + ), + Refusal::Api(error) => { + let uncertain = write && error.completion_uncertain(); + let mut result = api_failure(error).await; + if uncertain { + if let Some(Value::Object(body)) = result.structured_content.as_mut() { + body.insert("outcome_unknown".into(), Value::Bool(true)); + } + } + result + } + } + } + + /// The JSON-RPC error a resource read answers with. + async fn resource_error(self) -> ErrorData { + match self { + Refusal::Api(error) => { + let status = error.status; + let body = api_failure(error).await.structured_content; + if status == StatusCode::NOT_FOUND { + ErrorData::resource_not_found("resource not found", body) + } else if status.is_client_error() { + ErrorData::invalid_request("resource read refused", body) + } else { + ErrorData::internal_error("resource read failed", body) + } + } + Refusal::Unauthenticated => { + ErrorData::invalid_request("a verified bearer credential is required", None) + } + Refusal::MissingContext => { + ErrorData::internal_error("MCP admission context is missing", None) + } + Refusal::Capacity => ErrorData::internal_error( + "the server is handling its maximum concurrent MCP calls", + None, + ), + Refusal::Cancelled => ErrorData::internal_error("the read was cancelled", None), + Refusal::Deadline => { + ErrorData::internal_error("the read exceeded its 30 second deadline", None) + } + } + } +} + +/// The admitted HTTP request's actor and reservations, which the SDK carries +/// into its task as request extensions. +struct Admitted { + actor: AuthenticatedActor, + observer: crate::operations::ReadObserver, + input: IngressLease, + graph: McpGraphRequest, +} + +fn admitted(context: &RequestContext) -> Result { + let parts = context.extensions.get::(); + let actor = parts + .and_then(|parts| parts.extensions.get::()) + .cloned() + .ok_or(Refusal::Unauthenticated)?; + // Reuse the admitted request's one lease and observer instead of admitting + // again after collection or releasing them before the SDK task finishes. + let parts = parts.ok_or(Refusal::MissingContext)?; + let observer = parts.extensions.get::(); + let input = parts.extensions.get::(); + let graph = parts.extensions.get::(); + match (observer, input, graph) { + (Some(observer), Some(input), Some(graph)) => Ok(Admitted { + actor, + observer: observer.clone(), + input: input.clone(), + graph: graph.clone(), + }), + _ => Err(Refusal::MissingContext), + } +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct NoArguments {} @@ -242,6 +435,19 @@ struct GraphArguments { } #[derive(Deserialize)] #[serde(deny_unknown_fields)] +struct BranchArguments { + graph: String, + #[serde(default)] + branch: Option, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CommitArguments { + graph: String, + commit_id: String, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] struct QueryArguments { graph: String, name: String, @@ -264,15 +470,104 @@ struct ReadArguments { #[serde(default)] snapshot: Option, } +/// The change vocabulary's filters and page continuation, sent to the route +/// as the same repeated query parameters an HTTP caller sends. +#[derive(Deserialize, Default)] +#[serde(deny_unknown_fields)] +struct ChangeFilters { + #[serde(default)] + kind: Vec, + #[serde(default)] + r#type: Vec, + #[serde(default)] + op: Vec, + #[serde(default)] + page_token: Option, + #[serde(default)] + limit: Option, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CommitChangesArguments { + graph: String, + commit_id: String, + #[serde(flatten)] + filters: ChangeFilters, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ChangesPollArguments { + graph: String, + #[serde(default)] + branch: Option, + #[serde(default)] + cursor: Option, + #[serde(default)] + start: Option, + #[serde(flatten)] + filters: ChangeFilters, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct MutateArguments { + graph: String, + query: String, + #[serde(default)] + name: Option, + #[serde(default)] + params: Option>, + #[serde(default)] + branch: Option, + #[serde(default)] + if_graph_commit: Option, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct LoadArguments { + graph: String, + #[serde(default)] + branch: Option, + #[serde(default)] + from: Option, + mode: String, + data: String, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct BranchCreateArguments { + graph: String, + name: String, + #[serde(default)] + from: Option, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct BranchDeleteArguments { + graph: String, + name: String, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct BranchMergeArguments { + graph: String, + source: String, + #[serde(default)] + target: Option, +} impl ServerHandler for GraphTools { fn get_info(&self) -> ServerInfo { - ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) - .with_server_info(Implementation::new( - "omnigraph-server", - env!("CARGO_PKG_VERSION"), - )) - .with_instructions(instructions(self.state.mcp_settings())) + ServerInfo::new( + ServerCapabilities::builder() + .enable_tools() + .enable_resources() + .build(), + ) + .with_server_info(Implementation::new( + "omnigraph-server", + env!("CARGO_PKG_VERSION"), + )) + .with_instructions(instructions(self.state.mcp_settings())) } async fn list_tools( @@ -296,58 +591,143 @@ impl ServerHandler for GraphTools { request: CallToolRequestParams, context: RequestContext, ) -> Result { - if self.get_tool(&request.name).is_none() { + let Some(spec) = spec(self.state.mcp_settings(), &request.name) else { return Err(ErrorData::invalid_params("unknown MCP tool", None)); - } - let parts = context.extensions.get::(); - let actor = parts - .and_then(|parts| parts.extensions.get::()) - .cloned(); - let Some(actor) = actor else { - return Ok(failure( - "unauthenticated", - "A verified bearer credential is required.", - ) - .into()); }; - // The SDK transports the admitted HTTP request's extensions into its - // task. Reuse that one lease and observer instead of admitting again - // after collection or releasing them before the SDK task finishes. - let Some((observer, input, graph)) = parts.and_then(|parts| { - parts - .extensions - .get::() - .zip(parts.extensions.get::()) - .zip(parts.extensions.get::()) - .map(|((observer, input), graph)| (observer.clone(), input.clone(), graph.clone())) - }) else { - return Ok(failure("internal_error", "MCP admission context is missing.").into()); + let write = spec.gate == Gate::Writes; + let admitted = match admitted(&context) { + Ok(admitted) => admitted, + Err(refusal) => return Ok(refusal.tool_result(write).await.into()), }; let service = self.clone(); - let execution_input = input.clone(); - let execution_graph = graph.clone(); + let execution_input = admitted.input.clone(); + let execution_graph = admitted.graph.clone(); + let actor = admitted.actor; let result = self - .run_read( - observer, - input, - graph, + .run_owned( + admitted.observer, + admitted.input, + admitted.graph, async move { service .execute(request, actor, execution_input, execution_graph) .await }, context.ct.cancelled(), - READ_DEADLINE, + if write { WRITE_DEADLINE } else { READ_DEADLINE }, ) .await; + let result = match result { + Ok(result) => result, + Err(refusal) => refusal.tool_result(write).await, + }; Ok(fit_result(result).into()) } + + async fn list_resources( + &self, + _: Option, + _: RequestContext, + ) -> Result { + Ok(ListResourcesResult::with_all_items(vec![ + Resource::new(GRAPHS_RESOURCE, "graphs") + .with_title("Graphs") + .with_description("The graphs this server serves, as `graphs` returns them.") + .with_mime_type("application/json"), + ])) + } + + async fn list_resource_templates( + &self, + _: Option, + _: RequestContext, + ) -> Result { + Ok(ListResourceTemplatesResult::with_all_items(vec![ + ResourceTemplate::new("omnigraph://graphs/{graph}/schema", "schema") + .with_title("Graph schema") + .with_description("A graph's schema as `.pg` source, as `schema` returns it. Read it before writing a query.") + .with_mime_type("text/plain"), + ResourceTemplate::new("omnigraph://graphs/{graph}/branches", "branches") + .with_title("Graph branches") + .with_description("A graph's branch names, as `branches_list` returns them.") + .with_mime_type("application/json"), + ])) + } + + async fn read_resource( + &self, + request: ReadResourceRequestParams, + context: RequestContext, + ) -> Result { + let Some(resource) = GraphResource::parse(&request.uri) else { + return Err(ErrorData::resource_not_found("resource not found", None)); + }; + let admitted = match admitted(&context) { + Ok(admitted) => admitted, + Err(refusal) => return Err(refusal.resource_error().await), + }; + let service = self.clone(); + let execution_graph = admitted.graph.clone(); + let actor = admitted.actor; + let uri = request.uri; + let result = self + .run_owned( + admitted.observer, + admitted.input, + admitted.graph, + async move { service.read(resource, actor, execution_graph).await }, + context.ct.cancelled(), + READ_DEADLINE, + ) + .await; + match result { + Ok((mime_type, text)) => { + if text.len() > RESULT_BYTES { + return Err(ErrorData::internal_error( + "the resource exceeds 1 MiB", + None, + )); + } + Ok(ReadResourceResult::new(vec![ + ResourceContents::text(text, uri).with_mime_type(mime_type), + ]) + .into()) + } + Err(refusal) => Err(refusal.resource_error().await), + } + } +} + +const GRAPHS_RESOURCE: &str = "omnigraph://graphs"; + +/// A resource URI this server answers. +enum GraphResource { + Graphs, + Schema(String), + Branches(String), +} + +impl GraphResource { + fn parse(uri: &str) -> Option { + if uri == GRAPHS_RESOURCE { + return Some(Self::Graphs); + } + let rest = uri.strip_prefix("omnigraph://graphs/")?; + let (graph, kind) = rest.split_once('/')?; + if graph.is_empty() { + return None; + } + match kind { + "schema" => Some(Self::Schema(graph.to_owned())), + "branches" => Some(Self::Branches(graph.to_owned())), + _ => None, + } + } } impl GraphTools { - /// MCP's timeout and cancellation bound the caller's observation. The - /// engine execution retains the SDK-independent observer, input and MCP - /// slot until it completes, including its owned query-worker joins. + /// Tool-shaped wrapper over [`Self::run_owned`] for a read. + #[cfg(test)] async fn run_read( &self, observer: crate::operations::ReadObserver, @@ -360,21 +740,45 @@ impl GraphTools { where F: Future> + Send + 'static, C: Future, + { + match self + .run_owned(observer, input, graph, operation, cancelled, deadline) + .await + { + Ok(result) => result, + Err(refusal) => refusal.tool_result(false).await, + } + } + + /// MCP's timeout and cancellation bound the caller's observation. The + /// execution retains the SDK-independent observer, input and MCP slot + /// until it completes, including its owned query-worker joins and any + /// write it registered through the owned-write boundary. + async fn run_owned( + &self, + observer: crate::operations::ReadObserver, + input: IngressLease, + graph: McpGraphRequest, + operation: F, + cancelled: C, + deadline: Duration, + ) -> Result + where + T: Send + 'static, + F: Future> + Send + 'static, + C: Future, { let Ok(permit) = Arc::clone(&self.slots).try_acquire_owned() else { - return failure( - "capacity_exceeded", - "The server is handling its maximum concurrent MCP reads.", - ); + return Err(Refusal::Capacity); }; let read = observer.spawn_read((input, permit, graph), operation); tokio::select! { - _ = cancelled => failure("cancelled", "The read was cancelled."), + _ = cancelled => Err(Refusal::Cancelled), result = tokio::time::timeout(deadline, read.result()) => { match result { - Ok(Ok(result)) => result, - Ok(Err(error)) => api_failure(error).await, - Err(_) => failure("deadline_exceeded", "The read exceeded its 30 second deadline."), + Ok(Ok(result)) => Ok(result), + Ok(Err(error)) => Err(Refusal::Api(error)), + Err(_) => Err(Refusal::Deadline), } } } @@ -388,16 +792,22 @@ impl GraphTools { graph: McpGraphRequest, ) -> Result { let arguments = Value::Object(request.arguments.unwrap_or_default()); + let state = || State(self.state.clone()); match request.name.as_ref() { + "health" => { + let _: NoArguments = arguments_as(arguments)?; + bounded_result(&handlers::server_health().await.0) + } "graphs" => { let _: NoArguments = arguments_as(arguments)?; - let output = handlers::server_graphs_discovery( - State(self.state.clone()), - Some(Extension(actor)), - ) - .await? - .0; - bounded_result(&output) + bounded_result(&self.discover(actor).await?) + } + "schema" => { + let args: GraphArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = + handlers::server_schema_get(Extension(handle), Some(Extension(actor))).await?; + bounded_result(&output.0) } "queries" => { let args: GraphArguments = arguments_as(arguments)?; @@ -420,7 +830,7 @@ impl GraphTools { }) .map_err(|_| ApiError::bad_request("invalid query parameters"))?; let output = handlers::server_invoke_query( - State(self.state.clone()), + state(), Extension(handle), Extension(input), Some(Extension(actor)), @@ -428,43 +838,296 @@ impl GraphTools { HeaderMap::new(), Bytes::from(body), ) - .await? - .0; - bounded_result(&output) + .await?; + bounded_result(&output.0) } - "schema" => { - let args: GraphArguments = arguments_as(arguments)?; - let (handle, actor) = self.graph(&args.graph, actor, &graph)?; - let output = handlers::server_schema_get(Extension(handle), Some(Extension(actor))) - .await? - .0; - bounded_result(&output) - } - "read" if self.state.mcp_settings().allow_adhoc => { + "read" => { let args: ReadArguments = arguments_as(arguments)?; let (handle, actor) = self.graph(&args.graph, actor, &graph)?; - let request: crate::api::QueryRequest = serde_json::from_value(json!({ + let request = request_body(json!({ "query": args.query, "name": args.name, "params": args.params, "branch": args.branch, "snapshot": args.snapshot, - })) - .map_err(|_| ApiError::bad_request("invalid MCP tool arguments"))?; + }))?; let output = handlers::server_query( - State(self.state.clone()), + state(), Extension(handle), Some(Extension(actor)), Ok(Json(request)), ) - .await? - .0; - bounded_result(&output) + .await?; + bounded_result(&output.0) + } + "snapshot" => { + let args: BranchArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_snapshot( + Extension(handle), + Some(Extension(actor)), + Query(crate::api::SnapshotQuery { + branch: args.branch, + }), + ) + .await?; + bounded_result(&output.0) + } + "branches_list" => { + let args: GraphArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = + handlers::server_branch_list(Extension(handle), Some(Extension(actor))).await?; + bounded_result(&output.0) + } + "commits_list" => { + let args: BranchArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_commit_list( + Extension(handle), + Some(Extension(actor)), + Query(crate::api::CommitListQuery { + branch: args.branch, + unpublished: false, + unattached: false, + }), + ) + .await?; + bounded_result(&output.0) + } + "commits_get" => { + let args: CommitArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_commit_show( + Extension(handle), + Some(Extension(actor)), + Path(handlers::CommitPath { + commit_id: args.commit_id, + }), + ) + .await?; + bounded_result(&output.0) + } + "commits_changes" => { + let args: CommitChangesArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let query = change_query(&[], &args.filters); + let output = handlers::server_commit_changes( + Extension(handle), + Some(Extension(actor)), + Path(handlers::CommitPath { + commit_id: args.commit_id, + }), + RawQuery(query), + ) + .await?; + bounded_result(&output.0) + } + "changes_poll" => { + let args: ChangesPollArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let query = change_query( + &[ + ("branch", args.branch.as_deref()), + ("cursor", args.cursor.as_deref()), + ("start", args.start.as_deref()), + ], + &args.filters, + ); + let output = handlers::server_changes_feed( + Extension(handle), + Some(Extension(actor)), + RawQuery(query), + ) + .await?; + bounded_result(&output.0) + } + "mutate" => { + let ingress = self.write_ingress(&arguments)?; + let args: MutateArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let request = Json(request_body(json!({ + "query": args.query, + "name": args.name, + "params": args.params, + "branch": args.branch, + }))?); + let output = match args.if_graph_commit { + Some(head) => { + let mut headers = HeaderMap::new(); + headers.insert( + crate::api::GRAPH_COMMIT_PRECONDITION_HEADER, + HeaderValue::from_str(&head).map_err(|_| { + ApiError::bad_request("if_graph_commit is not a valid commit id") + })?, + ); + handlers::server_mutate_if_graph_commit( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + headers, + Ok(request), + ) + .await? + } + None => { + handlers::server_mutate( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + HeaderMap::new(), + Ok(request), + ) + .await? + } + }; + bounded_result(&output.0) + } + "load" => { + let ingress = self.write_ingress(&arguments)?; + let args: LoadArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let request = request_body(json!({ + "branch": args.branch, + "from": args.from, + "mode": args.mode, + "data": args.data, + }))?; + let output = handlers::server_load( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + Json(request), + ) + .await?; + bounded_result(&output.0) + } + "branches_create" => { + let ingress = self.write_ingress(&arguments)?; + let args: BranchCreateArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_branch_create( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + Json(crate::api::BranchCreateRequest { + from: args.from, + name: args.name, + }), + ) + .await?; + bounded_result(&output.0) + } + "branches_delete" => { + let ingress = self.write_ingress(&arguments)?; + let args: BranchDeleteArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let output = handlers::server_branch_delete( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + Path(handlers::BranchPath { branch: args.name }), + ) + .await?; + bounded_result(&output.0) + } + "branches_merge" => { + let ingress = self.write_ingress(&arguments)?; + let args: BranchMergeArguments = arguments_as(arguments)?; + let (handle, actor) = self.graph(&args.graph, actor, &graph)?; + let request = request_body(json!({ + "source": args.source, + "target": args.target, + }))?; + let output = handlers::server_branch_merge( + state(), + Extension(handle), + Extension(ingress), + Some(Extension(actor)), + Ok(Json(request)), + ) + .await?; + bounded_result(&output.0) } _ => Err(ApiError::bad_request("unknown MCP tool")), } } + /// One resource body and its media type, from the same handler and policy + /// action as the matching tool. + async fn read( + &self, + resource: GraphResource, + actor: AuthenticatedActor, + graph: McpGraphRequest, + ) -> Result<(&'static str, String), ApiError> { + match resource { + GraphResource::Graphs => { + Ok(("application/json", pretty(&self.discover(actor).await?)?)) + } + GraphResource::Schema(name) => { + let (handle, actor) = self.graph(&name, actor, &graph)?; + let output = + handlers::server_schema_get(Extension(handle), Some(Extension(actor))).await?; + Ok(("text/plain", output.0.schema_source)) + } + GraphResource::Branches(name) => { + let (handle, actor) = self.graph(&name, actor, &graph)?; + let output = + handlers::server_branch_list(Extension(handle), Some(Extension(actor))).await?; + Ok(("application/json", pretty(&output.0)?)) + } + } + } + + /// Graph ids and names. An identity credential uses `GET /graphs/discovery`, + /// which admits every identity; any other credential uses `GET /graphs` + /// under the `graph_list` action, narrowed to the same two fields. + async fn discover( + &self, + actor: AuthenticatedActor, + ) -> Result { + let state = State(self.state.clone()); + if actor.is_identity() { + return Ok( + handlers::server_graphs_discovery(state, Some(Extension(actor))) + .await? + .0, + ); + } + let listed = handlers::server_graphs_list(state, Some(Extension(actor))) + .await? + .0; + Ok(crate::api::GraphDiscoveryResponse { + graphs: listed + .graphs + .into_iter() + .map(|graph| crate::api::GraphDiscoveryEntry { + display_name: graph.graph_id.clone(), + graph_id: graph.graph_id, + }) + .collect(), + }) + } + + /// A write reserves write ingress for its argument bytes, as the write's + /// HTTP route does at admission; the handler's per-actor admission and + /// owned-write registration follow unchanged. + fn write_ingress(&self, arguments: &Value) -> Result { + let bytes = serde_json::to_vec(arguments) + .map_err(|_| ApiError::bad_request("invalid MCP tool arguments"))? + .len(); + self.state + .workload + .try_ingress(bytes as u64) + .map_err(ApiError::from_workload_reject) + } + fn graph( &self, name: &str, @@ -484,15 +1147,54 @@ impl GraphTools { } } +fn pretty(value: &T) -> Result { + serde_json::to_string_pretty(value) + .map_err(|_| ApiError::internal("could not encode MCP resource")) +} + fn arguments_as(value: Value) -> Result { serde_json::from_value(value).map_err(|_| ApiError::bad_request("invalid MCP tool arguments")) } +/// A route's typed request body, built from tool arguments through the same +/// deserializer its JSON extractor uses. +fn request_body(value: Value) -> Result { + serde_json::from_value(value).map_err(|_| ApiError::bad_request("invalid MCP tool arguments")) +} + +/// The query string an HTTP caller would send for these change arguments. +fn change_query(scalars: &[(&str, Option<&str>)], filters: &ChangeFilters) -> Option { + let mut query = url::form_urlencoded::Serializer::new(String::new()); + for (name, value) in scalars { + if let Some(value) = value { + query.append_pair(name, value); + } + } + for kind in &filters.kind { + query.append_pair("kind", kind); + } + for name in &filters.r#type { + query.append_pair("type", name); + } + for op in &filters.op { + query.append_pair("op", op); + } + if let Some(token) = &filters.page_token { + query.append_pair("page_token", token); + } + if let Some(limit) = filters.limit { + query.append_pair("limit", &limit.to_string()); + } + let query = query.finish(); + (!query.is_empty()).then_some(query) +} + /// Which boot setting, if any, lists and admits a tool. -#[derive(Clone, Copy)] +#[derive(Clone, Copy, PartialEq, Eq)] enum Gate { Always, Adhoc, + Writes, } impl Gate { @@ -500,10 +1202,22 @@ impl Gate { match self { Gate::Always => true, Gate::Adhoc => settings.allow_adhoc, + Gate::Writes => settings.allow_writes, } } } +/// What a tool does to the graph, as MCP annotation hints. +#[derive(Clone, Copy)] +enum Effect { + Read, + /// Creates state without replacing any. + Additive, + Destructive { + idempotent: bool, + }, +} + struct ToolSpec { name: &'static str, title: &'static str, @@ -513,16 +1227,29 @@ struct ToolSpec { /// schema also admits the error envelope: clients validate any structured /// content against it, including an `isError` result's. output: Option Value>, + effect: Effect, gate: Gate, } +const WRITE_CONTRACT: &str = " Never retried by the server. If the result says `outcome_unknown`, the write may or may not have happened: read the target and reconcile before replaying."; + const TOOLS: &[ToolSpec] = &[ + ToolSpec { + name: "health", + title: "Server health", + description: "Return the server's status, version and the storage format version it serves.", + input: no_arguments_schema, + output: Some(health_output_schema), + effect: Effect::Read, + gate: Gate::Always, + }, ToolSpec { name: "graphs", title: "List graphs", - description: "List the graphs this server serves: each graph's `graph_id` and display name. Every other tool takes one of these ids as `graph`. Listing a graph does not mean your credential may read it.", + description: "List the graphs this server serves: each graph's `graph_id` and display name. Every other tool takes one of these ids as `graph`. Listing a graph does not mean your credential may read it. A credential that is not an identity needs the `graph_list` permission.", input: no_arguments_schema, output: Some(graphs_output_schema), + effect: Effect::Read, gate: Gate::Always, }, ToolSpec { @@ -531,6 +1258,7 @@ const TOOLS: &[ToolSpec] = &[ description: "Return a graph's schema as `.pg` source (`schema_source`): its node and edge types, properties with their types, keys, edge endpoints and declared embeddings. Call this before writing any query: GQ names types, properties and edges exactly as declared here, and edges are matched in lowerCamelCase.", input: graph_schema, output: Some(schema_output_schema), + effect: Effect::Read, gate: Gate::Always, }, ToolSpec { @@ -539,6 +1267,7 @@ const TOOLS: &[ToolSpec] = &[ description: "List a graph's stored reads: each `name`, its description and instruction, and typed `params` (`name`, `kind`, `nullable`; `item_kind` for lists, `vector_dim` for vectors). A non-nullable parameter is required. Run one with `query`. Invocation is authorized separately, so a listed read can still be refused.", input: graph_schema, output: Some(queries_output_schema), + effect: Effect::Read, gate: Gate::Always, }, ToolSpec { @@ -547,6 +1276,7 @@ const TOOLS: &[ToolSpec] = &[ description: "Run a stored read by `name` with `params` keyed by its declared parameter names (see `queries`). Reads `branch`, `main` when omitted. Returns `rows`, `row_count` and the read's `graph_commit_id`. Stored mutations are refused.", input: query_schema, output: None, + effect: Effect::Read, gate: Gate::Always, }, ToolSpec { @@ -555,15 +1285,121 @@ const TOOLS: &[ToolSpec] = &[ description: "Run a GQ read query you write. Call `schema` first. Declare every value as a typed parameter (`query people($min: I32) { match { $p: Person $p.age >= $min } return { $p.name } }`) and pass the values in `params`; pick one declared query with `name` when the source declares several. Reads `branch` (default `main`) or a `snapshot`, never both. Search operators (`nearest`, `bm25`, `rrf`) need a trailing `limit`. Also answers `branch list`, `branch diff [into ]`, `branch merge-base …`, `branch merge … dry run` and `explain query …`, sent without `branch`, `snapshot`, `name` or `params`. Mutations and branch writes are refused. Returns `rows`, `row_count` and `graph_commit_id`.", input: read_schema, output: None, + effect: Effect::Read, gate: Gate::Adhoc, }, + ToolSpec { + name: "snapshot", + title: "Branch snapshot", + description: "Return a branch's current snapshot (default `main`): the graph manifest version, the storage format version and every node and edge table with its entity count. Use it to judge a graph's size before querying.", + input: branch_schema, + output: None, + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "branches_list", + title: "List branches", + description: "List a graph's branch names.", + input: graph_schema, + output: Some(branches_output_schema), + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "commits_list", + title: "List commits", + description: "List a branch's commits (default `main`), newest first, each with its `graph_commit_id`, parents, actor and time.", + input: branch_schema, + output: None, + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "commits_get", + title: "Get a commit", + description: "Return one commit by its `graph_commit_id`.", + input: commit_schema, + output: None, + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "commits_changes", + title: "A commit's entity changes", + description: "Return one bounded page of a commit's exact entity changes relative to its first parent, each with `before` and `after`. Filter by `kind`, `type` and `op`. Continue with the returned `next_page_token` as `page_token`, keeping the same filters; a page token is not a feed cursor.", + input: commit_changes_schema, + output: None, + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "changes_poll", + title: "Poll a change feed", + description: "Read one bounded page of a branch's change feed (default `main`), at least once, in first-parent commit order. Start from `start` (`now`, the default, `beginning`, or `after:`), a durable `cursor` from an earlier terminal page, or a `page_token`; these three are mutually exclusive. Keep filters unchanged across pages. Only a terminal page's `cursor` is durable: apply its complete commit blocks idempotently by `graph_commit_id`, then keep the cursor. A 410 `change_feed_gap` needs a new baseline, which this tool does not take.", + input: changes_poll_schema, + output: None, + effect: Effect::Read, + gate: Gate::Always, + }, + ToolSpec { + name: "mutate", + title: "Run a GQ mutation", + description: "Run a GQ mutation (`insert`, `update`, `delete`) on `branch` (default `main`) as one atomic commit, with every value passed as a declared, typed parameter in `params`. For read-modify-write, pass the `graph_commit_id` of your read as `if_graph_commit`: if the branch moved, the call fails with 412 and nothing is written; read again and reconsider, never retry without it. Returns affected counts and the exact `commit` (null for a successful no-op). Also runs `branch create`, `branch delete` and `branch merge` statements, sent without `branch`, `name` or `params`.", + input: mutate_schema, + output: None, + effect: Effect::Destructive { idempotent: false }, + gate: Gate::Writes, + }, + ToolSpec { + name: "load", + title: "Load NDJSON", + description: "Load NDJSON records (`{\"type\": \"\", \"data\": {...}}` per line) into `branch` (default `main`) as one commit. `mode`: `merge` upserts by key, `append` refuses an existing key, `overwrite` replaces the supplied types. Without `from` the branch must exist; with `from` a missing branch is created from it. The whole call must fit in 64 KiB; split larger loads into separate commits.", + input: load_schema, + output: None, + effect: Effect::Destructive { idempotent: false }, + gate: Gate::Writes, + }, + ToolSpec { + name: "branches_create", + title: "Create a branch", + description: "Create branch `name` from `from` (default `main`). Fails with 409 if it already exists.", + input: branch_create_schema, + output: None, + effect: Effect::Additive, + gate: Gate::Writes, + }, + ToolSpec { + name: "branches_delete", + title: "Delete a branch", + description: "Delete branch `name`. Its commits stay reachable only through other branches. Fails with 404 if it does not exist.", + input: branch_delete_schema, + output: None, + effect: Effect::Destructive { idempotent: true }, + gate: Gate::Writes, + }, + ToolSpec { + name: "branches_merge", + title: "Merge a branch", + description: "Merge branch `source` into `target` (default `main`). A merge that publishes returns its exact commit; an already up-to-date merge returns none. A conflict fails with 409 and `merge_conflicts`, writing nothing. To preview, run `branch merge into dry run` with `read`.", + input: branch_merge_schema, + output: None, + effect: Effect::Destructive { idempotent: false }, + gate: Gate::Writes, + }, ]; +fn spec(settings: &McpSettings, name: &str) -> Option<&'static ToolSpec> { + TOOLS + .iter() + .find(|spec| spec.name == name && spec.gate.open(settings)) +} + fn instructions(settings: &McpSettings) -> String { let mut text = String::from( "OmniGraph serves typed property graphs. Every call runs under the applied policy for your credential; a listed tool or graph can still be refused.\n\ 1. Call `graphs` for graph ids.\n\ - 2. Call `schema` for the graph before writing a query: it declares the types, properties, keys and edges a query may name.\n\ + 2. Call `schema` for the graph (or read `omnigraph://graphs/{graph}/schema`) before writing a query: it declares the types, properties, keys and edges a query may name.\n\ 3. Prefer a stored read: `queries` lists them with typed parameters, and `query` runs one.\n", ); if settings.allow_adhoc { @@ -571,8 +1407,13 @@ fn instructions(settings: &McpSettings) -> String { "4. Otherwise write a GQ read and run it with `read`, with every value passed as a declared, typed parameter. `read` refuses mutations.\n", ); } + if settings.allow_writes { + text.push_str( + "Writes: `mutate`, `load` and the `branches_*` tools each publish at most one commit. For read-modify-write, pass your read's `graph_commit_id` to `mutate` as `if_graph_commit`; a 412 means nothing was written. The server never retries a write. A result with `outcome_unknown` means the write may or may not have happened: read the target and reconcile before replaying. Schema and deployment changes are operator-owned and not available here.\n", + ); + } text.push_str( - "Results carry the read's `graph_commit_id`. A result is limited to 1 MiB and a read to 30 seconds; narrow the query with filters or `limit` when a result is too large. A refused call returns `isError` with the HTTP `status` and the server's `error` body, or an MCP `code` and `message`.", + "Reads carry their `graph_commit_id`. A result is limited to 1 MiB and a read to 30 seconds; narrow the query with filters or `limit` when a result is too large. A refused call returns `isError` with the HTTP `status` and the server's `error` body, or an MCP `code` and `message`.", ); text } @@ -582,9 +1423,25 @@ fn tools(settings: &McpSettings) -> Vec { .iter() .filter(|spec| spec.gate.open(settings)) .map(|spec| { - let mut tool = Tool::new(spec.name, spec.description, object((spec.input)())) + let annotations = match spec.effect { + Effect::Read => ToolAnnotations::new().read_only(true), + Effect::Additive => ToolAnnotations::new() + .read_only(false) + .destructive(false) + .idempotent(false), + Effect::Destructive { idempotent } => ToolAnnotations::new() + .read_only(false) + .destructive(true) + .idempotent(idempotent), + }; + let description = if spec.gate == Gate::Writes { + format!("{}{WRITE_CONTRACT}", spec.description) + } else { + spec.description.to_owned() + }; + let mut tool = Tool::new(spec.name, description, object((spec.input)())) .with_title(spec.title) - .with_annotations(ToolAnnotations::new().read_only(true).open_world(false)); + .with_annotations(annotations.open_world(false)); if let Some(output) = spec.output { tool = tool.with_raw_output_schema(Arc::new(object(json!({ "type": "object", @@ -611,51 +1468,153 @@ fn graph_argument() -> Value { text_schema("Graph id, from `graphs`.") } +fn arguments(properties: Value, required: &[&str]) -> Value { + let mut properties = object(properties); + properties.insert("graph".into(), graph_argument()); + let mut required: Vec<&str> = required.to_vec(); + required.insert(0, "graph"); + json!({ + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": false, + }) +} + fn no_arguments_schema() -> Value { json!({"type": "object", "properties": {}, "additionalProperties": false}) } fn graph_schema() -> Value { - json!({ - "type": "object", - "properties": {"graph": graph_argument()}, - "required": ["graph"], - "additionalProperties": false, - }) + arguments(json!({}), &[]) +} + +fn branch_schema() -> Value { + arguments( + json!({"branch": text_schema("Branch. Defaults to `main`.")}), + &[], + ) +} + +fn commit_schema() -> Value { + arguments( + json!({"commit_id": text_schema("A `graph_commit_id`.")}), + &["commit_id"], + ) } fn query_schema() -> Value { - json!({ - "type": "object", - "properties": { - "graph": graph_argument(), + arguments( + json!({ "name": text_schema("Stored read name, from `queries`."), "params": {"type": "object", "description": "Values keyed by the read's declared parameter names."}, "branch": text_schema("Branch to read. Defaults to `main`."), - }, - "required": ["graph", "name"], - "additionalProperties": false, - }) + }), + &["name"], + ) } fn read_schema() -> Value { - json!({ - "type": "object", - "properties": { - "graph": graph_argument(), + arguments( + json!({ "query": text_schema("GQ source declaring one or more read queries, or one branch read or `explain` statement."), "name": text_schema("Which declared query to run when `query` declares several."), "params": {"type": "object", "description": "Values keyed by the query's declared parameter names, without `$`."}, "branch": text_schema("Branch to read. Defaults to `main`. Not with `snapshot`."), "snapshot": text_schema("Graph commit id to read instead of a branch head."), - }, - "required": ["graph", "query"], - "additionalProperties": false, - }) + }), + &["query"], + ) +} + +fn change_filter_properties() -> serde_json::Map { + object(json!({ + "kind": {"type": "array", "items": {"type": "string", "enum": ["node", "edge"]}, "description": "Only these entity kinds."}, + "type": {"type": "array", "items": {"type": "string", "minLength": 1}, "description": "Only these schema type names."}, + "op": {"type": "array", "items": {"type": "string", "enum": ["insert", "update", "delete"]}, "description": "Only these operations."}, + "page_token": text_schema("The previous page's `next_page_token`."), + "limit": {"type": "integer", "minimum": 1, "description": "Maximum changes in this page."}, + })) +} + +fn commit_changes_schema() -> Value { + let mut properties = change_filter_properties(); + properties.insert("commit_id".into(), text_schema("A `graph_commit_id`.")); + arguments(Value::Object(properties), &["commit_id"]) +} + +fn changes_poll_schema() -> Value { + let mut properties = change_filter_properties(); + properties.insert( + "branch".into(), + text_schema("Branch whose feed to read. Defaults to `main`."), + ); + properties.insert( + "cursor".into(), + text_schema("A durable cursor from an earlier terminal page."), + ); + properties.insert( + "start".into(), + json!({"type": "string", "pattern": "^(now|beginning|after:.+)$", "description": "`now` (default), `beginning`, or `after:`."}), + ); + arguments(Value::Object(properties), &[]) +} + +fn mutate_schema() -> Value { + arguments( + json!({ + "query": text_schema("GQ source declaring one or more mutations, or one branch write statement."), + "name": text_schema("Which declared mutation to run when `query` declares several."), + "params": {"type": "object", "description": "Values keyed by the mutation's declared parameter names, without `$`."}, + "branch": text_schema("Branch to write. Defaults to `main`."), + "if_graph_commit": text_schema("Write only while the branch head is still this `graph_commit_id`."), + }), + &["query"], + ) +} + +fn load_schema() -> Value { + arguments( + json!({ + "branch": text_schema("Branch to load into. Defaults to `main`."), + "from": text_schema("Create `branch` from this branch when it does not exist."), + "mode": {"type": "string", "enum": ["merge", "append", "overwrite"], "description": "How existing keys are handled."}, + "data": text_schema("NDJSON: one `{\"type\": \"\", \"data\": {...}}` record per line."), + }), + &["mode", "data"], + ) +} + +fn branch_create_schema() -> Value { + arguments( + json!({ + "name": text_schema("The new branch's name."), + "from": text_schema("Branch to fork. Defaults to `main`."), + }), + &["name"], + ) +} + +fn branch_delete_schema() -> Value { + arguments( + json!({"name": text_schema("The branch to delete.")}), + &["name"], + ) +} + +fn branch_merge_schema() -> Value { + arguments( + json!({ + "source": text_schema("The branch to merge."), + "target": text_schema("The branch that receives the merge. Defaults to `main`."), + }), + &["source"], + ) } /// The `isError` envelope: an HTTP refusal (`status` and the route's `error` -/// body) or an MCP-side refusal (`code` and `message`). +/// body) or an MCP-side refusal (`code` and `message`); a write whose +/// completion is unknown adds `outcome_unknown`. fn error_schema() -> Value { json!({ "type": "object", @@ -664,11 +1623,25 @@ fn error_schema() -> Value { "error": {"type": "object"}, "code": {"type": "string"}, "message": {"type": "string"}, + "outcome_unknown": {"type": "boolean"}, }, "additionalProperties": false, }) } +fn health_output_schema() -> Value { + json!({ + "type": "object", + "properties": { + "status": {"type": "string"}, + "version": {"type": "string"}, + "internal_schema_version": {"type": "integer"}, + "source_version": {"type": "string"}, + }, + "required": ["status", "version", "internal_schema_version"], + }) +} + fn graphs_output_schema() -> Value { json!({ "type": "object", @@ -681,6 +1654,14 @@ fn graphs_output_schema() -> Value { }) } +fn branches_output_schema() -> Value { + json!({ + "type": "object", + "properties": {"branches": {"type": "array", "items": {"type": "string"}}}, + "required": ["branches"], + }) +} + fn schema_output_schema() -> Value { json!({ "type": "object", @@ -769,6 +1750,10 @@ fn failure(code: &str, message: &str) -> CallToolResult { CallToolResult::structured_error(json!({"code":code,"message":message})) } +fn outcome_unknown(code: &str, message: &str) -> CallToolResult { + CallToolResult::structured_error(json!({"code":code,"message":message,"outcome_unknown":true})) +} + async fn api_failure(error: ApiError) -> CallToolResult { let response = error.into_response(); let status = response.status().as_u16(); @@ -902,6 +1887,64 @@ mod tests { .expect("MCP read did not settle"); } + #[tokio::test] + async fn a_write_whose_wait_ends_first_reports_an_unknown_outcome() { + for refusal in [Refusal::Cancelled, Refusal::Deadline] { + let result = refusal.tool_result(true).await; + assert_eq!(result.is_error, Some(true)); + assert_eq!(result.structured_content.unwrap()["outcome_unknown"], true); + } + let read = Refusal::Deadline.tool_result(false).await; + assert!( + read.structured_content + .unwrap() + .get("outcome_unknown") + .is_none() + ); + // A handler failure the server cannot settle keeps that report, for a + // write only; a refusal before effects is a known outcome. + let uncertain = Refusal::Api(ApiError::internal("lost")) + .tool_result(true) + .await + .structured_content + .unwrap(); + assert_eq!(uncertain["status"], 500); + assert_eq!(uncertain["outcome_unknown"], true); + for (error, write) in [ + (ApiError::bad_request("refused"), true), + (ApiError::internal("lost"), false), + ] { + let body = Refusal::Api(error) + .tool_result(write) + .await + .structured_content + .unwrap(); + assert!(body.get("outcome_unknown").is_none(), "{body}"); + } + } + + #[test] + fn allowed_hosts_are_bare_authorities() { + for host in [ + "graphs.internal", + "graphs.internal:8080", + "10.0.0.7", + "[::1]:8080", + ] { + assert!(validate_allowed_host(host).is_ok(), "{host}"); + } + for host in [ + "", + "https://graphs.internal", + "graphs.internal/mcp", + "user@graphs.internal", + "graphs.internal?x", + "a b", + ] { + assert!(validate_allowed_host(host).is_err(), "{host}"); + } + } + #[test] fn complete_tool_result_including_legacy_text_has_one_byte_bound() { let small = json!({"rows":[{"name":"Alice"}]}); diff --git a/crates/omnigraph-server/tests/integration/mcp.rs b/crates/omnigraph-server/tests/integration/mcp.rs index d3db398f1..9319c8fa1 100644 --- a/crates/omnigraph-server/tests/integration/mcp.rs +++ b/crates/omnigraph-server/tests/integration/mcp.rs @@ -208,7 +208,22 @@ async fn oidc_mcp_reuses_discovery_cedar_and_stored_read_handlers_without_mutati .iter() .map(|t| t["name"].as_str().unwrap()) .collect(); - assert_eq!(names, vec!["graphs", "schema", "queries", "query"]); + assert_eq!( + names, + vec![ + "health", + "graphs", + "schema", + "queries", + "query", + "snapshot", + "branches_list", + "commits_list", + "commits_get", + "commits_changes", + "changes_poll", + ] + ); for tool in list["result"]["tools"].as_array().unwrap() { assert_eq!(tool["annotations"]["readOnlyHint"], true); } @@ -589,27 +604,21 @@ async fn mcp_shares_response_lifetimes_closed_admission_and_the_status_lane() { } #[tokio::test] -async fn mcp_requires_explicit_resource_identity_configuration() { +async fn mcp_needs_a_credential_source_and_discovery_needs_an_oidc_resource() { + // An unauthenticated server has no actor for MCP's policy checks. let (_temp, app) = app_for_loaded_graph().await; - assert_eq!( - app.clone() - .oneshot(rpc(None, "tools/list", json!({}))) - .await - .unwrap() - .status(), - StatusCode::NOT_FOUND - ); - assert_eq!( - app.oneshot( - Request::get("/.well-known/oauth-protected-resource") - .body(Body::empty()) - .unwrap() - ) - .await - .unwrap() - .status(), - StatusCode::NOT_FOUND - ); + for uri in ["/mcp", "/.well-known/oauth-protected-resource"] { + let request = if uri == "/mcp" { + rpc(None, "tools/list", json!({})) + } else { + Request::get(uri).body(Body::empty()).unwrap() + }; + assert_eq!( + app.clone().oneshot(request).await.unwrap().status(), + StatusCode::NOT_FOUND, + "{uri}" + ); + } } async fn tool_list(app: &Router, token: &str) -> Vec { @@ -642,20 +651,55 @@ async fn mcp_tool_list_follows_the_adhoc_setting_and_describes_every_tool() { ( omnigraph_server::McpSettings::default(), vec![ + "health [] read_only output", "graphs [] read_only output", "schema [graph] read_only output", "queries [graph] read_only output", "query [graph,name] read_only", + "snapshot [graph] read_only", + "branches_list [graph] read_only output", + "commits_list [graph] read_only", + "commits_get [graph,commit_id] read_only", + "commits_changes [graph,commit_id] read_only", + "changes_poll [graph] read_only", ], ), ( adhoc(), vec![ + "health [] read_only output", "graphs [] read_only output", "schema [graph] read_only output", "queries [graph] read_only output", "query [graph,name] read_only", "read [graph,query] read_only", + "snapshot [graph] read_only", + "branches_list [graph] read_only output", + "commits_list [graph] read_only", + "commits_get [graph,commit_id] read_only", + "commits_changes [graph,commit_id] read_only", + "changes_poll [graph] read_only", + ], + ), + ( + writes(omnigraph_server::McpSettings::default()), + vec![ + "health [] read_only output", + "graphs [] read_only output", + "schema [graph] read_only output", + "queries [graph] read_only output", + "query [graph,name] read_only", + "snapshot [graph] read_only", + "branches_list [graph] read_only output", + "commits_list [graph] read_only", + "commits_get [graph,commit_id] read_only", + "commits_changes [graph,commit_id] read_only", + "changes_poll [graph] read_only", + "mutate [graph,query]", + "load [graph,mode,data]", + "branches_create [graph,name]", + "branches_delete [graph,name]", + "branches_merge [graph,source]", ], ), ] { @@ -671,7 +715,7 @@ async fn mcp_tool_list_follows_the_adhoc_setting_and_describes_every_tool() { .unwrap_or_default(); assert_eq!(tool["inputSchema"]["additionalProperties"], false); assert!(tool["title"].as_str().is_some_and(|t| !t.is_empty())); - assert!(tool["description"].as_str().unwrap().len() > 60); + assert!(tool["description"].as_str().unwrap().len() > 20); assert_eq!(tool["annotations"]["openWorldHint"], false); let mut line = format!( "{} [{}]{}", @@ -821,3 +865,455 @@ async fn mcp_schema_and_adhoc_read_reuse_their_routes_and_policy() { StatusCode::PAYLOAD_TOO_LARGE ); } + +fn writes(mut settings: omnigraph_server::McpSettings) -> omnigraph_server::McpSettings { + settings.allow_writes = true; + settings +} + +/// A server with static bearer tokens only: actor `default` may do anything. +async fn static_token_app(settings: omnigraph_server::McpSettings) -> (tempfile::TempDir, Router) { + let temp = init_loaded_graph().await; + let graph = graph_path(temp.path()); + let policy_path = temp.path().join("policy.yaml"); + std::fs::write(&policy_path, permit_all_policy_yaml(&["default"])).unwrap(); + let state = AppState::open_with_bearer_tokens_and_policy( + graph.to_string_lossy().to_string(), + vec![("default".to_string(), "secret".to_string())], + Some(&policy_path), + ) + .await + .unwrap() + .with_mcp_settings(settings); + (temp, omnigraph_server::build_app(state)) +} + +#[tokio::test] +async fn static_token_server_serves_mcp_without_oauth_discovery() { + let mut settings = omnigraph_server::McpSettings::default(); + settings.allowed_hosts = vec!["graphs.internal:8080".into()]; + let (_temp, app) = static_token_app(settings).await; + + let response = app + .clone() + .oneshot(rpc(Some("secret"), "tools/list", json!({}))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + for credential in [None, Some("wrong")] { + let response = app + .clone() + .oneshot(rpc(credential, "tools/list", json!({}))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + // No OIDC resource: nothing to discover, so no OAuth challenge. + assert!(!response.headers().contains_key("www-authenticate")); + } + assert_eq!( + app.clone() + .oneshot( + Request::get("/.well-known/oauth-protected-resource") + .body(Body::empty()) + .unwrap() + ) + .await + .unwrap() + .status(), + StatusCode::NOT_FOUND + ); + for (host, expected) in [ + ("graphs.internal:8080", StatusCode::OK), + ("127.0.0.1", StatusCode::OK), + ("graphs.internal", StatusCode::FORBIDDEN), + ("attacker.example", StatusCode::FORBIDDEN), + ] { + let mut request = rpc(Some("secret"), "tools/list", json!({})); + request.headers_mut().insert("host", host.parse().unwrap()); + assert_eq!( + app.clone().oneshot(request).await.unwrap().status(), + expected, + "{host}" + ); + } + // No resource origin exists to allow, so any browser Origin is refused. + for origin in ["https://attacker.example", "http://localhost", "null"] { + let mut request = rpc(Some("secret"), "tools/list", json!({})); + request + .headers_mut() + .insert("origin", origin.parse().unwrap()); + assert_eq!( + app.clone().oneshot(request).await.unwrap().status(), + StatusCode::FORBIDDEN, + "{origin}" + ); + } + // A static token is not an identity, so `graphs` asks `GET /graphs`, + // whose `graph_list` action only a server-scoped policy grants. + let graphs = call(&app, "secret", "graphs", json!({})).await; + assert_eq!( + graphs["result"]["structuredContent"]["status"], 403, + "{graphs}" + ); + let (status, native) = json_response(&app, get_request("/graphs", "secret")).await; + assert_eq!(status, StatusCode::FORBIDDEN); + assert_eq!(graphs["result"]["structuredContent"]["error"], native); + let schema = call(&app, "secret", "schema", json!({"graph":"default"})).await; + assert_eq!(schema["result"]["isError"], false, "{schema}"); +} + +async fn http_get(app: &Router, uri: &str, token: &str) -> Value { + let (status, body) = json_response(app, get_request(uri, token)).await; + assert_eq!(status, StatusCode::OK, "{uri}: {body}"); + body +} + +#[tokio::test] +async fn mcp_read_parity_tools_equal_their_routes() { + let (_temp, app) = static_token_app(writes(omnigraph_server::McpSettings::default())).await; + // A commit to inspect, through the MCP write path. + let written = call( + &app, + "secret", + "mutate", + json!({"graph":"default","query":"query add($name: String) { insert Person { name: $name, age: 41 } }","params":{"name":"Dana"}}), + ) + .await; + assert_eq!(written["result"]["isError"], false, "{written}"); + let commit = written["result"]["structuredContent"]["commit"]["graph_commit_id"] + .as_str() + .unwrap() + .to_owned(); + + let health = call(&app, "secret", "health", json!({})).await; + let (_, native_health) = + json_response(&app, Request::get("/healthz").body(Body::empty()).unwrap()).await; + assert_eq!(health["result"]["structuredContent"], native_health); + + for (tool, arguments, uri) in [ + ( + "snapshot", + json!({"graph":"default"}), + "/graphs/default/snapshot".to_string(), + ), + ( + "snapshot", + json!({"graph":"default","branch":"main"}), + "/graphs/default/snapshot?branch=main".to_string(), + ), + ( + "branches_list", + json!({"graph":"default"}), + "/graphs/default/branches".to_string(), + ), + ( + "commits_list", + json!({"graph":"default"}), + "/graphs/default/commits".to_string(), + ), + ( + "commits_get", + json!({"graph":"default","commit_id":commit}), + format!("/graphs/default/commits/{commit}"), + ), + ( + "commits_changes", + json!({"graph":"default","commit_id":commit,"kind":["node"],"op":["insert"],"limit":10}), + format!("/graphs/default/commits/{commit}/changes?kind=node&op=insert&limit=10"), + ), + ( + "changes_poll", + json!({"graph":"default","start":"beginning","type":["Person"]}), + "/graphs/default/changes?start=beginning&type=Person".to_string(), + ), + ] { + let result = call(&app, "secret", tool, arguments).await; + assert_eq!(result["result"]["isError"], false, "{tool}: {result}"); + assert_eq!( + result["result"]["structuredContent"], + http_get(&app, &uri, "secret").await, + "{tool}" + ); + } + // The route's own refusals pass through unchanged. + let refused = call( + &app, + "secret", + "changes_poll", + json!({"graph":"default","start":"now","cursor":"x"}), + ) + .await; + assert_eq!(refused["result"]["isError"], true); + assert_eq!(refused["result"]["structuredContent"]["status"], 400); +} + +#[tokio::test] +async fn mcp_resources_read_graphs_schema_and_branches_under_policy() { + let (_temp, app, _state) = fixture_app().await; + let alice = token("alice", RESOURCE); + let bob = token("bob", RESOURCE); + let (status, listed) = + json_response(&app, rpc(Some(&alice), "resources/list", json!({}))).await; + assert_eq!(status, StatusCode::OK, "{listed}"); + let uris: Vec<&str> = listed["result"]["resources"] + .as_array() + .unwrap() + .iter() + .map(|r| r["uri"].as_str().unwrap()) + .collect(); + assert_eq!(uris, vec!["omnigraph://graphs"]); + let (_, templates) = json_response( + &app, + rpc(Some(&alice), "resources/templates/list", json!({})), + ) + .await; + let templates: Vec<&str> = templates["result"]["resourceTemplates"] + .as_array() + .unwrap() + .iter() + .map(|r| r["uriTemplate"].as_str().unwrap()) + .collect(); + assert_eq!( + templates, + vec![ + "omnigraph://graphs/{graph}/schema", + "omnigraph://graphs/{graph}/branches" + ] + ); + + let read = |token: String, uri: &'static str| { + let app = app.clone(); + async move { + json_response( + &app, + rpc(Some(&token), "resources/read", json!({"uri":uri})), + ) + .await + .1 + } + }; + let schema = read(alice.clone(), "omnigraph://graphs/default/schema").await; + let contents = &schema["result"]["contents"][0]; + assert_eq!(contents["mimeType"], "text/plain", "{schema}"); + let native = http_get(&app, "/graphs/default/schema", &alice).await; + assert_eq!(contents["text"], native["schema_source"]); + let branches = read(alice.clone(), "omnigraph://graphs/default/branches").await; + let text = branches["result"]["contents"][0]["text"].as_str().unwrap(); + assert_eq!( + serde_json::from_str::(text).unwrap(), + http_get(&app, "/graphs/default/branches", &alice).await + ); + let graphs = read(bob.clone(), "omnigraph://graphs").await; + let text = graphs["result"]["contents"][0]["text"].as_str().unwrap(); + assert!(serde_json::from_str::(text).unwrap()["graphs"].is_array()); + + // Policy decides, and an unknown resource is not found. + let denied = read(bob, "omnigraph://graphs/default/schema").await; + assert_eq!(denied["error"]["data"]["status"], 403, "{denied}"); + for uri in [ + "omnigraph://graphs/unknown/schema", + "omnigraph://graphs/default/secrets", + "omnigraph://schema", + ] { + let missing = json_response( + &app, + rpc(Some(&alice), "resources/read", json!({"uri":uri})), + ) + .await + .1; + assert_eq!(missing["error"]["code"], -32002, "{uri}: {missing}"); + } +} + +#[tokio::test] +async fn mcp_write_tools_are_opt_in_conditional_and_policy_gated() { + // Off by default: absent and refused, with nothing written. + let (temp, app) = static_token_app(omnigraph_server::McpSettings::default()).await; + let before = manifest_dataset_version(&graph_path(temp.path())).await; + let names: Vec = tool_list(&app, "secret") + .await + .iter() + .map(|tool| tool["name"].as_str().unwrap().to_owned()) + .collect(); + for write in [ + "mutate", + "load", + "branches_create", + "branches_delete", + "branches_merge", + ] { + assert!(!names.iter().any(|name| name == write), "{write}"); + let refused = call(&app, "secret", write, json!({"graph":"default"})).await; + assert!(refused.get("error").is_some(), "{write}: {refused}"); + } + assert_eq!( + manifest_dataset_version(&graph_path(temp.path())).await, + before + ); + + let (temp, app) = static_token_app(writes(adhoc())).await; + let tools = tool_list(&app, "secret").await; + for (name, destructive, idempotent) in [ + ("mutate", true, false), + ("load", true, false), + ("branches_create", false, false), + ("branches_delete", true, true), + ("branches_merge", true, false), + ] { + let tool = tools.iter().find(|tool| tool["name"] == name).unwrap(); + assert_eq!(tool["annotations"]["readOnlyHint"], false, "{name}"); + assert_eq!( + tool["annotations"]["destructiveHint"], destructive, + "{name}" + ); + assert_eq!(tool["annotations"]["idempotentHint"], idempotent, "{name}"); + assert!( + tool["description"] + .as_str() + .unwrap() + .contains("outcome_unknown") + ); + } + assert!( + instructions(&app, "secret") + .await + .contains("if_graph_commit") + ); + + // A conditional write against a stale head is refused before effects. + let read = call( + &app, + "secret", + "read", + json!({"graph":"default","query":"query n() { match { $p: Person } return { $p.name } }"}), + ) + .await; + let head = read["result"]["structuredContent"]["graph_commit_id"] + .as_str() + .unwrap() + .to_owned(); + let insert = "query add($name: String) { insert Person { name: $name, age: 7 } }"; + let first = call( + &app, + "secret", + "mutate", + json!({"graph":"default","query":insert,"params":{"name":"Eve"},"if_graph_commit":head}), + ) + .await; + assert_eq!(first["result"]["isError"], false, "{first}"); + let after_first = manifest_dataset_version(&graph_path(temp.path())).await; + let stale = call( + &app, + "secret", + "mutate", + json!({"graph":"default","query":insert,"params":{"name":"Frank"},"if_graph_commit":head}), + ) + .await; + assert_eq!(stale["result"]["isError"], true, "{stale}"); + assert_eq!( + stale["result"]["structuredContent"]["status"], 412, + "{stale}" + ); + assert!( + stale["result"]["structuredContent"] + .get("outcome_unknown") + .is_none() + ); + assert_eq!( + manifest_dataset_version(&graph_path(temp.path())).await, + after_first + ); + + // Load, then a branch round trip: create, write, merge, delete. + let loaded = call( + &app, + "secret", + "load", + json!({"graph":"default","mode":"merge","data":"{\"type\":\"Person\",\"data\":{\"name\":\"Gina\",\"age\":50}}"}), + ) + .await; + assert_eq!(loaded["result"]["isError"], false, "{loaded}"); + for (tool, arguments) in [ + ( + "branches_create", + json!({"graph":"default","name":"feature"}), + ), + ( + "mutate", + json!({"graph":"default","query":insert,"params":{"name":"Hal"},"branch":"feature"}), + ), + ( + "branches_merge", + json!({"graph":"default","source":"feature"}), + ), + ( + "branches_delete", + json!({"graph":"default","name":"feature"}), + ), + ] { + let result = call(&app, "secret", tool, arguments).await; + assert_eq!(result["result"]["isError"], false, "{tool}: {result}"); + } + let names = call( + &app, + "secret", + "read", + json!({"graph":"default","query":"query names() { match { $p: Person } return { $p.name } order { $p.name asc } }"}), + ) + .await; + let rows = names["result"]["structuredContent"]["rows"].to_string(); + for name in ["Eve", "Gina", "Hal"] { + assert!(rows.contains(name), "{name}: {rows}"); + } + assert!(!rows.contains("Frank")); + let branches = call(&app, "secret", "branches_list", json!({"graph":"default"})).await; + assert_eq!( + branches["result"]["structuredContent"]["branches"], + json!(["main"]) + ); + let missing = call( + &app, + "secret", + "branches_delete", + json!({"graph":"default","name":"feature"}), + ) + .await; + assert_eq!(missing["result"]["structuredContent"]["status"], 404); + + // Each write keeps its route's policy action: `change` without + // `branch_create` refuses branch creation. + let (_temp, app, _state) = + fixture_app_with(writes(omnigraph_server::McpSettings::default())).await; + let alice = token("alice", RESOURCE); + let bob = token("bob", RESOURCE); + let created = call( + &app, + &alice, + "branches_create", + json!({"graph":"default","name":"x"}), + ) + .await; + assert_eq!( + created["result"]["structuredContent"]["status"], 403, + "{created}" + ); + let denied = call( + &app, + &bob, + "mutate", + json!({"graph":"default","query":insert,"params":{"name":"Ivy"}}), + ) + .await; + assert_eq!( + denied["result"]["structuredContent"]["status"], 403, + "{denied}" + ); + let allowed = call( + &app, + &alice, + "mutate", + json!({"graph":"default","query":insert,"params":{"name":"Ivy"}}), + ) + .await; + assert_eq!(allowed["result"]["isError"], false, "{allowed}"); +} diff --git a/docs/user/operations/mcp.md b/docs/user/operations/mcp.md index cf1671064..5aeca433e 100644 --- a/docs/user/operations/mcp.md +++ b/docs/user/operations/mcp.md @@ -1,8 +1,9 @@ # MCP `omnigraph-server` serves the Model Context Protocol at `/mcp`, so Claude, -ChatGPT, Cursor and other MCP clients can read a graph with a URL and the -caller's own credential. It is the only OmniGraph MCP implementation. +ChatGPT, Cursor and other MCP clients can work with a graph with a URL and the +caller's own credential. It is the only OmniGraph MCP implementation; it +replaces the `@modernrelay/omnigraph-mcp` npm package. Every tool calls the same handler as its HTTP route, under the same actor and the same [policy](policy.md) action. A tool appearing in the list is not @@ -10,10 +11,21 @@ permission to use it: a call the policy refuses returns the route's refusal. ## Enable it -`/mcp` is mounted when the server is started with an +`/mcp` is mounted whenever the server has a bearer credential source: static +tokens, signed data tokens, or an [OIDC resource identity](server.md#oidc-resource-identities-and-mcp) -(`--oidc-identity-trust`). The resource identifier in that file is the URL users -enter in their MCP client, for example `https://graphs.example.com/mcp`. +(`--oidc-identity-trust`). A server without credentials has no MCP. + +| Credentials | `/mcp` | OAuth discovery and challenge | +|---|---|---| +| Static or signed data tokens | mounted | none | +| OIDC resource identity, alone or with tokens | mounted | served | + +With static or signed tokens, configure the client with the URL and an +`Authorization: Bearer ` header; Claude Code and Cursor support this +directly. With OIDC, the resource identifier in the trust file is the URL users +enter in their MCP client, for example `https://graphs.example.com/mcp`, and the +client signs in through OAuth. The endpoint is stateless Streamable HTTP built on the official Rust MCP SDK: each request is a `POST` that carries its own bearer credential and gets a JSON @@ -23,7 +35,7 @@ later unauthenticated request. ### Sign-in and discovery A request without a valid credential is refused with 401 before its body is -read. The response's challenge names the protected-resource metadata: +read. With OIDC, the response's challenge names the protected-resource metadata: ```text WWW-Authenticate: Bearer resource_metadata="https://graphs.example.com/.well-known/oauth-protected-resource/mcp" @@ -41,33 +53,49 @@ proxy rewrite is needed. ### Host and Origin The SDK refuses a request whose `Host` is not allowed with 403. Allowed hosts -are `localhost`, `127.0.0.1`, `::1` and the resource identifier's authority. -A browser `Origin` must be the resource identifier's origin; native clients send -no `Origin`. +are `localhost`, `127.0.0.1`, `::1`, the OIDC resource identifier's authority, +and every `--mcp-allowed-host`. A server reached by another name, for example +behind a load balancer without OIDC, needs that name listed. + +With OIDC, a browser `Origin` must be the resource identifier's origin. Without +OIDC there is no origin to allow, so a request carrying any `Origin` is refused +with 403. Native clients send no `Origin`. ## Settings | Flag | Environment | Default | Effect | |---|---|---|---| | `--mcp-allow-adhoc` | `OMNIGRAPH_MCP_ALLOW_ADHOC=1` | off | Lists and admits the `read` tool | +| `--mcp-allow-writes` | `OMNIGRAPH_MCP_ALLOW_WRITES=1` | off | Lists and admits the write tools | +| `--mcp-allowed-host `, repeatable | `OMNIGRAPH_MCP_ALLOWED_HOSTS=a,b` | none | Adds allowed `Host` values | -The setting is read once at boot. It enables a tool; it grants nothing. When it -is off, `read` is absent from `tools/list` and a call to it is refused as an -unknown tool. +Settings are read once at boot. They enable tools; they grant nothing. A tool +that is off is absent from `tools/list`, and a call to it is refused as an +unknown tool. The opt-ins stay off even when a policy is applied: turn them on +for a deployment whose policy was written with an LLM caller in mind. ## Tools +### Read tools + | Tool | Arguments | Runs | Policy action | |---|---|---|---| -| `graphs` | none | `GET /graphs/discovery` | any authenticated identity | +| `health` | none | `GET /healthz` | none | +| `graphs` | none | `GET /graphs/discovery` for an identity credential; otherwise `GET /graphs`, narrowed to ids and names | identity, or `graph_list` | | `schema` | `graph` | `GET /graphs/{graph}/schema` | `read` | | `queries` | `graph` | `GET /graphs/{graph}/queries`, stored mutations removed | `read` | | `query` | `graph`, `name`, optional `params`, `branch` | `POST /graphs/{graph}/queries/{name}` as a read | `invoke_query`, then `read` | | `read` (opt-in) | `graph`, `query`, optional `name`, `params`, `branch`, `snapshot` | `POST /graphs/{graph}/query` | `read` | +| `snapshot` | `graph`, optional `branch` | `GET /graphs/{graph}/snapshot` | `read` | +| `branches_list` | `graph` | `GET /graphs/{graph}/branches` | `read` | +| `commits_list` | `graph`, optional `branch` | `GET /graphs/{graph}/commits` | `read` | +| `commits_get` | `graph`, `commit_id` | `GET /graphs/{graph}/commits/{commit_id}` | `read` | +| `commits_changes` | `graph`, `commit_id`, optional `kind`, `type`, `op`, `page_token`, `limit` | `GET /graphs/{graph}/commits/{commit_id}/changes` | `read` | +| `changes_poll` | `graph`, optional `branch`, `cursor`, `start`, `kind`, `type`, `op`, `page_token`, `limit` | `GET /graphs/{graph}/changes` | `read` | -All tools are read-only. The server's `initialize` instructions tell a model to -call `graphs`, then `schema`, then a stored read with `query` or, when enabled, -its own GQ read with `read`. +The server's `initialize` instructions tell a model to call `graphs`, then +`schema`, then a stored read with `query` or, when enabled, its own GQ read with +`read`. - `queries` lists each stored read with its typed parameters (`name`, `kind`, `nullable`, plus `item_kind` for lists and `vector_dim` for vectors). @@ -77,11 +105,49 @@ its own GQ read with `read`. `branch merge-base`, `branch merge … dry run` and `explain query …`. It refuses `insert`, `update`, `delete` and branch writes with the route's 400. +### Write tools (opt-in) + +| Tool | Arguments | Runs | Policy action | +|---|---|---|---| +| `mutate` | `graph`, `query`, optional `name`, `params`, `branch`, `if_graph_commit` | `POST /graphs/{graph}/mutate`, or `/mutate/if-graph-commit` with `if_graph_commit` | `change`; a branch statement takes its branch action | +| `load` | `graph`, `mode`, `data`, optional `branch`, `from` | `POST /graphs/{graph}/load` | `change`, plus `branch_create` when `from` creates the branch | +| `branches_create` | `graph`, `name`, optional `from` | `POST /graphs/{graph}/branches` | `branch_create` | +| `branches_delete` | `graph`, `name` | `DELETE /graphs/{graph}/branches/{name}` | `branch_delete` | +| `branches_merge` | `graph`, `source`, optional `target` | `POST /graphs/{graph}/branches/merge` | `branch_merge` | + +Write tools follow one contract: + +- **Conditional writes.** Pass a read's `graph_commit_id` to `mutate` as + `if_graph_commit`. If the branch has moved, the call fails with 412 and nothing + is written. +- **No retries.** The server never replays a tool call. +- **Unknown outcomes are reported.** A write runs to completion even when the + caller stops waiting. A write still running after 60 seconds, a cancelled write, + or a write whose completion the server cannot confirm returns `isError` with + `"outcome_unknown": true`. Read the target and reconcile before replaying it. +- **Size.** The whole call, including `load` data, fits in the 64 KiB request. + Use `POST /load/ndjson` for larger loads. + +Schema changes and cluster deployments are not MCP tools; operators run them with +`omnigraph cluster plan` and `apply`. + +## Resources + +| URI | Content | +|---|---| +| `omnigraph://graphs` | The `graphs` result, JSON, under the same rule | +| `omnigraph://graphs/{graph}/schema` (template) | The graph's `.pg` source | +| `omnigraph://graphs/{graph}/branches` (template) | The `branches_list` result, JSON | + +A resource read runs the same handler and policy action as its tool. A refusal +is a JSON-RPC error whose `data` carries the HTTP `status` and `error` body; an +unknown resource or graph is error `-32002`. + ## Results and errors A successful call returns the route's JSON response body as structured content, -plus a text copy for older clients. `graphs`, `schema` and `queries` declare an -output schema. +plus a text copy for older clients. `health`, `graphs`, `schema`, `queries` and +`branches_list` declare an output schema. A refused call returns `isError: true` with one of two shapes: @@ -91,6 +157,8 @@ A refused call returns `isError: true` with one of two shapes: - `{"code": "...", "message": "..."}` when the MCP layer refused it: `capacity_exceeded`, `cancelled`, `deadline_exceeded` or `result_too_large`. +A write adds `"outcome_unknown": true` to either shape when it may have happened. + ## Limits | Limit | Value | @@ -98,6 +166,7 @@ A refused call returns `isError: true` with one of two shapes: | Request body | 64 KiB | | Complete tool result, including the text copy | 1 MiB | | Wait for one read | 30 seconds | +| Wait for one write | 60 seconds; the write continues past it | | Concurrent tool executions | 16 | A cancelled or expired call stops the caller's wait. The execution keeps its diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 209f9a549..bfa70deaa 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -187,8 +187,8 @@ prevent new OIDC access; accepted operations can finish. Admission refresh does not restart writers. Graph configuration retains its normal activation. With this profile configured, the server also serves public -protected-resource metadata and an OAuth challenge on `/mcp`, so standard OAuth -clients such as Claude can sign in. The [MCP guide](mcp.md) describes the +protected-resource metadata and an OAuth challenge on its [`/mcp`](mcp.md) +endpoint, so standard OAuth clients such as Claude can sign in. The [MCP guide](mcp.md) describes the endpoint, its tools, settings and limits. ## Route families @@ -203,7 +203,7 @@ endpoint, its tools, settings and limits. | `POST /cluster/deployments` | Submit an exact-ID deployment: graph creation/deletion or schema, query, policy, provider and Blob-binding changes | | `GET /cluster/deployments`, `GET /cluster/deployments/{id}` | Authorized deployment status and current-process activation observation | | `GET /.well-known/oauth-protected-resource`, also at the resource's path suffix | Public OIDC resource metadata; only when OIDC trust is configured | -| `/mcp` | The [MCP endpoint](mcp.md); only when OIDC trust is configured | +| `/mcp` | The [MCP endpoint](mcp.md); when static tokens, signed data tokens or OIDC trust are configured | | `/graphs/{id}/query`, `/mutate` | Run inline GQ source | | `/graphs/{id}/mutate/if-graph-commit` | Run an inline conditional mutation | | `/graphs/{id}/queries` | List and invoke stored queries, including conditional mutations |