diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock
index ed4ae4e3353..99a6133b4f1 100644
--- a/litellm-rust/Cargo.lock
+++ b/litellm-rust/Cargo.lock
@@ -2464,6 +2464,22 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "litellm-cache-gcs"
+version = "0.1.0"
+dependencies = [
+ "futures-util",
+ "litellm-auth-gcp",
+ "litellm-auth-types",
+ "litellm-cache",
+ "percent-encoding",
+ "reqwest 0.12.28",
+ "serde_json",
+ "tokio",
+ "url",
+ "wiremock",
+]
+
 [[package]]
 name = "litellm-cache-memory"
 version = "0.1.0"
@@ -2666,6 +2682,7 @@ dependencies = [
  "litellm-auth",
  "litellm-auth-gcp",
  "litellm-cache",
+ "litellm-cache-gcs",
  "litellm-cache-memory",
  "litellm-cache-redis",
  "litellm-cache-response",
diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml
index 570d0dd3568..80425d567c0 100644
--- a/litellm-rust/Cargo.toml
+++ b/litellm-rust/Cargo.toml
@@ -29,6 +29,7 @@ litellm-core-utils = { path = "crates/core-utils" }
 litellm-cache = { path = "crates/cache" }
 litellm-cache-memory = { path = "crates/cache-memory" }
 litellm-cache-redis = { path = "crates/cache-redis" }
+litellm-cache-gcs = { path = "crates/cache-gcs" }
 litellm-cache-response = { path = "crates/cache-response" }
 litellm-token-counter = { path = "crates/token-counter" }
 litellm-token-counter-fast = { path = "crates/token-counter-fast" }
@@ -66,6 +67,7 @@ base64 = "0.22"
 moka = { version = "0.12.16", features = ["future"] }
 strum = { version = "0.28.0", features = ["derive"] }
 url = "2.5.8"
+percent-encoding = "2.3"
 webpki-roots = "1"
 time = { version = "0.3.53", features = ["parsing"] }
 criterion = "0.8.2"
diff --git a/litellm-rust/crates/auth-gcp/src/lib.rs b/litellm-rust/crates/auth-gcp/src/lib.rs
index 8aeddae9efc..534d85acdb0 100644
--- a/litellm-rust/crates/auth-gcp/src/lib.rs
+++ b/litellm-rust/crates/auth-gcp/src/lib.rs
@@ -128,6 +128,14 @@ impl VertexAuth {
         }
     }
 
+    pub async fn access_token(
+        &self,
+        config: &VertexConfig,
+        env_lookup: &(dyn Fn(&str) -> Option<String> + Sync),
+    ) -> Result<String, Error> {
+        self.load_provider(config, env_lookup).await?.token().await
+    }
+
     pub async fn validate_environment(
         &self,
         headers: Vec<(String, String)>,
diff --git a/litellm-rust/crates/cache-gcs/Cargo.toml b/litellm-rust/crates/cache-gcs/Cargo.toml
new file mode 100644
index 00000000000..3890e60843c
--- /dev/null
+++ b/litellm-rust/crates/cache-gcs/Cargo.toml
@@ -0,0 +1,21 @@
+[package]
+name = "litellm-cache-gcs"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+futures-util.workspace = true
+litellm-auth-gcp.workspace = true
+litellm-auth-types.workspace = true
+litellm-cache.workspace = true
+percent-encoding.workspace = true
+reqwest.workspace = true
+tokio.workspace = true
+url.workspace = true
+
+[dev-dependencies]
+serde_json.workspace = true
+tokio.workspace = true
+wiremock = "0.6.5"
diff --git a/litellm-rust/crates/cache-gcs/src/cache.rs b/litellm-rust/crates/cache-gcs/src/cache.rs
new file mode 100644
index 00000000000..1097f3252b1
--- /dev/null
+++ b/litellm-rust/crates/cache-gcs/src/cache.rs
@@ -0,0 +1,285 @@
+use std::{future::Future, sync::Arc, time::Duration};
+
+use futures_util::future::try_join_all;
+use litellm_cache::{
+    BaseCache, BatchCache, BatchEntry, CacheCodec, CacheConnectionResult, Error, ExactCacheContext,
+    FlushCache,
+};
+use percent_encoding::{AsciiSet, CONTROLS, percent_encode};
+use reqwest::Client;
+
+use crate::{GcpTokenSource, TokenSource};
+
+pub const DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com";
+
+const OBJECT_NAME_ENCODE_SET: &AsciiSet = &CONTROLS
+    .add(b' ')
+    .add(b'!')
+    .add(b'"')
+    .add(b'#')
+    .add(b'$')
+    .add(b'%')
+    .add(b'&')
+    .add(b'\'')
+    .add(b'(')
+    .add(b')')
+    .add(b'*')
+    .add(b'+')
+    .add(b',')
+    .add(b'/')
+    .add(b':')
+    .add(b';')
+    .add(b'<')
+    .add(b'=')
+    .add(b'>')
+    .add(b'?')
+    .add(b'@')
+    .add(b'[')
+    .add(b'\\')
+    .add(b']')
+    .add(b'^')
+    .add(b'`')
+    .add(b'{')
+    .add(b'|')
+    .add(b'}');
+
+pub fn key_prefix(gcs_path: Option<&str>) -> String {
+    match gcs_path {
+        Some(path) if !path.is_empty() => format!("{}/", path.trim_end_matches('/')),
+        _ => String::new(),
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct GcsConfig {
+    pub bucket_name: String,
+    pub gcs_path: Option<String>,
+    pub path_service_account: Option<String>,
+    pub endpoint: String,
+}
+
+impl GcsConfig {
+    pub fn new(bucket_name: impl Into<String>) -> Self {
+        Self {
+            bucket_name: bucket_name.into(),
+            gcs_path: None,
+            path_service_account: None,
+            endpoint: DEFAULT_ENDPOINT.to_string(),
+        }
+    }
+}
+
+pub struct GcsCache<S: CacheCodec> {
+    config: GcsConfig,
+    key_prefix: String,
+    client: Client,
+    token: Arc<dyn TokenSource>,
+    codec: S,
+}
+
+impl<S: CacheCodec> GcsCache<S> {
+    pub fn new(config: GcsConfig, codec: S) -> Result<Self, Error> {
+        let token = Arc::new(GcpTokenSource::new(config.path_service_account.clone()));
+        Self::with_token_source(config, codec, token)
+    }
+
+    pub fn with_token_source(
+        config: GcsConfig,
+        codec: S,
+        token: Arc<dyn TokenSource>,
+    ) -> Result<Self, Error> {
+        let client = Client::builder().build().map_err(|_| Error::Unavailable)?;
+        let key_prefix = key_prefix(config.gcs_path.as_deref());
+        Ok(Self {
+            config,
+            key_prefix,
+            client,
+            token,
+            codec,
+        })
+    }
+
+    pub fn bucket_name(&self) -> &str {
+        &self.config.bucket_name
+    }
+
+    pub fn key_prefix(&self) -> &str {
+        &self.key_prefix
+    }
+
+    pub fn path_service_account(&self) -> Option<&str> {
+        self.config.path_service_account.as_deref()
+    }
+
+    pub fn object_name(&self, key: &str) -> String {
+        format!("{}{}", self.key_prefix, key)
+    }
+
+    fn encoded_object_name(&self, key: &str) -> String {
+        percent_encode(self.object_name(key).as_bytes(), OBJECT_NAME_ENCODE_SET).to_string()
+    }
+
+    fn endpoint(&self, path: &str) -> String {
+        format!("{}{}", self.config.endpoint.trim_end_matches('/'), path)
+    }
+
+    async fn async_set(&self, key: &str, value: S::Value) -> Result<(), Error> {
+        let token = self.token.bearer_token().await?;
+        let payload = self.codec.encode(&value)?;
+        let url = self.endpoint(&format!(
+            "/upload/storage/v1/b/{}/o?uploadType=media&name={}",
+            self.config.bucket_name,
+            self.encoded_object_name(key)
+        ));
+        let response = self
+            .client
+            .post(url)
+            .bearer_auth(token)
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .body(payload)
+            .send()
+            .await
+            .map_err(|_| Error::Unavailable)?;
+        if !response.status().is_success() {
+            return Err(Error::Unavailable);
+        }
+        Ok(())
+    }
+
+    async fn async_get(&self, key: &str) -> Result<Option<S::Value>, Error> {
+        let token = self.token.bearer_token().await?;
+        let url = self.endpoint(&format!(
+            "/storage/v1/b/{}/o/{}?alt=media",
+            self.config.bucket_name,
+            self.encoded_object_name(key)
+        ));
+        let response = self
+            .client
+            .get(url)
+            .bearer_auth(token)
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .send()
+            .await
+            .map_err(|_| Error::Unavailable)?;
+        if response.status() == reqwest::StatusCode::NOT_FOUND {
+            return Ok(None);
+        }
+        if !response.status().is_success() {
+            return Err(Error::Unavailable);
+        }
+        let body = response.bytes().await.map_err(|_| Error::Unavailable)?;
+        self.codec
+            .decode(&body)
+            .map(Some)
+            .map_err(|_| Error::InvalidEntry)
+    }
+
+    fn run_sync<T, F>(future: F) -> Result<T, Error>
+    where
+        F: Future<Output = Result<T, Error>> + Send,
+        T: Send,
+    {
+        let run = || {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .map_err(|_| Error::Unavailable)
+                .and_then(|runtime| runtime.block_on(future))
+        };
+        if let Ok(handle) = tokio::runtime::Handle::try_current() {
+            if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
+                return tokio::task::block_in_place(run);
+            }
+            return std::thread::scope(|scope| {
+                scope
+                    .spawn(run)
+                    .join()
+                    .map_err(|_| Error::Unavailable)
+                    .and_then(|result| result)
+            });
+        }
+        run()
+    }
+}
+
+impl<S: CacheCodec> BaseCache for GcsCache<S> {
+    type Value = S::Value;
+    type Context = ExactCacheContext;
+
+    fn get_ttl(&self, _: &Self::Context) -> Option<Duration> {
+        None
+    }
+
+    fn set_cache(&self, key: &str, value: Self::Value, _: &Self::Context) -> Result<(), Error> {
+        Self::run_sync(self.async_set(key, value))
+    }
+
+    fn get_cache(&self, key: &str, _: &Self::Context) -> Result<Option<Self::Value>, Error> {
+        Self::run_sync(self.async_get(key))
+    }
+
+    async fn async_set_cache(
+        &self,
+        key: &str,
+        value: Self::Value,
+        _: Self::Context,
+    ) -> Result<(), Error> {
+        self.async_set(key, value).await
+    }
+
+    async fn async_get_cache(
+        &self,
+        key: &str,
+        _: &Self::Context,
+    ) -> Result<Option<Self::Value>, Error> {
+        self.async_get(key).await
+    }
+
+    async fn async_set_cache_pipeline(
+        &self,
+        entries: Vec<(String, Self::Value)>,
+        context: Self::Context,
+    ) -> Result<(), Error> {
+        try_join_all(entries.into_iter().map(|(key, value)| {
+            let context = context.clone();
+            async move { self.async_set_cache(&key, value, context).await }
+        }))
+        .await
+        .map(|_| ())
+    }
+
+    async fn disconnect(&self) -> Result<(), Error> {
+        Ok(())
+    }
+
+    async fn test_connection(&self) -> Result<CacheConnectionResult, Error> {
+        Err(Error::UnsupportedOperation)
+    }
+}
+
+impl<S: CacheCodec> BatchCache for GcsCache<S> {
+    async fn async_batch_get_cache(
+        &self,
+        keys: Vec<String>,
+        context: Self::Context,
+    ) -> Result<Vec<BatchEntry<Self::Value>>, Error> {
+        try_join_all(keys.into_iter().map(|key| {
+            let context = context.clone();
+            async move {
+                match self.async_get_cache(&key, &context).await {
+                    Ok(Some(value)) => Ok(BatchEntry::Hit(value)),
+                    Ok(None) => Ok(BatchEntry::Miss),
+                    Err(Error::InvalidEntry) => Ok(BatchEntry::Invalid),
+                    Err(error) => Err(error),
+                }
+            }
+        }))
+        .await
+    }
+}
+
+impl<S: CacheCodec> FlushCache for GcsCache<S> {
+    fn flush_cache(&self) -> Result<(), Error> {
+        Ok(())
+    }
+}
diff --git a/litellm-rust/crates/cache-gcs/src/lib.rs b/litellm-rust/crates/cache-gcs/src/lib.rs
new file mode 100644
index 00000000000..cbb61cf0685
--- /dev/null
+++ b/litellm-rust/crates/cache-gcs/src/lib.rs
@@ -0,0 +1,5 @@
+mod cache;
+mod token;
+
+pub use cache::{DEFAULT_ENDPOINT, GcsCache, GcsConfig, key_prefix};
+pub use token::{GcpTokenSource, StaticTokenSource, TokenSource};
diff --git a/litellm-rust/crates/cache-gcs/src/token.rs b/litellm-rust/crates/cache-gcs/src/token.rs
new file mode 100644
index 00000000000..adb601c276c
--- /dev/null
+++ b/litellm-rust/crates/cache-gcs/src/token.rs
@@ -0,0 +1,44 @@
+use std::{future::Future, pin::Pin};
+
+use litellm_auth_gcp::{VertexAuth, VertexConfig};
+use litellm_auth_types::{InputSource, SecretValue, Sourced};
+use litellm_cache::Error;
+
+pub trait TokenSource: Send + Sync + 'static {
+    fn bearer_token(&self) -> Pin<Box<dyn Future<Output = Result<String, Error>> + Send + '_>>;
+}
+
+pub struct GcpTokenSource {
+    auth: VertexAuth,
+    config: VertexConfig,
+}
+
+impl GcpTokenSource {
+    pub fn new(path_service_account: Option<String>) -> Self {
+        let credentials = path_service_account
+            .map(|path| Sourced::new(SecretValue::new(path), InputSource::Deployment));
+        Self {
+            auth: VertexAuth::default(),
+            config: VertexConfig::new(credentials, None, None),
+        }
+    }
+}
+
+impl TokenSource for GcpTokenSource {
+    fn bearer_token(&self) -> Pin<Box<dyn Future<Output = Result<String, Error>> + Send + '_>> {
+        Box::pin(async move {
+            self.auth
+                .access_token(&self.config, &|name| std::env::var(name).ok())
+                .await
+                .map_err(|_| Error::Unavailable)
+        })
+    }
+}
+
+pub struct StaticTokenSource(pub String);
+
+impl TokenSource for StaticTokenSource {
+    fn bearer_token(&self) -> Pin<Box<dyn Future<Output = Result<String, Error>> + Send + '_>> {
+        Box::pin(async move { Ok(self.0.clone()) })
+    }
+}
diff --git a/litellm-rust/crates/cache-gcs/tests/cache.rs b/litellm-rust/crates/cache-gcs/tests/cache.rs
new file mode 100644
index 00000000000..79e4ed100f8
--- /dev/null
+++ b/litellm-rust/crates/cache-gcs/tests/cache.rs
@@ -0,0 +1,290 @@
+use std::{sync::Arc, time::Duration};
+
+use litellm_cache::{
+    BaseCache, BatchCache, BatchEntry, CacheContext, Error, ExactCacheContext, FlushCache,
+    JsonCodec,
+};
+use litellm_cache_gcs::{GcsCache, GcsConfig, StaticTokenSource, TokenSource, key_prefix};
+use serde_json::json;
+use wiremock::{
+    Mock, MockServer, ResponseTemplate,
+    matchers::{body_bytes, header, method, path, query_param},
+};
+
+fn config(server: &MockServer, gcs_path: Option<&str>) -> GcsConfig {
+    GcsConfig {
+        bucket_name: "bucket".into(),
+        gcs_path: gcs_path.map(str::to_string),
+        path_service_account: None,
+        endpoint: server.uri(),
+    }
+}
+
+fn cache(server: &MockServer, gcs_path: Option<&str>) -> GcsCache<JsonCodec<serde_json::Value>> {
+    GcsCache::with_token_source(
+        config(server, gcs_path),
+        JsonCodec::new(),
+        Arc::new(StaticTokenSource("tok".into())),
+    )
+    .unwrap()
+}
+
+#[tokio::test]
+async fn set_writes_encoded_object_and_headers() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/upload/storage/v1/b/bucket/o"))
+        .and(query_param("uploadType", "media"))
+        .and(header("authorization", "Bearer tok"))
+        .and(header("content-type", "application/json"))
+        .and(body_bytes(br#"{"value":"entry"}"#))
+        .respond_with(ResponseTemplate::new(200))
+        .expect(1)
+        .mount(&server)
+        .await;
+    cache(&server, Some("cache/"))
+        .set_cache(
+            "team:a b/c",
+            json!({"value": "entry"}),
+            &ExactCacheContext::default(),
+        )
+        .unwrap();
+    let requests = server.received_requests().await.unwrap();
+    assert_eq!(requests.len(), 1);
+    assert_eq!(
+        requests[0].url.query(),
+        Some("uploadType=media&name=cache%2Fteam%3Aa%20b%2Fc")
+    );
+}
+
+#[tokio::test]
+async fn get_maps_statuses_and_decode_failures() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/hit"))
+        .and(query_param("alt", "media"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"value": "entry"})))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/missing"))
+        .respond_with(ResponseTemplate::new(404))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/server-error"))
+        .respond_with(ResponseTemplate::new(500))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/invalid"))
+        .respond_with(ResponseTemplate::new(200).set_body_string("not json"))
+        .mount(&server)
+        .await;
+
+    let cache = cache(&server, None);
+    assert_eq!(
+        cache
+            .get_cache("hit", &ExactCacheContext::default())
+            .unwrap(),
+        Some(json!({"value": "entry"}))
+    );
+    assert_eq!(
+        cache
+            .get_cache("missing", &ExactCacheContext::default())
+            .unwrap(),
+        None
+    );
+    assert_eq!(
+        cache
+            .get_cache("server-error", &ExactCacheContext::default())
+            .unwrap_err(),
+        Error::Unavailable
+    );
+    assert_eq!(
+        cache
+            .get_cache("invalid", &ExactCacheContext::default())
+            .unwrap_err(),
+        Error::InvalidEntry
+    );
+}
+
+#[test]
+fn key_prefix_normalizes_paths() {
+    assert_eq!(key_prefix(None), "");
+    assert_eq!(key_prefix(Some("a/b/")), "a/b/");
+    assert_eq!(key_prefix(Some("a/b")), "a/b/");
+    assert_eq!(key_prefix(Some("")), "");
+}
+
+#[tokio::test]
+async fn ignores_ttl_and_writes_pipeline_concurrently() {
+    let server = MockServer::start().await;
+    for key in ["one", "two", "three"] {
+        Mock::given(method("POST"))
+            .and(path("/upload/storage/v1/b/bucket/o"))
+            .and(query_param("uploadType", "media"))
+            .and(query_param("name", key))
+            .respond_with(ResponseTemplate::new(200))
+            .expect(1)
+            .mount(&server)
+            .await;
+    }
+    let cache = cache(&server, None);
+    assert_eq!(cache.get_ttl(&ExactCacheContext::default()), None);
+    assert_eq!(
+        cache.get_ttl(&ExactCacheContext::default().with_ttl(Some(Duration::from_secs(5)))),
+        None
+    );
+    cache
+        .async_set_cache_pipeline(
+            vec![
+                ("one".into(), json!({"key": "one"})),
+                ("two".into(), json!({"key": "two"})),
+                ("three".into(), json!({"key": "three"})),
+            ],
+            ExactCacheContext::default().with_ttl(Some(Duration::from_secs(5))),
+        )
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn async_batch_get_preserves_hits_misses_and_invalid_entries() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/hit"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"value": "entry"})))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/missing"))
+        .respond_with(ResponseTemplate::new(404))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/invalid"))
+        .respond_with(ResponseTemplate::new(200).set_body_string("not json"))
+        .mount(&server)
+        .await;
+
+    assert_eq!(
+        cache(&server, None)
+            .async_batch_get_cache(
+                vec!["hit".into(), "missing".into(), "invalid".into()],
+                ExactCacheContext::default(),
+            )
+            .await
+            .unwrap(),
+        vec![
+            BatchEntry::Hit(json!({"value": "entry"})),
+            BatchEntry::Miss,
+            BatchEntry::Invalid,
+        ]
+    );
+}
+
+#[tokio::test]
+async fn lifecycle_operations_are_noops_and_connection_test_is_unsupported() {
+    let server = MockServer::start().await;
+    let cache = cache(&server, None);
+    assert_eq!(cache.flush_cache(), Ok(()));
+    assert_eq!(cache.disconnect().await, Ok(()));
+    assert_eq!(
+        cache.test_connection().await,
+        Err(Error::UnsupportedOperation)
+    );
+}
+
+#[test]
+fn sync_operations_work_without_an_active_runtime() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let server = runtime.block_on(MockServer::start());
+    runtime.block_on(
+        Mock::given(method("POST"))
+            .and(path("/upload/storage/v1/b/bucket/o"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server),
+    );
+    runtime.block_on(
+        Mock::given(method("GET"))
+            .and(path("/storage/v1/b/bucket/o/key"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({"value": "entry"})))
+            .mount(&server),
+    );
+    let cache = cache(&server, None);
+    cache
+        .set_cache(
+            "key",
+            json!({"value": "entry"}),
+            &ExactCacheContext::default(),
+        )
+        .unwrap();
+    assert_eq!(
+        cache
+            .get_cache("key", &ExactCacheContext::default())
+            .unwrap(),
+        Some(json!({"value": "entry"}))
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn sync_operations_work_inside_a_multi_thread_runtime() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/upload/storage/v1/b/bucket/o"))
+        .respond_with(ResponseTemplate::new(200))
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/storage/v1/b/bucket/o/key"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"value": "entry"})))
+        .mount(&server)
+        .await;
+    let cache = cache(&server, None);
+    cache
+        .set_cache(
+            "key",
+            json!({"value": "entry"}),
+            &ExactCacheContext::default(),
+        )
+        .unwrap();
+    assert_eq!(
+        cache
+            .get_cache("key", &ExactCacheContext::default())
+            .unwrap(),
+        Some(json!({"value": "entry"}))
+    );
+}
+
+struct FailingTokenSource;
+
+impl TokenSource for FailingTokenSource {
+    fn bearer_token(
+        &self,
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, Error>> + Send + '_>>
+    {
+        Box::pin(async { Err(Error::Unavailable) })
+    }
+}
+
+#[tokio::test]
+async fn token_source_failure_skips_http() {
+    let server = MockServer::start().await;
+    let cache = GcsCache::with_token_source(
+        config(&server, None),
+        JsonCodec::<serde_json::Value>::new(),
+        Arc::new(FailingTokenSource),
+    )
+    .unwrap();
+    assert_eq!(
+        cache
+            .get_cache("key", &ExactCacheContext::default())
+            .unwrap_err(),
+        Error::Unavailable
+    );
+    assert_eq!(server.received_requests().await.unwrap().len(), 0);
+}
diff --git a/litellm-rust/crates/cache/src/error.rs b/litellm-rust/crates/cache/src/error.rs
index ff3ff6572d4..1a381d0afd8 100644
--- a/litellm-rust/crates/cache/src/error.rs
+++ b/litellm-rust/crates/cache/src/error.rs
@@ -6,4 +6,6 @@ pub enum Error {
     InvalidEntry,
     #[error("flushing Redis requires an explicit namespace")]
     UnscopedFlush,
+    #[error("operation is not supported by this cache")]
+    UnsupportedOperation,
 }
diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml
index 1eb2ec28036..0f060c801a5 100644
--- a/litellm-rust/crates/python-bridge/Cargo.toml
+++ b/litellm-rust/crates/python-bridge/Cargo.toml
@@ -23,6 +23,7 @@ bytes.workspace = true
 litellm-cache.workspace = true
 litellm-cache-memory.workspace = true
 litellm-cache-redis.workspace = true
+litellm-cache-gcs.workspace = true
 litellm-cache-response.workspace = true
 serde.workspace = true
 litellm-auth.workspace = true
diff --git a/litellm-rust/crates/python-bridge/src/cache/config.rs b/litellm-rust/crates/python-bridge/src/cache/config.rs
index 0e7d6aee11d..66f493dd760 100644
--- a/litellm-rust/crates/python-bridge/src/cache/config.rs
+++ b/litellm-rust/crates/python-bridge/src/cache/config.rs
@@ -73,9 +73,17 @@ pub(super) struct RedisCacheConfig {
     pub(super) connection: RedisConnectionConfig,
 }
 
+#[derive(Debug, PartialEq)]
+pub(super) struct GcsCacheConfig {
+    pub(super) bucket_name: String,
+    pub(super) key_prefix: String,
+    pub(super) path_service_account: Option<String>,
+}
+
 pub(super) enum CacheBackendConfig {
     Memory(MemoryCacheConfig),
     Redis(Box<RedisCacheConfig>),
+    Gcs(GcsCacheConfig),
 }
 
 #[allow(dead_code, reason = "consumed by the cache activation follow-up")]
@@ -90,6 +98,7 @@ pub(super) enum UnsupportedCacheConfig {
     RedisCredentials,
     RedisConnection,
     RedisOption,
+    GcsBucket,
 }
 
 impl UnsupportedCacheConfig {
@@ -100,6 +109,7 @@ impl UnsupportedCacheConfig {
             Self::RedisCredentials => "native Redis credentials require Python",
             Self::RedisConnection => "native Redis connection type is not implemented",
             Self::RedisOption => "native Redis configuration requires Python",
+            Self::GcsBucket => "native GCS cache requires a configured bucket name",
         }
     }
 }
@@ -142,14 +152,20 @@ impl NativeCacheConfig {
                 }))),
                 Err(reason) => Ok(CacheConfigProjection::Unsupported(reason)),
             },
+            Some(CacheType::Gcs) => match project_gcs(&backend)? {
+                Ok(backend) => Ok(CacheConfigProjection::Native(Box::new(Self {
+                    policy,
+                    backend: CacheBackendConfig::Gcs(backend),
+                }))),
+                Err(reason) => Ok(CacheConfigProjection::Unsupported(reason)),
+            },
             Some(
                 CacheType::RedisSemantic
                 | CacheType::ValkeySemantic
                 | CacheType::S3
                 | CacheType::Disk
                 | CacheType::QdrantSemantic
-                | CacheType::AzureBlob
-                | CacheType::Gcs,
+                | CacheType::AzureBlob,
             )
             | None => Ok(CacheConfigProjection::Unsupported(
                 UnsupportedCacheConfig::Backend,
@@ -158,12 +174,12 @@ impl NativeCacheConfig {
     }
 
     pub(super) fn service_mismatch(&self, service: &NativeResponseCache) -> Option<&'static str> {
-        if service.default_ttl()
-            != Some(match &self.backend {
-                CacheBackendConfig::Memory(config) => config.default_ttl,
-                CacheBackendConfig::Redis(config) => config.default_ttl,
-            })
-        {
+        let expected = match &self.backend {
+            CacheBackendConfig::Memory(config) => Some(config.default_ttl),
+            CacheBackendConfig::Redis(config) => Some(config.default_ttl),
+            CacheBackendConfig::Gcs(_) => None,
+        };
+        if service.default_ttl() != expected {
             return Some("facade and native backend default TTLs must match");
         }
         match &self.backend {
@@ -185,6 +201,31 @@ impl NativeCacheConfig {
             CacheBackendConfig::Redis(config) => (service.namespace()
                 != config.namespace.as_deref())
             .then_some("facade and native backend namespaces must match"),
+            CacheBackendConfig::Gcs(_) if service.kind() != "gcs" => {
+                Some("facade and native backend types must match")
+            }
+            CacheBackendConfig::Gcs(config)
+                if service
+                    .gcs_backend()
+                    .is_none_or(|backend| backend.bucket_name() != config.bucket_name) =>
+            {
+                Some("facade and native backend buckets must match")
+            }
+            CacheBackendConfig::Gcs(config)
+                if service
+                    .gcs_backend()
+                    .is_none_or(|backend| backend.key_prefix() != config.key_prefix) =>
+            {
+                Some("facade and native backend key prefixes must match")
+            }
+            CacheBackendConfig::Gcs(config)
+                if service.gcs_backend().is_none_or(|backend| {
+                    backend.path_service_account() != config.path_service_account.as_deref()
+                }) =>
+            {
+                Some("facade and native backend credentials must match")
+            }
+            CacheBackendConfig::Gcs(_) => None,
         }
     }
 }
@@ -201,6 +242,23 @@ fn project_memory(backend: &Bound<'_, PyAny>) -> PyResult<MemoryCacheConfig> {
     })
 }
 
+#[inline(never)]
+fn project_gcs(
+    backend: &Bound<'_, PyAny>,
+) -> PyResult<Result<GcsCacheConfig, UnsupportedCacheConfig>> {
+    let bucket_name = match backend.getattr("bucket_name")?.extract::<Option<String>>() {
+        Ok(Some(bucket_name)) if !bucket_name.is_empty() => bucket_name,
+        _ => return Ok(Err(UnsupportedCacheConfig::GcsBucket)),
+    };
+    Ok(Ok(GcsCacheConfig {
+        bucket_name,
+        key_prefix: backend.getattr("key_prefix")?.extract::<String>()?,
+        path_service_account: backend
+            .getattr("path_service_account")?
+            .extract::<Option<String>>()?,
+    }))
+}
+
 #[inline(never)]
 fn project_redis(
     backend: &Bound<'_, PyAny>,
@@ -468,8 +526,8 @@ mod tests {
     use pyo3::{prelude::*, types::PyDict};
 
     use super::{
-        CacheBackendConfig, CacheConfigProjection, CertificateRequirement, NativeCacheConfig,
-        RedisProtocol,
+        CacheBackendConfig, CacheConfigProjection, CertificateRequirement, GcsCacheConfig,
+        NativeCacheConfig, RedisProtocol, UnsupportedCacheConfig,
     };
     use crate::cache::native::NativeResponseCache;
 
@@ -531,6 +589,71 @@ mod tests {
         });
     }
 
+    #[test]
+    fn projects_gcs_configuration() {
+        Python::initialize();
+        Python::attach(|py| {
+            let facade = facade(
+                py,
+                "backend = SimpleNamespace(bucket_name='bucket', key_prefix='cache/', path_service_account='credentials.json')\n\
+                 facade = SimpleNamespace(type='gcs', mode='default-on', ttl=None, namespace=None, supported_call_types=None, redis_flush_size=None, semantic_cache_scope='key', cache=backend)",
+            );
+            let CacheConfigProjection::Native(config) =
+                NativeCacheConfig::project(&facade).unwrap()
+            else {
+                panic!("GCS cache should be supported");
+            };
+            let CacheBackendConfig::Gcs(gcs) = config.backend else {
+                panic!("expected GCS configuration");
+            };
+            assert_eq!(
+                gcs,
+                GcsCacheConfig {
+                    bucket_name: "bucket".into(),
+                    key_prefix: "cache/".into(),
+                    path_service_account: Some("credentials.json".into()),
+                }
+            );
+            let matching = NativeResponseCache::gcs(
+                litellm_cache_gcs::GcsConfig {
+                    bucket_name: "bucket".into(),
+                    gcs_path: Some("cache/".into()),
+                    path_service_account: Some("credentials.json".into()),
+                    endpoint: litellm_cache_gcs::DEFAULT_ENDPOINT.into(),
+                },
+                Some("token".into()),
+            )
+            .unwrap();
+            let matching_config = NativeCacheConfig {
+                policy: config.policy,
+                backend: CacheBackendConfig::Gcs(gcs),
+            };
+            assert_eq!(matching_config.service_mismatch(&matching), None);
+        });
+    }
+
+    #[test]
+    fn rejects_gcs_without_a_bucket_name() {
+        Python::initialize();
+        Python::attach(|py| {
+            let facade = facade(
+                py,
+                "backend = SimpleNamespace(bucket_name=None, key_prefix='', path_service_account=None)\n\
+                 facade = SimpleNamespace(type='gcs', mode='default-on', ttl=None, namespace=None, supported_call_types=None, redis_flush_size=None, semantic_cache_scope='key', cache=backend)",
+            );
+            let CacheConfigProjection::Unsupported(reason) =
+                NativeCacheConfig::project(&facade).unwrap()
+            else {
+                panic!("GCS cache without a bucket should be unsupported");
+            };
+            assert!(matches!(&reason, UnsupportedCacheConfig::GcsBucket));
+            assert_eq!(
+                reason.message(),
+                "native GCS cache requires a configured bucket name"
+            );
+        });
+    }
+
     #[test]
     fn projects_resolved_redis_tls_configuration() {
         Python::initialize();
diff --git a/litellm-rust/crates/python-bridge/src/cache/facade.rs b/litellm-rust/crates/python-bridge/src/cache/facade.rs
index f2f86c14b37..6aa94754e2d 100644
--- a/litellm-rust/crates/python-bridge/src/cache/facade.rs
+++ b/litellm-rust/crates/python-bridge/src/cache/facade.rs
@@ -192,6 +192,7 @@ impl FacadeGuard {
         let (module, name, cache_kind) = match kind {
             "memory" => ("litellm.caching.in_memory_cache", "InMemoryCache", "local"),
             "redis" => ("litellm.caching.redis_cache", "RedisCache", "redis"),
+            "gcs" => ("litellm.caching.gcs_cache", "GCSCache", "gcs"),
             _ => unreachable!(),
         };
         let backend = facade.getattr("cache")?;
@@ -235,6 +236,9 @@ impl FacadeGuard {
                     "max_size_per_item",
                     "redis_kwargs",
                     "redis_flush_size",
+                    "bucket_name",
+                    "key_prefix",
+                    "path_service_account",
                 ],
             )?,
             redis_pool: (kind == "redis")
diff --git a/litellm-rust/crates/python-bridge/src/cache/handle.rs b/litellm-rust/crates/python-bridge/src/cache/handle.rs
index 8251b3df06c..87d9300b08b 100644
--- a/litellm-rust/crates/python-bridge/src/cache/handle.rs
+++ b/litellm-rust/crates/python-bridge/src/cache/handle.rs
@@ -1,6 +1,8 @@
 use litellm_host_python::release_gil;
 use pyo3::{PyTraverseError, PyVisit, exceptions::PyRuntimeError, prelude::*};
 
+use litellm_cache_gcs::{DEFAULT_ENDPOINT, GcsConfig};
+
 use super::{cache_error, facade::FacadeGuard, native::NativeResponseCache, request::duration};
 
 #[pyclass(frozen, name = "_CacheTestHandle")]
@@ -51,6 +53,31 @@ impl CacheTestHandle {
         })
     }
 
