fix(python-bridge): harden sync and async route boundaries

litellm-rust/Cargo.lock

@@ -1442,13 +1442,16 @@ name = "litellm-python-bridge"
 version = "0.1.0"
 dependencies = [
  "criterion",
+ "futures-util",
  "litellm-ai-gateway",
  "litellm-core",
  "litellm-python-interop",
  "pyo3",
  "pyo3-async-runtimes",
+ "serde",
  "serde_json",
  "tokio",
+ "tokio-tungstenite",
 ]
 
 [[package]]
@@ -1669,9 +1672,9 @@ dependencies = [
 
 [[package]]
 name = "pyo3"
-version = "0.29.0"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd274650b21d4bfc26a0a47587962c1edb425f69287324355cd040c3ea66071c"
+checksum = "4688ddedf473e32662b9b067670129a8afb8c18e351482c70d62ba4a88171e8b"
 dependencies = [
  "libc",
  "once_cell",
@@ -1697,18 +1700,18 @@ dependencies = [
 
 [[package]]
 name = "pyo3-build-config"
-version = "0.29.0"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c5e2a7d2f0d013342f295c048ad19237add5154a55b1c5a254c0ec93d4109078"
+checksum = "f41027e41b4bd03f6e60f9f417fe24a6341a6bb744edd62b6f709f2a52ea30e9"
 dependencies = [
  "target-lexicon",
 ]
 
 [[package]]
 name = "pyo3-ffi"
-version = "0.29.0"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ca85c467da1bbc8d866eea5deff9cf29ea5f7785054a17da36e65bda9c05845b"
+checksum = "e591a95526fead067432c3b3a33fc74770b87b1e04e73671090d9c2055a2b327"
 dependencies = [
  "libc",
  "pyo3-build-config",
@@ -1716,9 +1719,9 @@ dependencies = [
 
 [[package]]
 name = "pyo3-macros"
-version = "0.29.0"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ac53762fd065daa3194dd09337a38bd793a188100fd1a9304c4ab312d901771"
+checksum = "73225868fc1cd84eef2c3c230ddb91273bf1de46aeb8a4248da76d32a0924a1c"
 dependencies = [
  "proc-macro2",
  "pyo3-macros-backend",
@@ -1728,9 +1731,9 @@ dependencies = [
 
 [[package]]
 name = "pyo3-macros-backend"
-version = "0.29.0"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ca3a1557399783172dc5bf39cfca835157732532cba56b71d2292161e53b362"
+checksum = "571575aa3749fa6216757dd47d2a3e7ef360f329a40f0666a9fbd14889024952"
 dependencies = [
  "heck",
  "proc-macro2",

litellm-rust/Cargo.toml

@@ -18,7 +18,7 @@ litellm-core = { path = "crates/core" }
 litellm-ai-gateway = { path = "crates/ai-gateway", default-features = false }
 litellm-python-interop = { path = "crates/python-interop" }
 axum = "0.7"
-pyo3 = "0.29.0"
+pyo3 = "0.29.2"
 pyo3-async-runtimes = { version = "0.29.0", features = ["tokio-runtime"] }
 pythonize = "0.29.0"
 rand = "0.8"

litellm-rust/crates/python-bridge/Cargo.toml

@@ -16,16 +16,19 @@ extension-module = ["pyo3/extension-module"]
 panic-test = []
 
 [dependencies]
+futures-util.workspace = true
 litellm-core = { workspace = true, features = ["bedrock-auth"] }
 litellm-ai-gateway = { workspace = true, default-features = false }
 litellm-python-interop.workspace = true
 pyo3.workspace = true
 pyo3-async-runtimes.workspace = true
+serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
 
 [dev-dependencies]
 criterion = "0.8.2"
+tokio-tungstenite.workspace = true
 
 [[bench]]
 name = "serialization"

litellm-rust/crates/python-bridge/src/lib.rs

@@ -6,6 +6,7 @@ mod routes;
 use litellm_ai_gateway::io::responses_ws::ResponsesWebSocketConnection as RustResponsesWebSocketConnection;
 use pyo3::prelude::*;
 use pyo3::types::PyAny;
+use serde_json::Value;
 
 use crate::errors::core_error_to_pyerr;
 use crate::marshal::{marshal_headers, optional_timeout};
@@ -23,16 +24,16 @@ impl ResponsesWebSocketConnection {
         _cls: &Bound<'py, pyo3::types::PyType>,
         py: Python<'py>,
         url: String,
-        headers: Option<Py<PyAny>>,
+        #[pyo3(from_py_with = litellm_python_interop::from_py)] headers: Option<Value>,
         timeout_seconds: Option<f64>,
     ) -> PyResult<Bound<'py, PyAny>> {
-        let headers = marshal_headers(py, headers)?;
+        let headers = marshal_headers(headers)?;
         let timeout = optional_timeout(timeout_seconds);
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
             let inner = RustResponsesWebSocketConnection::connect_url(&url, &headers, timeout)
                 .await
                 .map_err(core_error_to_pyerr)?;
-            Python::attach(|py| Py::new(py, ResponsesWebSocketConnection { inner }))
+            Ok(ResponsesWebSocketConnection { inner })
         })
     }
 
@@ -58,24 +59,36 @@ impl ResponsesWebSocketConnection {
     }
 }
 
-#[pymodule]
-fn _native(module: &Bound<'_, PyModule>) -> PyResult<()> {
-    errors::register(module)?;
-    routes::register(module)?;
-    module.add_class::<ResponsesWebSocketConnection>()?;
-    diagnostics::register(module)
+#[pymodule(gil_used = false)]
+mod _native {
+    use pyo3::prelude::*;
+
+    #[pymodule_init]
+    fn init(module: &Bound<'_, PyModule>) -> PyResult<()> {
+        super::errors::register(module)?;
+        super::routes::register(module)?;
+        module.add_class::<super::ResponsesWebSocketConnection>()?;
+        super::diagnostics::register(module)
+    }
 }
 
 #[cfg(test)]
 mod tests {
+    use std::ffi::CString;
+    use std::time::Duration;
+
+    use futures_util::{SinkExt, StreamExt};
+    use pyo3::types::PyDict;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::{accept_async, tungstenite::Message};
+
     use super::*;
 
     #[test]
     fn module_registration_preserves_the_public_surface() {
         Python::initialize();
         Python::attach(|py| {
-            let module = PyModule::new(py, "_native").expect("module should be created");
-            _native(&module).expect("module should register");
+            let module = pyo3::wrap_pymodule!(_native)(py).into_bound(py);
 
             let expected = [
                 "RustBridgeDeclined",
@@ -93,13 +106,79 @@ mod tests {
                 "gil_stats",
             ];
 
-            for name in expected {
-                assert!(
-                    module
-                        .hasattr(name)
-                        .expect("attribute lookup should succeed")
-                );
-            }
+            let public_names: Vec<String> = module
+                .dict()
+                .keys()
+                .extract::<Vec<String>>()
+                .expect("module names should be strings")
+                .into_iter()
+                .filter(|name| !name.starts_with("__"))
+                .collect();
+            assert_eq!(public_names, expected);
         });
     }
+
+    #[test]
+    fn responses_websocket_connection_round_trips_through_python() {
+        Python::initialize();
+        let runtime = pyo3_async_runtimes::tokio::get_runtime();
+        let listener = runtime
+            .block_on(TcpListener::bind("127.0.0.1:0"))
+            .expect("listener should bind");
+        let address = listener
+            .local_addr()
+            .expect("listener should have an address");
+        let server = runtime.spawn(async move {
+            let (stream, _) = listener.accept().await.expect("server should accept");
+            let mut socket = accept_async(stream)
+                .await
+                .expect("handshake should succeed");
+
+            let message = socket
+                .next()
+                .await
+                .expect("client should send a frame")
+                .expect("client frame should be valid");
+            assert_eq!(message, Message::Text("from-python".into()));
+            socket
+                .send(Message::Text("from-server".into()))
+                .await
+                .expect("server should reply");
+            assert!(matches!(socket.next().await, Some(Ok(Message::Close(_)))));
+        });
+
+        Python::attach(|py| {
+            let module = pyo3::wrap_pymodule!(_native)(py).into_bound(py);
+            let locals = PyDict::new(py);
+            locals
+                .set_item("native", &module)
+                .expect("module should enter Python locals");
+            locals
+                .set_item("url", format!("ws://{address}"))
+                .expect("URL should enter Python locals");
+            let code = CString::new(
+                r#"
+import asyncio
+
+async def exercise():
+    connection = await native.ResponsesWebSocketConnection.connect(url)
+    assert type(connection) is native.ResponsesWebSocketConnection
+    await connection.send_text("from-python")
+    assert await connection.recv_text() == "from-server"
+    await connection.close()
+    assert await connection.recv_text() is None
+
+asyncio.run(asyncio.wait_for(exercise(), timeout=5))
+"#,
+            )
+            .expect("Python source should not contain null bytes");
+            py.run(&code, Some(&locals), Some(&locals))
+                .expect("Python WebSocket methods should round trip");
+        });
+
+        runtime
+            .block_on(async { tokio::time::timeout(Duration::from_secs(5), server).await })
+            .expect("server should finish")
+            .expect("server task should not panic");
+    }
 }

litellm-rust/crates/python-bridge/src/marshal.rs

@@ -30,12 +30,9 @@ pub(crate) fn optional_timeout(timeout_seconds: Option<f64>) -> Option<Duration>
     })
 }
 
-pub(crate) fn marshal_headers(
-    py: Python<'_>,
-    headers: Option<Py<PyAny>>,
-) -> PyResult<HashMap<String, String>> {
+pub(crate) fn marshal_headers(headers: Option<Value>) -> PyResult<HashMap<String, String>> {
     let value = match headers {
-        Some(headers) => from_py(headers.bind(py))?,
+        Some(headers) => headers,
         None => Value::Object(Map::new()),
     };
     let Value::Object(headers) = value else {

litellm-rust/crates/python-bridge/src/routes/audio_transcription.rs

@@ -1,12 +1,14 @@
 use litellm_ai_gateway::io::audio_transcription::{
     AudioTranscriptionRequest, audio_transcription as run_audio_transcription,
 };
-use litellm_python_interop::{from_py, release_gil, to_py};
+use litellm_python_interop::from_py;
 use pyo3::prelude::*;
 
 use crate::errors::core_error_to_pyerr;
 use crate::marshal::{optional_object_to_map, optional_timeout};
 
+use super::runtime::{run_async, run_sync};
+
 #[pyfunction]
 #[pyo3(signature = (model, audio, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, timeout_seconds=None))]
 #[allow(clippy::too_many_arguments)]
@@ -28,9 +30,10 @@ fn transcription(
     };
     let optional_params = optional_object_to_map(py, "optional_params", optional_params)?;
     let timeout = optional_timeout(timeout_seconds);
-    let result = release_gil(py, || {
-        pyo3_async_runtimes::tokio::get_runtime().block_on(run_audio_transcription(
-            AudioTranscriptionRequest {
+    run_sync(
+        py,
+        async move {
+            run_audio_transcription(AudioTranscriptionRequest {
                 model: &model,
                 audio,
                 api_key: api_key.as_deref(),
@@ -43,13 +46,11 @@ fn transcription(
                 guardrails: Vec::new(),
                 request_metadata: Default::default(),
                 litellm_call_id: None,
-            },
-        ))
-    });
-    match result {
-        Ok(value) => to_py(py, &value),
-        Err(err) => Err(core_error_to_pyerr(err)),
-    }
+            })
+            .await
+        },
+        core_error_to_pyerr,
+    )
 }
 
 #[pyfunction]
@@ -73,25 +74,27 @@ fn atranscription(
     };
     let optional_params = optional_object_to_map(py, "optional_params", optional_params)?;
     let timeout = optional_timeout(timeout_seconds);
-    pyo3_async_runtimes::tokio::future_into_py(py, async move {
-        let value = run_audio_transcription(AudioTranscriptionRequest {
-            model: &model,
-            audio,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            optional_params,
-            timeout,
-            callbacks: Vec::new(),
-            guardrails: Vec::new(),
-            request_metadata: Default::default(),
-            litellm_call_id: None,
-        })
-        .await
-        .map_err(core_error_to_pyerr)?;
-        Python::attach(|py| to_py(py, &value))
-    })
+    run_async(
+        py,
+        async move {
+            run_audio_transcription(AudioTranscriptionRequest {
+                model: &model,
+                audio,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                optional_params,
+                timeout,
+                callbacks: Vec::new(),
+                guardrails: Vec::new(),
+                request_metadata: Default::default(),
+                litellm_call_id: None,
+            })
+            .await
+        },
+        core_error_to_pyerr,
+    )
 }
 
 pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {

litellm-rust/crates/python-bridge/src/routes/chat_completions.rs

@@ -1,10 +1,10 @@
 use std::time::Duration;
 
-use litellm_core::chat_completions::types::{ChatCompletionsRequest, ChatCompletionsResponse};
+use litellm_core::chat_completions::types::ChatCompletionsRequest;
 use litellm_core::chat_completions::{
     chat_completions as run_chat_completions, chat_completions_decline_reason,
 };
-use litellm_python_interop::{from_py, release_gil, to_py};
+use litellm_python_interop::from_py;
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use serde_json::{Map, Value};
@@ -12,12 +12,7 @@ use serde_json::{Map, Value};
 use crate::errors::fallback_route_error_to_pyerr;
 use crate::marshal::{optional_object_to_map, optional_timeout};
 
-fn chat_completions_response_to_py(
-    py: Python<'_>,
-    response: ChatCompletionsResponse,
-) -> PyResult<Py<PyAny>> {
-    to_py(py, &response)
-}
+use super::runtime::{run_async, run_sync};
 
 type MarshaledChatCompletionsInputs = (
     Value,
@@ -95,9 +90,10 @@ fn chat_completions(
         timeout_seconds,
     )?;
 
-    let result = release_gil(py, || {
-        pyo3_async_runtimes::tokio::get_runtime().block_on(run_chat_completions(
-            ChatCompletionsRequest {
+    run_sync(
+        py,
+        async move {
+            run_chat_completions(ChatCompletionsRequest {
                 model: &model,
                 messages,
                 optional_params,
@@ -106,14 +102,11 @@ fn chat_completions(
                 custom_llm_provider: custom_llm_provider.as_deref(),
                 extra_headers,
                 timeout,
-            },
-        ))
-    });
-
-    match result {
-        Ok(response) => chat_completions_response_to_py(py, response),
-        Err(err) => Err(fallback_route_error_to_pyerr(err)),
-    }
+            })
+            .await
+        },
+        fallback_route_error_to_pyerr,
+    )
 }
 
 #[pyfunction]
@@ -138,22 +131,23 @@ fn achat_completions(
         timeout_seconds,
     )?;
 
-    pyo3_async_runtimes::tokio::future_into_py(py, async move {
-        let response = run_chat_completions(ChatCompletionsRequest {
-            model: &model,
-            messages,
-            optional_params,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            timeout,
-        })
-        .await
-        .map_err(fallback_route_error_to_pyerr)?;
-
-        Python::attach(|py| chat_completions_response_to_py(py, response))
-    })
+    run_async(
+        py,
+        async move {
+            run_chat_completions(ChatCompletionsRequest {
+                model: &model,
+                messages,
+                optional_params,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                timeout,
+            })
+            .await
+        },
+        fallback_route_error_to_pyerr,
+    )
 }
 
 pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {

litellm-rust/crates/python-bridge/src/routes/messages.rs

@@ -1,8 +1,8 @@
 use std::time::Duration;
 
 use litellm_core::messages::messages as run_messages;
-use litellm_core::messages::types::{AnthropicMessagesResponse, MessagesRequest};
-use litellm_python_interop::{from_py, release_gil, to_py};
+use litellm_core::messages::types::MessagesRequest;
+use litellm_python_interop::from_py;
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use serde_json::{Map, Value};
@@ -10,12 +10,7 @@ use serde_json::{Map, Value};
 use crate::errors::fallback_route_error_to_pyerr;
 use crate::marshal::{optional_object_to_map, optional_timeout};
 
-fn messages_response_to_py(
-    py: Python<'_>,
-    response: AnthropicMessagesResponse,
-) -> PyResult<Py<PyAny>> {
-    to_py(py, &response)
-}
+use super::runtime::{run_async, run_sync};
 
 type MarshaledMessagesInputs = (Value, Option<Map<String, Value>>, Option<Duration>);
 
@@ -52,22 +47,22 @@ fn messages(
     let (body, extra_headers, timeout) =
         marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?;
 
-    let result = release_gil(py, || {
-        pyo3_async_runtimes::tokio::get_runtime().block_on(run_messages(MessagesRequest {
-            model: &model,
-            body,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            timeout,
-        }))
-    });
-
-    match result {
-        Ok(response) => messages_response_to_py(py, response),
-        Err(err) => Err(fallback_route_error_to_pyerr(err)),
-    }
+    run_sync(
+        py,
+        async move {
+            run_messages(MessagesRequest {
+                model: &model,
+                body,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                timeout,
+            })
+            .await
+        },
+        fallback_route_error_to_pyerr,
+    )
 }
 
 #[pyfunction]
@@ -86,21 +81,22 @@ fn amessages(
     let (body, extra_headers, timeout) =
         marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?;
 
-    pyo3_async_runtimes::tokio::future_into_py(py, async move {
-        let response = run_messages(MessagesRequest {
-            model: &model,
-            body,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            timeout,
-        })
-        .await
-        .map_err(fallback_route_error_to_pyerr)?;
-
-        Python::attach(|py| messages_response_to_py(py, response))
-    })
+    run_async(
+        py,
+        async move {
+            run_messages(MessagesRequest {
+                model: &model,
+                body,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                timeout,
+            })
+            .await
+        },
+        fallback_route_error_to_pyerr,
+    )
 }
 
 pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {

litellm-rust/crates/python-bridge/src/routes/mod.rs

@@ -4,6 +4,7 @@ mod audio_transcription;
 mod chat_completions;
 mod messages;
 mod ocr;
+mod runtime;
 
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     ocr::register(module)?;

litellm-rust/crates/python-bridge/src/routes/ocr.rs

@@ -1,13 +1,15 @@
 use std::time::Duration;
 
 use litellm_ai_gateway::io::ocr::{OcrRequest, ocr as run_ocr};
-use litellm_python_interop::{from_py, release_gil, to_py};
+use litellm_python_interop::from_py;
 use pyo3::prelude::*;
 use serde_json::{Map, Value};
 
 use crate::errors::core_error_to_pyerr;
 use crate::marshal::{optional_object_to_map, optional_timeout};
 
+use super::runtime::{run_async, run_sync};
+
 type MarshaledOcrInputs = (
     Value,
     Option<Map<String, Value>>,
@@ -55,27 +57,27 @@ fn ocr(
         timeout_seconds,
     )?;
 
-    let result = release_gil(py, || {
-        pyo3_async_runtimes::tokio::get_runtime().block_on(run_ocr(OcrRequest {
-            model: &model,
-            document,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            optional_params,
-            timeout,
-            callbacks: Vec::new(),
-            guardrails: Vec::new(),
-            request_metadata: Default::default(),
-            litellm_call_id: None,
-        }))
-    });
-
-    match result {
-        Ok(value) => to_py(py, &value),
-        Err(err) => Err(core_error_to_pyerr(err)),
-    }
+    run_sync(
+        py,
+        async move {
+            run_ocr(OcrRequest {
+                model: &model,
+                document,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                optional_params,
+                timeout,
+                callbacks: Vec::new(),
+                guardrails: Vec::new(),
+                request_metadata: Default::default(),
+                litellm_call_id: None,
+            })
+            .await
+        },
+        core_error_to_pyerr,
+    )
 }
 
 #[pyfunction]
@@ -100,26 +102,27 @@ fn aocr(
         timeout_seconds,
     )?;
 
-    pyo3_async_runtimes::tokio::future_into_py(py, async move {
-        let value = run_ocr(OcrRequest {
-            model: &model,
-            document,
-            api_key: api_key.as_deref(),
-            api_base: api_base.as_deref(),
-            custom_llm_provider: custom_llm_provider.as_deref(),
-            extra_headers,
-            optional_params,
-            timeout,
-            callbacks: Vec::new(),
-            guardrails: Vec::new(),
-            request_metadata: Default::default(),
-            litellm_call_id: None,
-        })
-        .await
-        .map_err(core_error_to_pyerr)?;
-
-        Python::attach(|py| to_py(py, &value))
-    })
+    run_async(
+        py,
+        async move {
+            run_ocr(OcrRequest {
+                model: &model,
+                document,
+                api_key: api_key.as_deref(),
+                api_base: api_base.as_deref(),
+                custom_llm_provider: custom_llm_provider.as_deref(),
+                extra_headers,
+                optional_params,
+                timeout,
+                callbacks: Vec::new(),
+                guardrails: Vec::new(),
+                request_metadata: Default::default(),
+                litellm_call_id: None,
+            })
+            .await
+        },
+        core_error_to_pyerr,
+    )
 }
 
 pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {

litellm-rust/crates/python-bridge/src/routes/runtime.rs

@@ -0,0 +1,423 @@
+use std::future::Future;
+use std::panic::AssertUnwindSafe;
+use std::time::Duration;
+
+use futures_util::FutureExt;
+use litellm_core::error::{CoreError, CoreResult};
+use litellm_python_interop::{Pythonized, panic_to_pyerr, release_gil};
+use pyo3::exceptions::PyRuntimeError;
+use pyo3::prelude::*;
+use serde::Serialize;
+use tokio::runtime::{Handle, Runtime};
+use tokio::time::{self, MissedTickBehavior};
+
+pub(super) fn run_sync<T, F>(
+    py: Python<'_>,
+    future: F,
+    map_error: fn(CoreError) -> PyErr,
+) -> PyResult<Py<PyAny>>
+where
+    T: Serialize + Send + 'static,
+    F: Future<Output = CoreResult<T>> + Send + 'static,
+{
+    run_sync_on(
+        py,
+        pyo3_async_runtimes::tokio::get_runtime(),
+        future,
+        map_error,
+    )
+}
+
+fn run_sync_on<T, F>(
+    py: Python<'_>,
+    runtime: &Runtime,
+    future: F,
+    map_error: fn(CoreError) -> PyErr,
+) -> PyResult<Py<PyAny>>
+where
+    T: Serialize + Send + 'static,
+    F: Future<Output = CoreResult<T>> + Send + 'static,
+{
+    if Handle::try_current().is_ok() {
+        return Err(PyRuntimeError::new_err(
+            "synchronous native routes cannot run from a Tokio context; use the async route",
+        ));
+    }
+
+    let result = release_gil(py, move || runtime.block_on(wait_for_sync_result(future)))?;
+    let result = map_core_result(result, map_error)?;
+    Pythonized(result).into_pyobject(py).map(Bound::unbind)
+}
+
+pub(super) fn run_async<T, F>(
+    py: Python<'_>,
+    future: F,
+    map_error: fn(CoreError) -> PyErr,
+) -> PyResult<Bound<'_, PyAny>>
+where
+    T: Serialize + Send + 'static,
+    F: Future<Output = CoreResult<T>> + Send + 'static,
+{
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let result = catch_route_panic(future).await?;
+        let result = map_core_result(result, map_error)?;
+        Ok(Pythonized(result))
+    })
+}
+
+fn map_core_result<T>(result: CoreResult<T>, map_error: fn(CoreError) -> PyErr) -> PyResult<T> {
+    match result {
+        Ok(value) => Ok(value),
+        Err(error) => Err(
+            std::panic::catch_unwind(AssertUnwindSafe(|| map_error(error)))
+                .map_err(panic_to_pyerr)?,
+        ),
+    }
+}
+
+async fn catch_route_panic<T, F>(future: F) -> PyResult<CoreResult<T>>
+where
+    F: Future<Output = CoreResult<T>>,
+{
+    AssertUnwindSafe(future)
+        .catch_unwind()
+        .await
+        .map_err(panic_to_pyerr)
+}
+
+async fn wait_for_sync_result<T, F>(future: F) -> PyResult<CoreResult<T>>
+where
+    F: Future<Output = CoreResult<T>>,
+{
+    let future = catch_route_panic(future);
+    tokio::pin!(future);
+
+    let signal_interval = Duration::from_millis(50);
+    let mut signal_checks =
+        time::interval_at(time::Instant::now() + signal_interval, signal_interval);
+    signal_checks.set_missed_tick_behavior(MissedTickBehavior::Delay);
+    loop {
+        tokio::select! {
+            result = &mut future => return result,
+            _ = signal_checks.tick() => Python::attach(|py| py.check_signals())?,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::ffi::CString;
+    use std::future::poll_fn;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::{Arc, mpsc};
+    use std::task::Poll;
+    use std::thread;
+    use std::time::Instant;
+
+    use pyo3::panic::PanicException;
+    use pyo3::types::{PyDict, PyModule};
+    use serde::Serializer;
+    use tokio::runtime::Builder;
+
+    use super::*;
+
+    fn runtime_error(error: CoreError) -> PyErr {
+        PyRuntimeError::new_err(error.to_string())
+    }
+
+    fn panicking_error_mapper(_error: CoreError) -> PyErr {
+        panic!("error mapper panicked")
+    }
+
+    struct PanickingOutput;
+
+    static ASYNC_PROBE_COMPLETED: AtomicUsize = AtomicUsize::new(0);
+
+    impl Serialize for PanickingOutput {
+        fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
+        where
+            S: Serializer,
+        {
+            panic!("serializer panicked")
+        }
+    }
+
+    #[pyfunction]
+    fn async_serialization_panic(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
+        run_async(py, async { Ok(PanickingOutput) }, runtime_error)
+    }
+
+    #[pyfunction]
+    fn async_runtime_probe(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
+        run_async(
+            py,
+            async {
+                ASYNC_PROBE_COMPLETED.fetch_add(1, Ordering::SeqCst);
+                Ok(true)
+            },
+            runtime_error,
+        )
+    }
+
+    #[pyfunction]
+    fn runtime_worker_count() -> usize {
+        pyo3_async_runtimes::tokio::get_runtime()
+            .metrics()
+            .num_workers()
+    }
+
+    #[pyfunction]
+    fn runtime_is_responsive(_py: Python<'_>, expected_completions: usize) -> bool {
+        let completion_deadline = Instant::now() + Duration::from_secs(2);
+        while ASYNC_PROBE_COMPLETED.load(Ordering::SeqCst) < expected_completions {
+            if Instant::now() >= completion_deadline {
+                return false;
+            }
+            thread::sleep(Duration::from_millis(1));
+        }
+
+        let (heartbeat_tx, heartbeat_rx) = mpsc::sync_channel(1);
+        pyo3_async_runtimes::tokio::get_runtime().spawn(async move {
+            let _ = heartbeat_tx.send(());
+        });
+        heartbeat_rx.recv_timeout(Duration::from_secs(2)).is_ok()
+    }
+
+    fn extract_bool(py: Python<'_>, result: PyResult<Py<PyAny>>) -> bool {
+        result
+            .expect("route should complete")
+            .bind(py)
+            .extract()
+            .expect("result should convert")
+    }
+
+    #[test]
+    fn sync_runner_polls_future_on_the_caller_thread() {
+        Python::initialize();
+        Python::attach(|py| {
+            let caller_thread = std::thread::current().id();
+            let result = run_sync(
+                py,
+                async move { Ok(std::thread::current().id() == caller_thread) },
+                runtime_error,
+            );
+
+            assert!(extract_bool(py, result));
+        });
+    }
+
+    #[test]
+    fn sync_runner_releases_gil_while_waiting() {
+        Python::initialize();
+        Python::attach(|py| {
+            let result = run_sync(
+                py,
+                async {
+                    let gil_acquired = tokio::time::timeout(
+                        Duration::from_secs(2),
+                        tokio::task::spawn_blocking(|| Python::attach(|_| true)),
+                    )
+                    .await;
+                    Ok(matches!(gil_acquired, Ok(Ok(true))))
+                },
+                runtime_error,
+            );
+
+            assert!(extract_bool(py, result));
+        });
+    }
+
+    #[test]
+    fn sync_runner_rejects_calls_from_a_tokio_context() {
+        Python::initialize();
+        let runtime = Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime should build");
+
+        let error = runtime.block_on(async {
+            Python::attach(|py| {
+                run_sync::<bool, _>(py, async { Ok(true) }, runtime_error)
+                    .expect_err("sync route should reject a nested Tokio runtime")
+            })
+        });
+
+        assert_eq!(
+            error.to_string(),
+            "RuntimeError: synchronous native routes cannot run from a Tokio context; use the async route"
+        );
+    }
+
+    #[test]
+    fn sync_runner_can_drive_a_current_thread_runtime() {
+        Python::initialize();
+        let runtime = Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime should build");
+        Python::attach(|py| {
+            let result = run_sync_on(
+                py,
+                &runtime,
+                async {
+                    tokio::task::yield_now().await;
+                    Ok(true)
+                },
+                runtime_error,
+            );
+            assert!(extract_bool(py, result));
+        });
+    }
+
+    #[test]
+    fn sync_runner_maps_a_panicked_future() {
+        Python::initialize();
+        Python::attach(|py| {
+            let error = run_sync::<bool, _>(
+                py,
+                poll_fn(|_| -> Poll<CoreResult<bool>> { panic!("route future panicked") }),
+                runtime_error,
+            )
+            .expect_err("panicked route should become a Python exception");
+
+            assert!(error.is_instance_of::<PanicException>(py));
+            assert_eq!(error.to_string(), "PanicException: route future panicked");
+        });
+    }
+
+    #[test]
+    fn sync_runner_maps_a_panicked_error_mapper() {
+        Python::initialize();
+        Python::attach(|py| {
+            let error = run_sync::<bool, _>(
+                py,
+                async { Err(CoreError::InvalidRequest("invalid".to_string())) },
+                panicking_error_mapper,
+            )
+            .expect_err("panicked mapper should become a Python exception");
+
+            assert!(error.is_instance_of::<PanicException>(py));
+            assert_eq!(error.to_string(), "PanicException: error mapper panicked");
+        });
+    }
+
+    #[test]
+    fn sync_runner_surfaces_serializer_panics() {
+        Python::initialize();
+        Python::attach(|py| {
+            let error = run_sync(py, async { Ok(PanickingOutput) }, runtime_error)
+                .expect_err("serializer panic should become a Python exception");
+
+            assert!(error.is_instance_of::<PanicException>(py));
+            assert_eq!(error.to_string(), "PanicException: serializer panicked");
+        });
+    }
+
+    #[test]
+    fn sync_runner_supports_concurrent_callers_on_the_shared_runtime() {
+        Python::initialize();
+        let barrier = Arc::new(tokio::sync::Barrier::new(2));
+        let callers: Vec<_> = (0..2)
+            .map(|_| {
+                let barrier = Arc::clone(&barrier);
+                thread::spawn(move || {
+                    Python::attach(|py| {
+                        extract_bool(
+                            py,
+                            run_sync(
+                                py,
+                                async move {
+                                    Ok(tokio::time::timeout(Duration::from_secs(2), barrier.wait())
+                                        .await
+                                        .is_ok())
+                                },
+                                runtime_error,
+                            ),
+                        )
+                    })
+                })
+            })
+            .collect();
+        let results: Vec<_> = callers
+            .into_iter()
+            .map(|caller| caller.join().expect("caller should not panic"))
+            .collect();
+
+        assert_eq!(results, vec![true, true]);
+    }
+
+    #[test]
+    fn async_runner_surfaces_serializer_panics() {
+        Python::initialize();
+        Python::attach(|py| {
+            let module = PyModule::new(py, "runtime").expect("module should be created");
+            module
+                .add_function(
+                    wrap_pyfunction!(async_serialization_panic, &module)
+                        .expect("function should wrap"),
+                )
+                .expect("function should register");
+            let locals = PyDict::new(py);
+            locals
+                .set_item("runtime", &module)
+                .expect("module should enter Python locals");
+            let code = CString::new(
+                r#"
+import asyncio
+
+async def exercise():
+    try:
+        await runtime.async_serialization_panic()
+    except BaseException as error:
+        assert type(error).__name__ == "PanicException"
+        assert str(error) == "serializer panicked"
+    else:
+        raise AssertionError("serializer panic was not raised")
+
+asyncio.run(exercise())
+"#,
+            )
+            .expect("Python source should not contain null bytes");
+            py.run(&code, Some(&locals), Some(&locals))
+                .expect("serializer panic should reach the Python awaiter");
+        });
+    }
+
+    #[test]
+    fn async_result_delivery_does_not_stall_tokio_workers() {
+        Python::initialize();
+        ASYNC_PROBE_COMPLETED.store(0, Ordering::SeqCst);
+        Python::attach(|py| {
+            let module = PyModule::new(py, "runtime").expect("module should be created");
+            for function in [
+                wrap_pyfunction!(async_runtime_probe, &module).expect("function should wrap"),
+                wrap_pyfunction!(runtime_worker_count, &module).expect("function should wrap"),
+                wrap_pyfunction!(runtime_is_responsive, &module).expect("function should wrap"),
+            ] {
+                module
+                    .add_function(function)
+                    .expect("function should register");
+            }
+            let locals = PyDict::new(py);
+            locals
+                .set_item("runtime", &module)
+                .expect("module should enter Python locals");
+            let code = CString::new(
+                r#"
+import asyncio
+
+async def exercise():
+    worker_count = runtime.runtime_worker_count()
+    awaitables = [runtime.async_runtime_probe() for _ in range(worker_count)]
+    assert runtime.runtime_is_responsive(worker_count)
+    assert await asyncio.gather(*awaitables) == [True] * worker_count
+
+asyncio.run(exercise())
+"#,
+            )
+            .expect("Python source should not contain null bytes");
+            py.run(&code, Some(&locals), Some(&locals))
+                .expect("result delivery should leave Tokio workers responsive");
+        });
+    }
+}

litellm-rust/crates/python-interop/src/lib.rs

@@ -2,4 +2,4 @@ mod gil;
 mod marshal;
 
 pub use gil::{release_count, release_gil};
-pub use marshal::{from_py, to_py};
+pub use marshal::{Pythonized, from_py, panic_to_pyerr, to_py};

litellm-rust/crates/python-interop/src/marshal.rs

@@ -1,4 +1,8 @@
+use std::any::Any;
+use std::panic::{AssertUnwindSafe, catch_unwind};
+
 use pyo3::exceptions::PyValueError;
+use pyo3::panic::PanicException;
 use pyo3::prelude::*;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
@@ -18,3 +22,71 @@ where
         .map(Bound::unbind)
         .map_err(|error| PyValueError::new_err(error.to_string()))
 }
+
+pub struct Pythonized<T>(pub T);
+
+impl<'py, T> IntoPyObject<'py> for Pythonized<T>
+where
+    T: Serialize,
+{
+    type Target = PyAny;
+    type Output = Bound<'py, PyAny>;
+    type Error = PyErr;
+
+    fn into_pyobject(self, py: Python<'py>) -> PyResult<Self::Output> {
+        catch_unwind(AssertUnwindSafe(|| pythonize::pythonize(py, &self.0)))
+            .map_err(panic_to_pyerr)?
+            .map_err(|error| PyValueError::new_err(error.to_string()))
+    }
+}
+
+pub fn panic_to_pyerr(payload: Box<dyn Any + Send>) -> PyErr {
+    let message = payload
+        .downcast_ref::<String>()
+        .map(String::as_str)
+        .or_else(|| payload.downcast_ref::<&str>().copied())
+        .unwrap_or("panic from Rust code");
+    PanicException::new_err(message.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use serde::Serializer;
+
+    use super::*;
+
+    struct PanickingSerializer;
+
+    impl Serialize for PanickingSerializer {
+        fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
+        where
+            S: Serializer,
+        {
+            panic!("serializer panicked")
+        }
+    }
+
+    #[test]
+    fn pythonized_converts_on_the_attached_thread() {
+        Python::initialize();
+        Python::attach(|py| {
+            let value: Vec<i32> = Pythonized(vec![1, 2, 3])
+                .into_pyobject(py)
+                .and_then(|value| value.extract())
+                .expect("value should convert");
+            assert_eq!(value, vec![1, 2, 3]);
+        });
+    }
+
+    #[test]
+    fn pythonized_maps_serializer_panics_to_a_base_exception() {
+        Python::initialize();
+        Python::attach(|py| {
+            let error = Pythonized(PanickingSerializer)
+                .into_pyobject(py)
+                .expect_err("serializer panic should become a Python exception");
+            assert!(error.is_instance_of::<PanicException>(py));
+            assert_eq!(error.to_string(), "PanicException: serializer panicked");
+        });
+    }
+}