crates/gunnar-mcp/src/rpc.rs
//! **`POST /mcp`**: JSON-RPC 2.0 over MCP's streamable HTTP transport.
//!
//! Every request needs a bearer token this server issued ([`crate::oauth`]);
//! without one the answer is 401 with the RFC 9728 `resource_metadata`
//! pointer, which is how claude.ai and Claude Code find the login. Each
//! connection has a call budget (429 with `Retry-After` beyond it). The
//! answer to a request is always `application/json`; this server opens no
//! event stream (`GET /mcp` is 405), keeps no session, and refuses batches.
use std::sync::Arc;
use std::time::SystemTime;
use axum::extract::{Request, State};
use axum::response::{IntoResponse, Response};
use axum::Json;
use http::{HeaderMap, HeaderValue, StatusCode};
use serde_json::{json, Value};
use crate::tools::{self, Scope};
use crate::{digest, now_unix, Audit, Backend, Failure, Mcp, HIDDEN_WORDS, PROTOCOL_VERSIONS};
/// The most bytes one JSON-RPC message may carry.
const MESSAGE_MAX: usize = 256 * 1024;
/// Origins a browser-based client may send. Server-side clients (claude.ai's
/// connector, Claude Code) send none.
const BROWSER_ORIGINS: &[&str] = &["https://claude.ai", "https://claude.com"];
/// `GET`/`DELETE /mcp`: no event stream, no session to end.
pub async fn not_here() -> Response {
(
StatusCode::METHOD_NOT_ALLOWED,
[(http::header::ALLOW, "POST")],
"This MCP server answers POST only; it opens no event stream.",
)
.into_response()
}
fn unauthorized<B: Backend>(mcp: &Mcp<B>, error: Option<&str>) -> Response {
let meta = format!(
"{}/.well-known/oauth-protected-resource/mcp",
mcp.config().issuer
);
let mut challenge = format!(
"Bearer resource_metadata=\"{meta}\", scope=\"{}\"",
tools::SCOPE_READ
);
if let Some(e) = error {
challenge.push_str(&format!(", error=\"{e}\""));
}
let mut r = (StatusCode::UNAUTHORIZED, Json(json!({ "error": error.unwrap_or("unauthorized"), "error_description": "Connect to Gunnar first: this server needs a bearer token it issued." }))).into_response();
if let Ok(v) = HeaderValue::from_str(&challenge) {
r.headers_mut().insert(http::header::WWW_AUTHENTICATE, v);
}
r
}
/// The connection a bearer token names, its budget taken. `Err` is the
/// response to send instead.
struct Bound<A> {
grant: u64,
scope: Scope,
actor: String,
user: String,
agent: A,
}
fn bind<B: Backend>(mcp: &Mcp<B>, headers: &HeaderMap) -> Result<Bound<B::Agent>, Response> {
let token = headers
.get(http::header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| {
v.strip_prefix("Bearer ")
.or_else(|| v.strip_prefix("bearer "))
})
.map(str::trim);
let Some(token) = token.filter(|t| !t.is_empty()) else {
return Err(unauthorized(mcp, None));
};
let now = SystemTime::now();
let mut st = mcp.lock();
let Some(access) = st.access.get(&digest(token)).cloned() else {
return Err(unauthorized(mcp, Some("invalid_token")));
};
if access.expires <= now {
st.access.remove(&digest(token));
return Err(unauthorized(mcp, Some("invalid_token")));
}
let Some(grant) = st.grants.get_mut(&access.grant) else {
return Err(unauthorized(mcp, Some("invalid_token")));
};
if grant.expires <= now {
return Err(unauthorized(mcp, Some("invalid_token")));
}
if let Err(wait) = grant.bucket.take() {
let actor = grant.actor.clone();
let id = grant.id;
drop(st);
mcp.backend.audit(&Audit {
event: "mcp.refused",
actor,
tool: String::new(),
outcome: format!("rate limited; retry after {wait} s"),
grant: id,
});
let mut r = (StatusCode::TOO_MANY_REQUESTS, Json(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32000, "message": format!("Too many calls on this connection; try again in {wait} seconds.") } }))).into_response();
r.headers_mut()
.insert(http::header::RETRY_AFTER, HeaderValue::from(wait));
return Err(r);
}
grant.last_used_unix = Some(now_unix());
Ok(Bound {
grant: grant.id,
scope: grant.scope,
actor: grant.actor.clone(),
user: grant.user.clone(),
agent: grant.agent.clone(),
})
}
fn reply(id: &Value, result: Value) -> Response {
Json(json!({ "jsonrpc": "2.0", "id": id, "result": result })).into_response()
}
fn fault(status: StatusCode, id: &Value, code: i64, message: &str) -> Response {
(
status,
Json(json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })),
)
.into_response()
}
/// `POST /mcp`.
pub async fn post<B: Backend>(State(mcp): State<Arc<Mcp<B>>>, request: Request) -> Response {
let (parts, body) = request.into_parts();
if let Some(origin) = parts
.headers
.get(http::header::ORIGIN)
.and_then(|v| v.to_str().ok())
{
if origin != mcp.config().issuer && !BROWSER_ORIGINS.contains(&origin) {
return (
StatusCode::FORBIDDEN,
"This origin may not call this MCP server.",
)
.into_response();
}
}
let bound = match bind(&mcp, &parts.headers) {
Ok(b) => b,
Err(r) => return r,
};
if let Some(v) = parts
.headers
.get("mcp-protocol-version")
.and_then(|v| v.to_str().ok())
{
if !PROTOCOL_VERSIONS.contains(&v) {
return fault(
StatusCode::BAD_REQUEST,
&Value::Null,
-32600,
&format!(
"Unsupported MCP-Protocol-Version {:?}; this server speaks {}",
tools::said_name(v),
PROTOCOL_VERSIONS.join(", ")
),
);
}
}
let Ok(bytes) = axum::body::to_bytes(body, MESSAGE_MAX).await else {
return fault(
StatusCode::PAYLOAD_TOO_LARGE,
&Value::Null,
-32600,
"The message is too large.",
);
};
let message: Value = match serde_json::from_slice(&bytes) {
Ok(v) => v,
Err(_) => {
return fault(
StatusCode::BAD_REQUEST,
&Value::Null,
-32700,
"Parse error: the body is not JSON.",
)
}
};
if message.is_array() {
return fault(
StatusCode::BAD_REQUEST,
&Value::Null,
-32600,
"Batches are not supported; send one message per request.",
);
}
let Some(obj) = message.as_object() else {
return fault(
StatusCode::BAD_REQUEST,
&Value::Null,
-32600,
"Invalid request: not a JSON-RPC object.",
);
};
let id = obj.get("id").cloned();
let method = obj.get("method").and_then(Value::as_str);
// A notification, or a response to something we never asked: accepted.
let (Some(id), Some(method)) = (id, method) else {
return StatusCode::ACCEPTED.into_response();
};
let params = obj.get("params");
match method {
"initialize" => {
let asked = params
.and_then(|p| p.get("protocolVersion"))
.and_then(Value::as_str)
.unwrap_or("");
let version = if PROTOCOL_VERSIONS.contains(&asked) {
asked
} else {
PROTOCOL_VERSIONS[0]
};
mcp.backend.audit(&Audit {
event: "mcp.session",
actor: bound.actor.clone(),
tool: String::new(),
outcome: format!("initialized ({version})"),
grant: bound.grant,
});
reply(
&id,
json!({
"protocolVersion": version,
"capabilities": { "tools": { "listChanged": false } },
"serverInfo": { "name": "gunnar", "title": "Gunnar", "version": env!("CARGO_PKG_VERSION") },
"instructions": "Read-only access to the Gunnar repositories you can see. Start with list_repositories, then list_commits (each commit names its tree), browse_tree and read_file by oid. Nothing here takes an address.",
}),
)
}
"ping" => reply(&id, json!({})),
"tools/list" => reply(&id, tools::listed(bound.scope)),
"tools/call" => call(&mcp, &bound, &id, params).await,
_ => fault(
StatusCode::OK,
&id,
-32601,
&format!("Method not found: {}", tools::said_name(method)),
),
}
}
async fn call<B: Backend>(
mcp: &Mcp<B>,
bound: &Bound<B::Agent>,
id: &Value,
params: Option<&Value>,
) -> Response {
let name = params
.and_then(|p| p.get("name"))
.and_then(Value::as_str)
.unwrap_or("");
let audit = |tool: &str, outcome: String| {
mcp.backend.audit(&Audit {
event: "mcp.call",
actor: bound.actor.clone(),
tool: tools::said_name(tool),
outcome,
grant: bound.grant,
})
};
let parsed = match tools::parse(name, params.and_then(|p| p.get("arguments")), bound.scope) {
Ok(c) => c,
Err(refused @ (tools::Refused::Unknown(_) | tools::Refused::NeedsWrite(_))) => {
audit(
name,
format!(
"refused: {}",
match refused {
tools::Refused::Unknown(_) => "not on the allowlist",
_ => "needs the write grant",
}
),
);
return fault(StatusCode::OK, id, -32602, &refused.words());
}
Err(refused) => {
audit(name, "refused: arguments".to_owned());
return text(id, &refused.words(), true);
}
};
let answered = tokio::time::timeout(
mcp.config().call_timeout,
mcp.backend.call(&bound.agent, &bound.user, &parsed),
)
.await;
match answered {
Ok(Ok(answer)) => {
audit(name, "ok".to_owned());
text(id, &tools::render(&answer), false)
}
Ok(Err(Failure::KeyGone)) => {
audit(
name,
"refused: the agent's key is gone at gunnar".to_owned(),
);
mcp.end_grant(bound.grant, "its key is gone at gunnar")
.await;
unauthorized(mcp, Some("invalid_token"))
}
Ok(Err(failure)) => {
let (outcome, words) = match failure {
Failure::Hidden => ("hidden", HIDDEN_WORDS.to_owned()),
Failure::Refused(w) => ("refused by gunnar", w),
Failure::Busy(w) => ("busy", format!("Gunnar is busy; try again shortly. {w}")),
Failure::Unavailable(w) => ("unavailable", format!("Gunnar did not answer. {w}")),
Failure::KeyGone => unreachable!("handled above"),
};
audit(name, outcome.to_owned());
text(id, &tools::redact_emails(&words), true)
}
Err(_) => {
audit(name, "timed out".to_owned());
text(
id,
"Gunnar did not answer in time; try a narrower call.",
true,
)
}
}
}
fn text(id: &Value, words: &str, is_error: bool) -> Response {
reply(
id,
json!({ "content": [{ "type": "text", "text": words }], "isError": is_error }),
)
}