+    #[staticmethod]
+    #[pyo3(signature = (bucket_name, *, gcs_path=None, path_service_account=None, endpoint=None, token=None))]
+    fn gcs(
+        py: Python<'_>,
+        bucket_name: String,
+        gcs_path: Option<String>,
+        path_service_account: Option<String>,
+        endpoint: Option<String>,
+        token: Option<String>,
+    ) -> PyResult<Self> {
+        let config = GcsConfig {
+            bucket_name,
+            gcs_path,
+            path_service_account,
+            endpoint: endpoint.unwrap_or_else(|| DEFAULT_ENDPOINT.to_string()),
+        };
+        let service = release_gil(py, move || NativeResponseCache::gcs(config, token))
+            .map_err(cache_error)?;
+        Ok(Self {
+            service,
+            guard: None,
+            pid: std::process::id(),
+        })
+    }
+
     #[getter]
     fn backend(&self) -> &'static str {
         self.service.kind()
diff --git a/litellm-rust/crates/python-bridge/src/cache/mod.rs b/litellm-rust/crates/python-bridge/src/cache/mod.rs
index aec08610f6e..ac4494150d9 100644
--- a/litellm-rust/crates/python-bridge/src/cache/mod.rs
+++ b/litellm-rust/crates/python-bridge/src/cache/mod.rs
@@ -10,7 +10,7 @@ mod resolver;
 
 use litellm_cache::Error;
 use pyo3::{
-    exceptions::{PyRuntimeError, PyValueError},
+    exceptions::{PyNotImplementedError, PyRuntimeError, PyValueError},
     prelude::*,
 };
 
@@ -21,6 +21,7 @@ pub(crate) use self::{
 fn cache_error(error: Error) -> PyErr {
     match error {
         Error::InvalidEntry => PyValueError::new_err(error.to_string()),
+        Error::UnsupportedOperation => PyNotImplementedError::new_err(error.to_string()),
         _ => PyRuntimeError::new_err(error.to_string()),
     }
 }
diff --git a/litellm-rust/crates/python-bridge/src/cache/native.rs b/litellm-rust/crates/python-bridge/src/cache/native.rs
index a9475429e45..9a030ac0434 100644
--- a/litellm-rust/crates/python-bridge/src/cache/native.rs
+++ b/litellm-rust/crates/python-bridge/src/cache/native.rs
@@ -1,6 +1,7 @@
 use std::{sync::Arc, time::Duration};
 
 use litellm_cache::{CacheCodec, CacheConnectionResult, Error};
+use litellm_cache_gcs::{GcsCache, GcsConfig, StaticTokenSource};
 use litellm_cache_memory::InMemoryCache;
 use litellm_cache_redis::RedisCache;
 use litellm_cache_response::{
@@ -15,6 +16,7 @@ pub(super) enum NativeResponseCache {
         cache: Arc<ResponseCache<RedisCache<ResponseCacheCodec>>>,
         buffer: Option<Arc<WriteBuffer>>,
     },
+    Gcs(Arc<ResponseCache<GcsCache<ResponseCacheCodec>>>),
 }
 
 impl NativeResponseCache {
@@ -43,6 +45,18 @@ impl NativeResponseCache {
             buffer: None,
         })
     }
+
+    pub fn gcs(config: GcsConfig, token: Option<String>) -> Result<Self, Error> {
+        let backend = match token {
+            Some(token) => GcsCache::with_token_source(
+                config,
+                ResponseCacheCodec,
+                Arc::new(StaticTokenSource(token)),
+            )?,
+            None => GcsCache::new(config, ResponseCacheCodec)?,
+        };
+        Ok(Self::Gcs(Arc::new(ResponseCache::new(Arc::new(backend)))))
+    }
 }
 
 impl NativeResponseCache {
@@ -50,6 +64,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(_) => "memory",
             Self::Redis { .. } => "redis",
+            Self::Gcs(_) => "gcs",
         }
     }
 
@@ -57,6 +72,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.default_ttl(),
             Self::Redis { cache, .. } => cache.default_ttl(),
+            Self::Gcs(cache) => cache.default_ttl(),
         }
     }
 
@@ -64,6 +80,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(_) => None,
             Self::Redis { cache, .. } => cache.backend().namespace(),
+            Self::Gcs(_) => None,
         }
     }
 
@@ -71,6 +88,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => Some(cache.backend().max_size_in_memory()),
             Self::Redis { .. } => None,
+            Self::Gcs(_) => None,
         }
     }
 
@@ -78,6 +96,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.backend().max_entry_bytes(),
             Self::Redis { .. } => None,
+            Self::Gcs(_) => None,
         }
     }
 
@@ -99,6 +118,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.lookup(request, now),
             Self::Redis { cache, .. } => cache.lookup(request, now),
+            Self::Gcs(cache) => cache.lookup(request, now),
         }
     }
 
@@ -111,6 +131,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.store(request, response, now),
             Self::Redis { cache, .. } => cache.store(request, response, now),
+            Self::Gcs(cache) => cache.store(request, response, now),
         }
     }
 
@@ -122,6 +143,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.lookup_batch(requests, now),
             Self::Redis { cache, .. } => cache.lookup_batch(requests, now),
+            Self::Gcs(cache) => cache.lookup_batch(requests, now),
         }
     }
 
@@ -133,6 +155,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.async_lookup(request, now).await,
             Self::Redis { cache, .. } => cache.async_lookup(request, now).await,
+            Self::Gcs(cache) => cache.async_lookup(request, now).await,
         }
     }
 
@@ -152,6 +175,7 @@ impl NativeResponseCache {
                 cache,
                 buffer: Some(buffer),
             } => buffer.async_store(cache, request, response, now).await,
+            Self::Gcs(cache) => cache.async_store(request, response, now).await,
         }
     }
 
@@ -163,6 +187,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.async_lookup_batch(requests, now).await,
             Self::Redis { cache, .. } => cache.async_lookup_batch(requests, now).await,
+            Self::Gcs(cache) => cache.async_lookup_batch(requests, now).await,
         }
     }
 
@@ -174,6 +199,7 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.async_store_batch(entries, now).await,
             Self::Redis { cache, .. } => cache.async_store_batch(entries, now).await,
+            Self::Gcs(cache) => cache.async_store_batch(entries, now).await,
         }
     }
 
@@ -186,6 +212,7 @@ impl NativeResponseCache {
                 }
                 cache.async_flush().await
             }
+            Self::Gcs(cache) => cache.async_flush().await,
         }
     }
 
@@ -193,6 +220,14 @@ impl NativeResponseCache {
         match self {
             Self::Memory(cache) => cache.test_connection().await,
             Self::Redis { cache, .. } => cache.test_connection().await,
+            Self::Gcs(cache) => cache.test_connection().await,
+        }
+    }
+
+    pub fn gcs_backend(&self) -> Option<&GcsCache<ResponseCacheCodec>> {
+        match self {
+            Self::Gcs(cache) => Some(cache.backend()),
+            _ => None,
         }
     }
 }