feat: Support sigv4 auth mechanism for MCP servers.

Cargo.toml

@@ -21,7 +21,6 @@ aws-sdk-bedrockruntime = { version = "1.133.0", default-features = false, featur
 # Capped below 1.5.0: it makes DeferredSignerSender::send generic, which breaks
 # type inference in aws-runtime 1.7.4 (latest). Lift once aws-runtime ships a fix.
 aws-smithy-types = ">=1.4.9, <1.9.0"
-aws-smithy-runtime-api = "1.14.0"
 aws-credential-types = "1.3.0"
 aws-sigv4 = "1.5.1"
 
@@ -51,6 +50,7 @@ tokio-tungstenite = { version = "0.30.0", features = ["rustls-tls-webpki-roots"]
 
 # MCP and API clients
 rmcp = { version = "^3.5.1", default-features = false }
+sse-stream = "0.2.6"
 async-openai = { version = "^0.42.0", features = ["byot", "chat-completion", "responses"] }
 reqwest = { version = "^0.13.4", default-features = false, features = ["json", "query", "rustls", "http2", "stream"] }
 oauth2 = "5.0"

crates/aether-auth/Cargo.toml

@@ -2,7 +2,7 @@
 name = "aether-auth"
 version = "0.1.17"
 edition = "2024"
-description = "OAuth credential storage and authorization flows for the Aether AI agent framework"
+description = "OAuth credential storage, authorization flows and AWS SigV4 signing for the Aether AI agent framework"
 license = "MIT"
 repository = "https://github.com/contextbridge/aether"
 readme = "README.md"
@@ -22,6 +22,7 @@ keyring = [
     "dep:dbus-secret-service-keyring-store",
 ]
 mcp = ["dep:rmcp"]
+sigv4 = ["dep:aws-config", "dep:aws-credential-types", "dep:aws-sigv4"]
 
 [dependencies]
 age = { workspace = true }
@@ -43,6 +44,11 @@ oauth2 = { workspace = true }
 reqwest = { workspace = true }
 url = { workspace = true }
 
+# sigv4 feature
+aws-config = { workspace = true, optional = true }
+aws-credential-types = { workspace = true, optional = true }
+aws-sigv4 = { workspace = true, optional = true }
+
 [target.'cfg(target_os = "macos")'.dependencies]
 apple-native-keyring-store = { workspace = true, optional = true }
 

crates/aether-auth/src/lib.rs

@@ -25,3 +25,8 @@ pub mod mcp;
 pub use mcp::{
     McpCredentialStore, OAuthClientRegistration, OAuthFlowOptions, create_auth_manager_from_store, perform_oauth_flow,
 };
+
+#[cfg(feature = "sigv4")]
+mod sigv4;
+#[cfg(feature = "sigv4")]
+pub use sigv4::{SigV4Config, SigV4Credentials, SigV4Error, SigV4Signer};

crates/aether-auth/src/sigv4.rs

@@ -0,0 +1,129 @@
+use aws_config::Region;
+use aws_config::default_provider::credentials::DefaultCredentialsChain;
+use aws_credential_types::Credentials;
+use aws_credential_types::provider::{ProvideCredentials, SharedCredentialsProvider};
+use aws_sigv4::http_request::{SignableBody, SignableRequest, SigningSettings, sign};
+use aws_sigv4::sign::v4;
+use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
+use std::error::Error;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime};
+use tokio::sync::{Mutex, OnceCell};
+
+#[derive(Debug, Clone)]
+pub struct SigV4Signer {
+    state: Arc<SignerState>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SigV4Config {
+    pub service: String,
+    pub region: String,
+    pub credentials: SigV4Credentials,
+    pub refresh_before_expiry: Duration,
+}
+
+#[derive(Debug, Clone)]
+pub enum SigV4Credentials {
+    DefaultChain { profile: Option<String> },
+    Provider(SharedCredentialsProvider),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SigV4Error {
+    #[error("Could not resolve AWS credentials: {0}")]
+    Credentials(String),
+
+    #[error("SigV4 signing failed: {0}")]
+    Signing(String),
+}
+
+impl SigV4Signer {
+    pub fn new(config: SigV4Config) -> Self {
+        Self { state: Arc::new(SignerState { config, default_chain: OnceCell::new(), cached: Mutex::default() }) }
+    }
+
+    pub fn config(&self) -> &SigV4Config {
+        &self.state.config
+    }
+
+    pub async fn headers(&self, method: &str, url: &str, body: &[u8]) -> Result<HeaderMap, SigV4Error> {
+        let identity = self.credentials().await?.into();
+        let params = v4::SigningParams::builder()
+            .identity(&identity)
+            .region(&self.config().region)
+            .name(&self.config().service)
+            .time(SystemTime::now())
+            .settings(SigningSettings::default())
+            .build()
+            .map_err(signing_error)?
+            .into();
+
+        let signable =
+            SignableRequest::new(method, url, std::iter::empty(), SignableBody::Bytes(body)).map_err(signing_error)?;
+
+        let (instructions, _signature) = sign(signable, &params).map_err(signing_error)?.into_parts();
+        instructions.headers().map(|(name, value)| sensitive_header(name, value)).collect()
+    }
+
+    async fn credentials(&self) -> Result<Credentials, SigV4Error> {
+        let mut cached = self.state.cached.lock().await;
+        if let Some(credentials) = cached.as_ref().filter(|credentials| self.is_fresh(credentials)) {
+            return Ok(credentials.clone());
+        }
+        let credentials = self
+            .provider()
+            .await
+            .provide_credentials()
+            .await
+            .map_err(|error| SigV4Error::Credentials(error_chain(&error)))?;
+
+        *cached = Some(credentials.clone());
+        Ok(credentials)
+    }
+
+    async fn provider(&self) -> &SharedCredentialsProvider {
+        let config = self.config();
+        match &config.credentials {
+            SigV4Credentials::Provider(provider) => provider,
+            SigV4Credentials::DefaultChain { profile } => {
+                self.state.default_chain.get_or_init(|| default_chain(&config.region, profile.as_deref())).await
+            }
+        }
+    }
+
+    fn is_fresh(&self, credentials: &Credentials) -> bool {
+        credentials.expiry().is_none_or(|expiry| expiry > SystemTime::now() + self.config().refresh_before_expiry)
+    }
+}
+
+#[derive(Debug)]
+struct SignerState {
+    config: SigV4Config,
+    default_chain: OnceCell<SharedCredentialsProvider>,
+    cached: Mutex<Option<Credentials>>,
+}
+
+async fn default_chain(region: &str, profile: Option<&str>) -> SharedCredentialsProvider {
+    let chain = DefaultCredentialsChain::builder().region(Region::new(region.to_string()));
+    let chain = match profile {
+        Some(profile) => chain.profile_name(profile),
+        None => chain,
+    };
+    SharedCredentialsProvider::new(chain.build().await)
+}
+
+fn sensitive_header(name: &str, value: &str) -> Result<(HeaderName, HeaderValue), SigV4Error> {
+    let invalid = || SigV4Error::Signing(format!("invalid signed header `{name}`"));
+    let mut value = HeaderValue::from_str(value).map_err(|_| invalid())?;
+    value.set_sensitive(true);
+    Ok((HeaderName::try_from(name).map_err(|_| invalid())?, value))
+}
+
+fn signing_error(error: impl Error + 'static) -> SigV4Error {
+    SigV4Error::Signing(error_chain(&error))
+}
+
+fn error_chain(error: &(dyn Error + 'static)) -> String {
+    std::iter::successors(Some(error), |&error| error.source()).map(ToString::to_string).collect::<Vec<_>>().join(": ")
+}

crates/aether-auth/tests/sigv4.rs

@@ -0,0 +1,92 @@
+#![cfg(feature = "sigv4")]
+
+use aether_auth::{SigV4Config, SigV4Credentials, SigV4Error, SigV4Signer};
+use aws_credential_types::Credentials;
+use aws_credential_types::credential_fn::provide_credentials_fn;
+use aws_credential_types::provider::error::CredentialsError;
+use aws_credential_types::provider::{ProvideCredentials, SharedCredentialsProvider};
+use reqwest::header::HeaderValue;
+use std::future::ready;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, SystemTime};
+
+#[tokio::test]
+async fn credentials_are_reused_while_they_are_fresh() {
+    let signer = signer(rotating(Duration::from_hours(1)));
+    let first = signing_key(&signer).await;
+    let second = signing_key(&signer).await;
+    assert_eq!([first, second], ["AKID1", "AKID1"]);
+}
+
+#[tokio::test]
+async fn credentials_expiring_within_the_refresh_window_are_refreshed() {
+    let signer = SigV4Signer::new(SigV4Config {
+        refresh_before_expiry: Duration::from_hours(2),
+        ..config(rotating(Duration::from_hours(1)))
+    });
+
+    let first = signing_key(&signer).await;
+    let second = signing_key(&signer).await;
+    assert_eq!([first, second], ["AKID1", "AKID2"]);
+}
+
+#[tokio::test]
+async fn clones_share_resolved_credentials() {
+    let signer = signer(rotating(Duration::from_hours(1)));
+    let original = signing_key(&signer).await;
+    let clone = signing_key(&signer.clone()).await;
+    assert_eq!([original, clone], ["AKID1", "AKID1"]);
+}
+
+#[tokio::test]
+async fn signed_headers_carry_the_session_token_and_are_sensitive() {
+    let credentials = Credentials::new("AKID", "secret", Some("session".to_string()), None, "test");
+    let signer = signer(credentials);
+    let headers = signer.headers("POST", "https://example.com/mcp", b"{}").await.unwrap();
+
+    assert_eq!(headers["x-amz-security-token"], "session");
+    assert!(headers.contains_key("x-amz-date"));
+    assert!(headers.values().all(HeaderValue::is_sensitive), "{headers:?}");
+}
+
+#[tokio::test]
+async fn credential_failures_explain_why() {
+    let signer = signer(missing("the SSO session has expired"));
+    let error = signer.headers("GET", "https://example.com/mcp", &[]).await.unwrap_err();
+
+    assert!(matches!(error, SigV4Error::Credentials(_)));
+    assert!(error.to_string().contains("the SSO session has expired"), "{error}");
+}
+
+fn rotating(lifetime: Duration) -> impl ProvideCredentials + 'static {
+    let issued = AtomicUsize::new(0);
+    provide_credentials_fn(move || {
+        let issued = issued.fetch_add(1, Ordering::SeqCst) + 1;
+        let expiry = SystemTime::now() + lifetime;
+        ready(Ok(Credentials::new(format!("AKID{issued}"), "secret", None, Some(expiry), "sts")))
+    })
+}
+
+fn missing(reason: &'static str) -> impl ProvideCredentials + 'static {
+    provide_credentials_fn(move || ready(Err(CredentialsError::provider_error(reason))))
+}
+
+fn signer(credentials: impl ProvideCredentials + 'static) -> SigV4Signer {
+    SigV4Signer::new(config(credentials))
+}
+
+fn config(credentials: impl ProvideCredentials + 'static) -> SigV4Config {
+    SigV4Config {
+        service: "execute-api".to_string(),
+        region: "us-east-1".to_string(),
+        credentials: SigV4Credentials::Provider(SharedCredentialsProvider::new(credentials)),
+        refresh_before_expiry: Duration::from_mins(5),
+    }
+}
+
+async fn signing_key(signer: &SigV4Signer) -> String {
+    let headers = signer.headers("POST", "https://example.com/mcp", b"{}").await.unwrap();
+    let authorization = headers["authorization"].to_str().unwrap();
+    let credential = authorization.split("Credential=").nth(1).expect("a credential scope");
+    credential.split('/').next().unwrap().to_string()
+}

crates/aether-cli/src/acp/protocol/mcp.rs

@@ -28,7 +28,7 @@ fn try_map_mcp_server(server: McpServer, cwd: &Path) -> Option<ServerSpec> {
                 cwd: Some(cwd.to_path_buf()),
             },
         ),
-        Http(http) => (http.name, http_transport(http.url, header_map(&http.headers)?, None)),
+        Http(http) => (http.name, http_transport(http.url, header_map(&http.headers)?)),
         _ => return None,
     };
     Some(ServerSpec { name, transport })
@@ -42,6 +42,7 @@ fn header_map(headers: &[HttpHeader]) -> Option<HeaderMap> {
 mod tests {
     use super::*;
     use agent_client_protocol::schema::v2 as acp;
+    use mcp_utils::client::HttpAuth;
 
     fn map(servers: Vec<acp::McpServer>) -> Vec<ServerSpec> {
         map_acp_mcp_servers(servers, Path::new("/workspace"))
@@ -81,11 +82,11 @@ mod tests {
         assert_eq!(configs.len(), 1);
 
         match &configs[0].transport {
-            Transport::Http { url, headers, oauth } => {
+            Transport::Http { url, headers, auth } => {
                 assert_eq!(configs[0].name, "http-server");
                 assert_eq!(url, "https://example.com/mcp");
                 assert_eq!(headers["authorization"], "Bearer token123");
-                assert!(oauth.is_none(), "an Authorization header replaces OAuth");
+                assert!(matches!(auth, HttpAuth::None), "an Authorization header replaces OAuth");
             }
             other => panic!("Expected Http, got {other:?}"),
         }
@@ -129,9 +130,9 @@ mod tests {
         let configs = map(servers);
         let Transport::Stdio { args, env, .. } = &configs[0].transport else { panic!("expected stdio") };
         assert!(args.is_empty() && env.is_empty());
-        let Transport::Http { headers, oauth, .. } = &configs[1].transport else { panic!("expected http") };
+        let Transport::Http { headers, auth, .. } = &configs[1].transport else { panic!("expected http") };
         assert!(headers.is_empty());
-        assert!(oauth.is_some(), "servers without credentials can sign in with OAuth");
+        assert!(matches!(auth, HttpAuth::OAuth(_)), "servers without credentials can sign in with OAuth");
     }
 
     #[test]

crates/aether-core/Cargo.toml

@@ -40,7 +40,7 @@ tracing = { workspace = true }
 thiserror = { workspace = true }
 glob = { workspace = true }
 http = { workspace = true }
-aether-auth = { path = "../aether-auth", version = "0.1.17" }
+aether-auth = { path = "../aether-auth", version = "0.1.17", features = ["sigv4"] }
 mcp_utils = { package = "aether-mcp-utils", path = "../mcp-utils", version = "0.7.13" }
 llm = { package = "aether-llm", path = "../llm", version = "0.9.13", features = ["codex"] }
 utils = { package = "aether-utils", path = "../utils", version = "0.2.19" }

crates/aether-core/src/docs/mcp_server_config.md

@@ -35,8 +35,18 @@ The `oauth` object is optional. Set `clientMetadataUrl` for a custom CIMD, or
 `clientId` for a pre-registered public client; a configured `clientId` takes
 priority. `callbackPort` defaults to 3118 and must exactly match the client's
 registered redirect URI. If the authorization server does not advertise CIMD,
-Aether falls back to deprecated Dynamic Client Registration. An explicit
-`Authorization` header bypasses OAuth entirely.
+Aether falls back to deprecated Dynamic Client Registration. An
+`Authorization` header on its own turns off OAuth. 
+
+A server can configure one of `oauth`, `sigv4`, or an `Authorization` header. For example:
+
+```json
+{
+  "type": "http",
+  "url": "https://abc123.execute-api.us-east-1.amazonaws.com/mcp",
+  "sigv4": { "service": "execute-api", "region": "us-east-1", "profile": "dev" }
+}
+```
 
 A remote server using a bearer token: