diff --git a/docs/platforms/python/tracing/configure-sampling/index.mdx b/docs/platforms/python/tracing/configure-sampling/index.mdx
index 0627639b8f1a4..3cd27fac4961f 100644
--- a/docs/platforms/python/tracing/configure-sampling/index.mdx
+++ b/docs/platforms/python/tracing/configure-sampling/index.mdx
@@ -8,23 +8,29 @@ If you find that Sentry's tracing functionality is generating too much data, for
 
 Effective sampling is key to getting the most value from Sentry's performance monitoring while minimizing overhead. The Python SDK provides two ways to control the sampling rate. You can review the options and [examples](#traces-sampler-examples) below.
 
+<Alert>
+
+This page covers both transaction mode (default) and stream mode. See <PlatformLink to="/tracing/new-spans/">New Spans</PlatformLink> to learn more.
+
+</Alert>
+
 ## Sampling Configuration Options
 
 ### 1. Uniform Sample Rate (`traces_sample_rate`)
 
-`traces_sample_rate` is a floating-point value between `0.0` and `1.0`, inclusive, which controls the probability with which each transaction will be sampled:
+`traces_sample_rate` is a floating-point value between `0.0` and `1.0`, inclusive, which controls the probability with which each transaction (or service spans in <PlatformLink to="/tracing/new-spans">stream mode</PlatformLink>) will be sampled. This works the same way in both transaction mode and stream mode:
 
 <PlatformContent includePath="/performance/traces-sample-rate" />
 
-With `traces_sample_rate` set to `0.25`, each transaction in your application is randomly sampled with a probability of `0.25`, so you can expect that one in every four transactions will be sent to Sentry.
+With `traces_sample_rate` set to `0.25`, each transaction/service span in your application is randomly sampled with a probability of `0.25`, so you can expect that one in every four transactions/service spans will be sent to Sentry.
 
 ### 2. Sampling Function (`traces_sampler`)
 
 For more granular control, you can provide a `traces_sampler` function. This approach allows you to:
 
-- Apply different sampling rates to different types of transactions
-- Filter out specific transactions entirely
-- Make sampling decisions based on transaction data
+- Apply different sampling rates to different types of transactions/service spans
+- Filter out specific transactions/service spans entirely
+- Make sampling decisions based on transaction/service span data
 - Control the inheritance of sampling decisions in distributed traces
 - Use custom attributes to modify sampling
 
@@ -38,14 +44,11 @@ In distributed systems, implementing inheritance logic when trace information is
 
 <PlatformContent includePath="/performance/traces-sampler-as-sampler" />
 
-<details>
-<summary className="text-xl font-semibold">Traces Sampler Examples</summary>
-
-#### Traces Sampler Examples
+<Expandable title="Traces Sampler Examples">
 
 1. Prioritizing Critical User Flows
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -73,14 +76,49 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
     return 0.1
 
 sentry_sdk.init(
-    dsn="your-dsn",
+    # ...
     traces_sampler=traces_sampler,
 )
 ```
 
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    span_ctx = sampling_context["span_context"]
+    name = span_ctx["name"]
+    op = span_ctx["attributes"].get("sentry.op")
+
+    # Sample all checkout spans
+    if name and ('/checkout' in name or op == 'checkout'):
+        return 1.0
+
+    # Sample 50% of login spans
+    if name and ('/login' in name or op == 'login'):
+        return 0.5
+
+    # Sample 10% of everything else
+    return 0.1
+
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+```
+
 2. Handling Different Environments and Error Rates
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
+import os
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -92,7 +130,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
     if parent_sampling_decision is not None:
         return float(parent_sampling_decision)
 
-    custom_sampling_ctx = sampling_context["custom_sampling_context"]
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
     environment = os.environ.get("ENVIRONMENT", "development")
 
     # Sample all transactions in development
@@ -115,7 +153,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
 
 # Initialize the SDK with the sampling function
 sentry_sdk.init(
-    dsn="your-dsn",
+    # ...
     traces_sampler=traces_sampler,
 )
 
@@ -130,9 +168,62 @@ with sentry_sdk.start_transaction(
     # your code here
 ```
 
+```python {tabTitle:Stream Mode}
+import os
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
+    environment = os.environ.get("ENVIRONMENT", "development")
+
+    # Sample all spans in development
+    if environment == "development":
+        return 1.0
+
+    # Sample more spans if there are recent errors
+    # Note: hasRecentErrors is a custom attribute that needs to be set
+    if custom_sampling_ctx.get("hasRecentErrors") is True:
+        return 0.8
+
+    # Sample based on environment
+    if environment == "production":
+        return 0.05  # 5% in production
+    elif environment == "staging":
+        return 0.2   # 20% in staging
+
+    # Default sampling rate
+    return 0.1
+
+# Initialize the SDK with the sampling function
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+
+# Custom attributes need to be set on the scope, after continue_trace
+# (which resets the propagation context) and before start_span
+# (which is when sampling happens), in order to be available
+# in the traces_sampler
+sentry_sdk.Scope.set_custom_sampling_context({"hasRecentErrors": True})
+with sentry_sdk.traces.start_span(
+    name="GET /api/users",
+    attributes={"sentry.op": "http.request"},
+) as span:
+    # your code here
+```
+
 3. Controlling Sampling Based on User and Transaction Properties
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -144,7 +235,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
     if parent_sampling_decision is not None:
         return float(parent_sampling_decision)
 
-    custom_sampling_ctx = sampling_context["custom_sampling_context"]
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
 
     # Always sample for premium users
     # Note: user.tier is a custom attribute that needs to be set
@@ -166,7 +257,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
 
 # Initialize the SDK with the sampling function
 sentry_sdk.init(
-    dsn="your-dsn",
+    # ...
     traces_sampler=traces_sampler,
 )
 
@@ -179,9 +270,59 @@ with sentry_sdk.start_transaction(
     # Your code here
 ```
 
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
+
+    # Always sample for premium users
+    # Note: user.tier is a custom attribute that needs to be set
+    if custom_sampling_ctx.get("user", {}).get("tier") == "premium":
+        return 1.0
+
+    # Sample more spans for users experiencing errors
+    # Note: hasRecentErrors is a custom attribute
+    if custom_sampling_ctx.get("hasRecentErrors") is True:
+        return 0.8
+
+    # Sample less for high-volume, low-value paths
+    name = sampling_context["span_context"]["name"]
+    if name and name.startswith("/api/metrics"):
+        return 0.01
+
+    # Default sampling rate
+    return 0.2
+
+# Initialize the SDK with the sampling function
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+
+# To set custom attributes for this example:
+sentry_sdk.Scope.set_custom_sampling_context(
+    {"user": {"tier": "premium"}, "hasRecentErrors": True}
+)
+with sentry_sdk.traces.start_span(
+    name="GET /api/users",
+    attributes={"sentry.op": "http.request"},
+) as span:
+    # Your code here
+```
+
 4. Complex Business Logic Sampling
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -199,7 +340,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
     if transaction_ctx["op"] in ["payment.process", "order.create", "user.verify"]:
         return 1.0
 
-    custom_sampling_context = sampling_context["custom_sampling_context"]
+    custom_sampling_context = sampling_context.get("custom_sampling_context") or {}
 
     # Sample based on user segment
     # Note: user.segment is a custom attribute
@@ -226,7 +367,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
 
 # Initialize the SDK with the sampling function
 sentry_sdk.init(
-    dsn="your-dsn",
+    # ...
     traces_sampler=traces_sampler,
 )
 
@@ -237,12 +378,72 @@ with sentry_sdk.start_transaction(
     custom_sampling_context={"user": {"segment": "enterprise"}, "transaction": {"value": 1500}, "service": {"error_rate": 0.03}},
 ) as transaction:
     # Your code here
+```
 
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    # Always sample critical business operations
+    # Note: sentry.op is an SDK-provided attribute
+    span_ctx = sampling_context["span_context"]
+    if span_ctx["attributes"].get("sentry.op") in ["payment.process", "order.create", "user.verify"]:
+        return 1.0
+
+    custom_sampling_context = sampling_context.get("custom_sampling_context") or {}
+
+    # Sample based on user segment
+    # Note: user.segment is a custom attribute
+    user_segment = custom_sampling_context.get("user", {}).get("segment")
+    if user_segment == "enterprise":
+        return 0.8
+    elif user_segment == "premium":
+        return 0.5
+
+    # Sample based on transaction value
+    # Note: transaction.value is a custom attribute
+    transaction_value = custom_sampling_context.get("transaction", {}).get("value")
+    if transaction_value is not None and transaction_value > 1000:  # High-value transactions
+        return 0.7
+
+    # Sample based on error rate in the service
+    # Note: service.error_rate is a custom attribute
+    error_rate = custom_sampling_context.get("service", {}).get("error_rate")
+    if error_rate is not None and error_rate > 0.05:  # Error rate above 5%
+        return 0.9
+
+    # Default sampling rate
+    return 0.1
+
+# Initialize the SDK with the sampling function
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+
+# To set custom attributes for this example:
+sentry_sdk.Scope.set_custom_sampling_context(
+    {"user": {"segment": "enterprise"}, "transaction": {"value": 1500}, "service": {"error_rate": 0.03}}
+)
+with sentry_sdk.traces.start_span(
+    name="Process Payment",
+    attributes={"sentry.op": "payment.process"},
+) as span:
+    # Your code here
 ```
 
 5. Performance-Based Sampling
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -254,7 +455,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
     if parent_sampling_decision is not None:
         return float(parent_sampling_decision)
 
-    custom_sampling_ctx = sampling_context["custom_sampling_context"]
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
 
     # Sample more transactions with high memory usage
     # Note: memory_usage_mb is a custom attribute
@@ -279,7 +480,7 @@ def traces_sampler(sampling_context: SamplingContext) -> float:
 
 # Initialize the SDK with the sampling function
 sentry_sdk.init(
-    dsn="your-dsn",
+    # ...
     traces_sampler=traces_sampler,
 )
 
@@ -291,12 +492,68 @@ with sentry_sdk.start_transaction(
 ) as transaction:
     # Your code here
 ```
-</details>
+
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    custom_sampling_ctx = sampling_context.get("custom_sampling_context") or {}
+
+    # Sample more spans with high memory usage
+    # Note: memory_usage_mb is a custom attribute
+    memory_usage = custom_sampling_ctx.get("memory_usage_mb")
+    if memory_usage is not None and memory_usage > 500:
+        return 0.8
+
+    # Sample more spans with high CPU usage
+    # Note: cpu_percent is a custom attribute
+    cpu_percent = custom_sampling_ctx.get("cpu_percent")
+    if cpu_percent is not None and cpu_percent > 80:
+        return 0.8
+
+    # Sample more spans with high database load
+    # Note: db_connections is a custom attribute
+    db_connections = custom_sampling_ctx.get("db_connections")
+    if db_connections is not None and db_connections > 100:
+        return 0.7
+
+    # Default sampling rate
+    return 0.1
+
+# Initialize the SDK with the sampling function
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+
+# To set custom attributes for this example:
+sentry_sdk.Scope.set_custom_sampling_context(
+    {"memory_usage_mb": 600, "cpu_percent": 85, "db_connections": 120}
+)
+with sentry_sdk.traces.start_span(
+    name="Process Data",
+    attributes={"sentry.op": "data.process"},
+) as span:
+    # Your code here
+```
+
+</Expandable>
 
 ## The Sampling Context Object
 
 When the `traces_sampler` function is called, the Sentry SDK passes a `sampling_context` object with information from the relevant span to help make sampling decisions:
 
+### Transaction Mode (Default)
+
 ```python
 {
     "transaction_context": {
@@ -310,28 +567,76 @@ When the `traces_sampler` function is called, the Sentry SDK passes a `sampling_
 }
 ```
 
+### Stream Mode
+
+```python
+{
+    "span_context": {
+        "name": str,  # span title at creation time (SDK-provided)
+        "trace_id": str,  # the trace ID (SDK-provided)
+        "parent_span_id": Optional[str],  # the parent span's ID, if any (SDK-provided)
+        "parent_sampled": Optional[bool],  # whether the parent span was sampled (SDK-provided)
+        "parent_sample_rate": Optional[float],  # the sample rate used by the parent (SDK-provided)
+        "attributes": dict[str, Any]  # attributes set at span-creation time (SDK- and user-provided)
+    },
+    "custom_sampling_context": Optional[dict[str, Any]]  # additional custom data for sampling
+}
+```
+
 ### SDK-Provided vs. Custom Attributes
 
 The sampling context contains both SDK-provided attributes and custom attributes:
 
+#### Transaction Mode (Default)
+
 **SDK-Provided Attributes:**
+
 - `transaction_context.name`: The name of the transaction
 - `transaction_context.op`: The operation type
 - `parent_sampled`: Whether the parent transaction was sampled
 - `parent_sample_rate`: The sample rate used by the parent
 
 **Custom Attributes:**
+
 - Any data you add to the `custom_sampling_context` parameter in `start_transaction`. Use this for data that you want to use for sampling decisions but don't want to include in the transaction data that gets sent to Sentry. Read more about sampling context [here](/platforms/python/configuration/sampling/#sampling-context).
 
+#### Stream Mode
+
+**SDK-Provided Attributes:**
+
+- `span_context.name`: The name of the span
+- `span_context.attributes`: Attributes set on the span at creation time, including `sentry.op`
+- `span_context.parent_sampled`: Whether the parent span was sampled
+- `span_context.parent_sample_rate`: The sample rate used by the parent
+
+**Custom Attributes:**
+
+- Any data you add via `sentry_sdk.Scope.set_custom_sampling_context()`, called after `continue_trace` and before `start_span`. Use this for data that you want to use for sampling decisions but don't want to include in the span data that gets sent to Sentry. Read more about sampling context [here](/platforms/python/configuration/sampling/#sampling-context).
+
 ## Sampling Decision Precedence
 
 When multiple sampling mechanisms could apply, Sentry follows this order of precedence:
 
-1. If a sampling decision is passed to `start_transaction`, that decision is used
-2. If `traces_sampler` is defined, its decision is used. Although the `traces_sampler` can override the parent sampling decision, most users will want to ensure their `traces_sampler` respects the parent sampling decision
-3. If no `traces_sampler` is defined, but there is a parent sampling decision from an incoming distributed trace, we use the parent sampling decision
-4. If neither of the above, `traces_sample_rate` is used
-5. If none of the above are set, no transactions are sampled. This is equivalent to setting `traces_sample_rate=0.0`
+### Transaction Mode (Default)
+
+1. If a sampling decision is passed to `start_transaction`, that decision is used.
+2. If `traces_sampler` is defined, its decision is used. Although the `traces_sampler` can override the parent sampling decision, most users will want to ensure their `traces_sampler` respects the parent sampling decision.
+3. If no `traces_sampler` is defined, but there is a parent sampling decision from an incoming distributed trace, we use the parent sampling decision.
+4. If neither of the above, `traces_sample_rate` is used.
+5. If none of the above are set, no transactions are sampled. This is equivalent to setting `traces_sample_rate=0.0`.
+
+### Stream Mode
+
+1. If `traces_sampler` is defined, its decision is used. Although the `traces_sampler` can override the parent sampling decision, most users will want to ensure their `traces_sampler` respects the parent sampling decision.
+2. If no `traces_sampler` is defined, but there is a parent sampling decision from an incoming distributed trace, we use the parent sampling decision.
+3. If neither of the above, `traces_sample_rate` is used
+4. If none of the above are set, no spans are sampled. This is equivalent to setting `traces_sample_rate=0.0`.
+
+<Alert level="info">
+
+Sampling decisions are made when spans are created. Child spans inherit the sampling decision of their parent span unless filtered by `ignore_spans`.
+
+</Alert>
 
 ## How Sampling Propagates in Distributed Traces
 
@@ -341,7 +646,8 @@ Sentry uses a "head-based" sampling approach:
 - This decision is propagated to all downstream services
 
 The two key headers are:
+
 - `sentry-trace`: Contains trace ID, span ID, and sampling decision
 - `baggage`: Contains additional trace metadata including sample rate
 
-The Sentry Python SDK automatically attaches these headers to outgoing HTTP requests when using auto-instrumentation with libraries like `requests`, `urllib3`, or `httpx`. For other communication channels, you can manually propagate trace information. Learn more about customizing tracing in [custom trace propagation](/platforms/python/tracing/distributed-tracing/custom-trace-propagation/)
+The Sentry Python SDK automatically attaches these headers to outgoing HTTP requests when using auto-instrumentation with libraries like `requests`, `urllib3`, or `httpx`. For other communication channels, you can manually propagate trace information. Learn more about customizing tracing in [custom trace propagation](/platforms/python/tracing/distributed-tracing/custom-trace-propagation/).
diff --git a/docs/platforms/python/tracing/span-metrics/examples.mdx b/docs/platforms/python/tracing/span-metrics/examples.mdx
index 5fe2452ebcc23..74cf671c45aa8 100644
--- a/docs/platforms/python/tracing/span-metrics/examples.mdx
+++ b/docs/platforms/python/tracing/span-metrics/examples.mdx
@@ -12,6 +12,11 @@ These examples assume you have already <PlatformLink to="/tracing/">set up traci
 
 This guide provides practical examples of using span attributes and metrics to solve common monitoring and debugging challenges in Python applications. Each example demonstrates how to instrument different components, showing how they work together within a distributed trace to provide end-to-end visibility.
 
+<Alert>
+  This page covers both transaction mode (default) and stream mode. See{" "}
+  <PlatformLink to="/tracing/new-spans/">New Spans</PlatformLink> to learn more.
+</Alert>
+
 ## File Upload and Processing Pipeline
 
 **Challenge:** Understanding bottlenecks and failures in multi-step file processing operations across request handling and processing services.
@@ -19,7 +24,8 @@ This guide provides practical examples of using span attributes and metrics to s
 **Solution:** Track the entire file processing pipeline with detailed metrics at each stage, from initial file handling through processing and storage.
 
 **Client Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # File upload request handling
 import sentry_sdk
 import time
@@ -27,63 +33,134 @@ from pathlib import Path
 import requests
 from contextlib import contextmanager
 
-with sentry_sdk.start_span(op="upload", name="Upload File") as span:
-    try:
-        # Begin upload process with the requests library
-        file_path = Path("/path/to/uploads/user-profile.jpg")
-        
-        # Track progress with a context manager
-        @contextmanager
-        def track_progress(total_size):
-            start = time.time()
-            bytes_sent = 0
-            
-            class ProgressTracker:
-                def __call__(self, monitor):
-                    nonlocal bytes_sent
-                    bytes_sent = monitor.bytes_read
-                    progress_percent = (bytes_sent / total_size) * 100
-                    span.set_data("upload.percent_complete", progress_percent)
-                    span.set_data("upload.bytes_transferred", bytes_sent)
-            
-            yield ProgressTracker()
-            
-            # Record total time after context exits
-            span.set_data("upload.total_time_ms", (time.time() - start) * 1000)
-        
-        # Use the progress tracker with requests
-        with track_progress(file_path.stat().st_size) as progress_callback:
-            with open(file_path, "rb") as f:
-                response = requests.post(
-                    "https://api.example.com/upload",
-                    files={"file": f},
-                    headers={"X-Sentry-Trace": sentry_sdk.get_traceparent()},  # Propagate trace context
-                    stream=True,
-                    hooks={"response": progress_callback}
-                )
-            
-            # Set final data after completion
-            span.set_data("upload.success", response.ok)
-            if response.ok:
-                result = response.json()
-                span.set_data("upload.server_file_id", result["file_id"])
-            
-            return response
-        
-    except Exception as error:
-        # Record failure information
-        span.set_data("upload.success", False)
-        span.set_data("upload.error_type", error.__class__.__name__)
-        span.set_data("upload.error_message", str(error))
-        span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
-        raise
+def upload_file():
+    with sentry_sdk.start_span(op="upload", name="Upload File") as span:
+        try:
+            # Begin upload process with the requests library
+            file_path = Path("/path/to/uploads/user-profile.jpg")
+
+            # Track progress with a context manager
+            @contextmanager
+            def track_progress(total_size):
+                start = time.time()
+                bytes_sent = 0
+
+                class ProgressTracker:
+                    def __call__(self, monitor):
+                        nonlocal bytes_sent
+                        bytes_sent = monitor.bytes_read
+                        progress_percent = (bytes_sent / total_size) * 100
+                        span.set_data("upload.percent_complete", progress_percent)
+                        span.set_data("upload.bytes_transferred", bytes_sent)
+
+                yield ProgressTracker()
+
+                # Record total time after context exits
+                span.set_data("upload.total_time_ms", (time.time() - start) * 1000)
+
+            # Use the progress tracker with requests
+            with track_progress(file_path.stat().st_size) as progress_callback:
+                with open(file_path, "rb") as f:
+                    response = requests.post(
+                        "https://api.example.com/upload",
+                        files={"file": f},
+                        headers={"X-Sentry-Trace": sentry_sdk.get_traceparent()},  # Propagate trace context
+                        stream=True,
+                        hooks={"response": progress_callback}
+                    )
+
+                # Set final data after completion
+                span.set_data("upload.success", response.ok)
+                if response.ok:
+                    result = response.json()
+                    span.set_data("upload.server_file_id", result["file_id"])
+
+                return response
+
+        except Exception as error:
+            # Record failure information
+            span.set_data("upload.success", False)
+            span.set_data("upload.error_type", error.__class__.__name__)
+            span.set_data("upload.error_message", str(error))
+            span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
+            raise
+
+upload_file()
+```
+
+```python {tabTitle:Stream Mode}
+# File upload request handling
+import sentry_sdk
+import time
+from pathlib import Path
+import requests
+from contextlib import contextmanager
+
+def upload_file():
+    with sentry_sdk.traces.start_span(
+        name="Upload File",
+        attributes={"sentry.op": "upload"},
+    ) as span:
+        try:
+            # Begin upload process with the requests library
+            file_path = Path("/path/to/uploads/user-profile.jpg")
+
+            # Track progress with a context manager
+            @contextmanager
+            def track_progress(total_size):
+                start = time.time()
+                bytes_sent = 0
+
+                class ProgressTracker:
+                    def __call__(self, monitor):
+                        nonlocal bytes_sent
+                        bytes_sent = monitor.bytes_read
+                        progress_percent = (bytes_sent / total_size) * 100
+                        span.set_attribute("upload.percent_complete", progress_percent)
+                        span.set_attribute("upload.bytes_transferred", bytes_sent)
+
+                yield ProgressTracker()
+
+                # Record total time after context exits
+                span.set_attribute("upload.total_time_ms", (time.time() - start) * 1000)
+
+            # Use the progress tracker with requests
+            with track_progress(file_path.stat().st_size) as progress_callback:
+                with open(file_path, "rb") as f:
+                    response = requests.post(
+                        "https://api.example.com/upload",
+                        files={"file": f},
+                        headers={"X-Sentry-Trace": sentry_sdk.get_traceparent()},  # Propagate trace context
+                        stream=True,
+                        hooks={"response": progress_callback}
+                    )
+
+                # Set final data after completion
+                span.set_attribute("upload.success", response.ok)
+                if response.ok:
+                    result = response.json()
+                    span.set_attribute("upload.server_file_id", result["file_id"])
+
+                return response
+
+        except Exception as error:
+            # Record failure information
+            span.set_attribute("upload.success", False)
+            span.set_attribute("upload.error_type", error.__class__.__name__)
+            span.set_attribute("upload.error_message", str(error))
+            span.status = "error"
+            raise
+
+upload_file()
 ```
 
 **Server Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # File processing service
 import sentry_sdk
 from pathlib import Path
+import time
 import boto3
 
 with sentry_sdk.start_span(
@@ -92,7 +169,7 @@ with sentry_sdk.start_span(
 ) as span:
     # File processing implementation
     file_path = Path("/tmp/uploads/user-profile.jpg")
-    
+
     # Process the file
     try:
         # Track individual processing steps
@@ -100,7 +177,7 @@ with sentry_sdk.start_span(
             # Virus scan implementation
             scan_span.set_data("scan.engine", "clamav")
             scan_span.set_data("scan.result", "clean")
-        
+
         # Upload to S3
         s3_client = boto3.client('s3', region_name='us-west-2')
         upload_start = time.time()
@@ -109,16 +186,59 @@ with sentry_sdk.start_span(
             'my-bucket',
             'uploads/user-profile.jpg'
         )
-        
-        span.set_data("storage.actual_upload_time_ms", 
+
+        span.set_data("storage.actual_upload_time_ms",
                          (time.time() - upload_start) * 1000)
-    
+
     except Exception as e:
         span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
         span.set_data("error.message", str(e))
         raise
 ```
 
+```python {tabTitle:Stream Mode}
+# File processing service
+import sentry_sdk
+from pathlib import Path
+import time
+import boto3
+
+with sentry_sdk.traces.start_span(
+    name="File Processing Service",
+    attributes={"sentry.op": "file.process.service"},
+) as span:
+    # File processing implementation
+    file_path = Path("/tmp/uploads/user-profile.jpg")
+
+    # Process the file
+    try:
+        # Track individual processing steps
+        with sentry_sdk.traces.start_span(
+            name="Virus Scan",
+            attributes={"sentry.op": "scan"},
+        ) as scan_span:
+            # Virus scan implementation
+            scan_span.set_attribute("scan.engine", "clamav")
+            scan_span.set_attribute("scan.result", "clean")
+
+        # Upload to S3
+        s3_client = boto3.client('s3', region_name='us-west-2')
+        upload_start = time.time()
+        s3_client.upload_file(
+            str(file_path),
+            'my-bucket',
+            'uploads/user-profile.jpg'
+        )
+
+        span.set_attribute("storage.actual_upload_time_ms",
+                         (time.time() - upload_start) * 1000)
+
+    except Exception as e:
+        span.status = "error"
+        span.set_attribute("error.message", str(e))
+        raise
+```
+
 **How the Trace Works Together:**
 The client application span initiates the trace and handles the file upload. It propagates the trace context to the server through the request headers. The server span continues the trace, processing the file and storing it. This creates a complete picture of the file's journey, allowing you to:
 
@@ -134,25 +254,26 @@ The client application span initiates the trace and handles the file upload. It
 **Solution:** Tracking of the entire LLM interaction flow, from initial request through response processing.
 
 **Client Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # LLM request handling in a Flask application
 import sentry_sdk
 import time
 import openai
-from flask import jsonify
+from flask import jsonify, request
 
 @app.route("/ask", methods=["POST"])
 def handle_llm_request():
     with sentry_sdk.start_span(op="gen_ai.chat", name="Generate Text") as span:
         start_time = time.time() * 1000  # Convert to milliseconds
-        
+
         # Begin streaming response from LLM API
         user_input = request.json["question"]
-        
+
         response_chunks = []
         first_token_received = False
         tokens_received = 0
-        
+
         # Using OpenAI's streaming API
         try:
             for chunk in openai.ChatCompletion.create(
@@ -163,25 +284,25 @@ def handle_llm_request():
                 tokens_received += 1
                 content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
                 response_chunks.append(content)
-                
+
                 # Record time to first token
                 if not first_token_received and content:
                     first_token_received = True
                     time_to_first_token = (time.time() * 1000) - start_time
                     span.set_data("response.time_to_first_token_ms", time_to_first_token)
-            
+
             # Record final metrics after stream completes
             total_request_time = (time.time() * 1000) - start_time
-            
+
             span.set_data("response.total_time_ms", total_request_time)
             span.set_data("response.format", "text")
             span.set_data("response.tokens_received", tokens_received)
-            
+
             return jsonify({
                 "response": "".join(response_chunks),
                 "tokens": tokens_received
             })
-            
+
         except Exception as error:
             span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
             span.set_data("error.type", error.__class__.__name__)
@@ -189,8 +310,69 @@ def handle_llm_request():
             return jsonify({"error": str(error)}), 500
 ```
 
+```python {tabTitle:Stream Mode}
+# LLM request handling in a Flask application
+import sentry_sdk
+import time
+import openai
+from flask import jsonify, request
+
+@app.route("/ask", methods=["POST"])
+def handle_llm_request():
+    with sentry_sdk.traces.start_span(
+        name="Generate Text",
+        attributes={"sentry.op": "gen_ai.chat"},
+    ) as span:
+        start_time = time.time() * 1000  # Convert to milliseconds
+
+        # Begin streaming response from LLM API
+        user_input = request.json["question"]
+
+        response_chunks = []
+        first_token_received = False
+        tokens_received = 0
+
+        # Using OpenAI's streaming API
+        try:
+            for chunk in openai.ChatCompletion.create(
+                model="gpt-4",
+                messages=[{"role": "user", "content": user_input}],
+                stream=True
+            ):
+                tokens_received += 1
+                content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
+                response_chunks.append(content)
+
+                # Record time to first token
+                if not first_token_received and content:
+                    first_token_received = True
+                    time_to_first_token = (time.time() * 1000) - start_time
+                    span.set_attribute("response.time_to_first_token_ms", time_to_first_token)
+
+            # Record final metrics after stream completes
+            total_request_time = (time.time() * 1000) - start_time
+
+            span.set_attributes({
+                "response.total_time_ms": total_request_time,
+                "response.format": "text",
+                "response.tokens_received": tokens_received
+            })
+
+            return jsonify({
+                "response": "".join(response_chunks),
+                "tokens": tokens_received
+            })
+
+        except Exception as error:
+            span.status = "error"
+            span.set_attribute("error.type", error.__class__.__name__)
+            span.set_attribute("error.message", str(error))
+            return jsonify({"error": str(error)}), 500
+```
+
 **Server Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # LLM processing service (e.g., in a separate microservice)
 import sentry_sdk
 import time
@@ -202,15 +384,15 @@ def process_llm_request(request_data):
         name="Generate Text"
     ) as span:
         start_time = int(time.time() * 1000)  # Current time in milliseconds
-        
+
         try:
             # Check rate limits before processing
             rate_limits = check_rate_limits()
             span.set_data("llm.rate_limit_remaining", rate_limits["remaining"])
-            
+
             # Prepare the prompt with additional context
             prepared_prompt = enhance_prompt(request_data["question"])
-            
+
             # Make the actual API call to the LLM provider
             response = openai.ChatCompletion.create(
                 model="gpt-4",
@@ -219,13 +401,13 @@ def process_llm_request(request_data):
                 temperature=0.7,
                 max_tokens=4096
             )
-            
+
             # Track token usage and performance metrics
             span.set_data("llm.prompt_tokens", response.usage.prompt_tokens)
             span.set_data("llm.completion_tokens", response.usage.completion_tokens)
             span.set_data("llm.total_tokens", response.usage.total_tokens)
             span.set_data("llm.api_latency_ms", int(time.time() * 1000) - start_time)
-            
+
             # Calculate and record cost based on token usage
             cost = calculate_cost(
                 response.usage.prompt_tokens,
@@ -233,23 +415,88 @@ def process_llm_request(request_data):
                 "gpt-4"
             )
             span.set_data("llm.cost_usd", cost)
-            
+
             return {
                 "response": response.choices[0].message.content,
                 "usage": response.usage
             }
-            
+
         except Exception as error:
             # Track error information
             span.set_data("error", True)
             span.set_data("error.type", error.__class__.__name__)
             span.set_data("error.message", str(error))
-            
+
             # Check if it's a rate limit error
             is_rate_limit = "rate_limit" in str(error).lower()
             span.set_data("error.is_rate_limit", is_rate_limit)
             span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
-            
+
+            raise
+```
+
+```python {tabTitle:Stream Mode}
+# LLM processing service (e.g., in a separate microservice)
+import sentry_sdk
+import time
+import openai
+
+def process_llm_request(request_data):
+    with sentry_sdk.traces.start_span(
+        name="Generate Text",
+        attributes={"sentry.op": "gen_ai.chat"},
+    ) as span:
+        start_time = int(time.time() * 1000)  # Current time in milliseconds
+
+        try:
+            # Check rate limits before processing
+            rate_limits = check_rate_limits()
+            span.set_attribute("llm.rate_limit_remaining", rate_limits["remaining"])
+
+            # Prepare the prompt with additional context
+            prepared_prompt = enhance_prompt(request_data["question"])
+
+            # Make the actual API call to the LLM provider
+            response = openai.ChatCompletion.create(
+                model="gpt-4",
+                messages=[{"role": "system", "content": "You are a helpful assistant."},
+                         {"role": "user", "content": prepared_prompt}],
+                temperature=0.7,
+                max_tokens=4096
+            )
+
+            # Track token usage and performance metrics
+            span.set_attributes({
+                "llm.prompt_tokens": response.usage.prompt_tokens,
+                "llm.completion_tokens": response.usage.completion_tokens,
+                "llm.total_tokens": response.usage.total_tokens,
+                "llm.api_latency_ms": int(time.time() * 1000) - start_time
+            })
+
+            # Calculate and record cost based on token usage
+            cost = calculate_cost(
+                response.usage.prompt_tokens,
+                response.usage.completion_tokens,
+                "gpt-4"
+            )
+            span.set_attribute("llm.cost_usd", cost)
+
+            return {
+                "response": response.choices[0].message.content,
+                "usage": response.usage
+            }
+
+        except Exception as error:
+            # Track error information
+            span.set_attribute("error", True)
+            span.set_attribute("error.type", error.__class__.__name__)
+            span.set_attribute("error.message", str(error))
+
+            # Check if it's a rate limit error
+            is_rate_limit = "rate_limit" in str(error).lower()
+            span.set_attribute("error.is_rate_limit", is_rate_limit)
+            span.status = "error"
+
             raise
 ```
 
@@ -269,7 +516,8 @@ The client application span captures the initial request handling, while the ser
 **Solution:** Track the full transaction process from API request to order fulfillment.
 
 **Client Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # Django view handling checkout request
 import sentry_sdk
 import time
@@ -282,28 +530,28 @@ class CheckoutView(View):
             # Validate the checkout request
             validation_start = time.time()
             validation_result = self.validate_checkout_data(request.POST)
-            span.set_data("request.validation_time_ms", 
+            span.set_data("request.validation_time_ms",
                              (time.time() - validation_start) * 1000)
-            
+
             if not validation_result["valid"]:
                 span.set_data("request.validation_success", False)
                 span.set_data("request.validation_errors", validation_result["errors"])
                 return JsonResponse({"errors": validation_result["errors"]}, status=400)
-            
+
             # Process the order
             try:
                 order_result = self.process_order(request)
-                
+
                 # Update span with order results
                 span.set_data("order.id", order_result["order_id"])
                 span.set_data("order.success", True)
-                
+
                 # Clear the cart and return success
                 request.session["cart"] = []
                 request.session["cart_total"] = 0
-                
+
                 return JsonResponse({"order_id": order_result["order_id"]})
-                
+
             except Exception as e:
                 span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
                 span.set_data("order.success", False)
@@ -311,11 +559,58 @@ class CheckoutView(View):
                 return JsonResponse({"error": str(e)}, status=500)
 ```
 
+```python {tabTitle:Stream Mode}
+# Django view handling checkout request
+import sentry_sdk
+import time
+from django.views import View
+from django.http import JsonResponse
+
+class CheckoutView(View):
+    def post(self, request):
+        with sentry_sdk.traces.start_span(
+            name="Process Order",
+            attributes={"sentry.op": "order"},
+        ) as span:
+            # Validate the checkout request
+            validation_start = time.time()
+            validation_result = self.validate_checkout_data(request.POST)
+            span.set_attribute("request.validation_time_ms",
+                             (time.time() - validation_start) * 1000)
+
+            if not validation_result["valid"]:
+                span.set_attribute("request.validation_success", False)
+                span.set_attribute("request.validation_errors", validation_result["errors"])
+                return JsonResponse({"errors": validation_result["errors"]}, status=400)
+
+            # Process the order
+            try:
+                order_result = self.process_order(request)
+
+                # Update span with order results
+                span.set_attribute("order.id", order_result["order_id"])
+                span.set_attribute("order.success", True)
+
+                # Clear the cart and return success
+                request.session["cart"] = []
+                request.session["cart_total"] = 0
+
+                return JsonResponse({"order_id": order_result["order_id"]})
+
+            except Exception as e:
+                span.status = "error"
+                span.set_attribute("order.success", False)
+                span.set_attribute("error.message", str(e))
+                return JsonResponse({"error": str(e)}, status=500)
+```
+
 **Server Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # Order processing service
 import sentry_sdk
 import stripe
+import time
 from decimal import Decimal
 
 def process_order(order_data):
@@ -327,15 +622,15 @@ def process_order(order_data):
             # Check inventory availability
             inventory_start = time.time()
             inventory_result = check_inventory(order_data["items"])
-            span.set_data("inventory.check_time_ms", 
+            span.set_data("inventory.check_time_ms",
                              (time.time() - inventory_start) * 1000)
             span.set_data("inventory.all_available", inventory_result["all_available"])
-            
+
             if not inventory_result["all_available"]:
-                span.set_data("inventory.unavailable_items", 
+                span.set_data("inventory.unavailable_items",
                                  inventory_result["unavailable_items"])
                 raise ValueError("Some items are out of stock")
-            
+
             # Process payment via Stripe
             payment_start = time.time()
             stripe.api_key = "sk_test_..."
@@ -345,27 +640,27 @@ def process_order(order_data):
                 payment_method=order_data["payment_method_id"],
                 confirm=True
             )
-            
-            span.set_data("payment.processing_time_ms", 
+
+            span.set_data("payment.processing_time_ms",
                              (time.time() - payment_start) * 1000)
             span.set_data("payment.transaction_id", payment_intent.id)
             span.set_data("payment.success", payment_intent.status == "succeeded")
-            
+
             # Create fulfillment record
             fulfillment = create_fulfillment(order_data["order_id"])
             span.set_data("fulfillment.id", fulfillment["id"])
             span.set_data("fulfillment.warehouse", fulfillment["warehouse"])
             span.set_data("fulfillment.shipping_method", fulfillment["shipping_method"])
-            span.set_data("fulfillment.estimated_delivery", 
+            span.set_data("fulfillment.estimated_delivery",
                              fulfillment["estimated_delivery"].isoformat())
-            
+
             return {
                 "success": True,
                 "order_id": order_data["order_id"],
                 "payment_id": payment_intent.id,
                 "fulfillment_id": fulfillment["id"]
             }
-            
+
         except Exception as e:
             span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
             span.set_data("error.message", str(e))
@@ -373,6 +668,70 @@ def process_order(order_data):
             raise
 ```
 
+```python {tabTitle:Stream Mode}
+# Order processing service
+import sentry_sdk
+import stripe
+import time
+from decimal import Decimal
+
+def process_order(order_data):
+    with sentry_sdk.traces.start_span(
+        name="Check Inventory",
+        attributes={"sentry.op": "inventory"},
+    ) as span:
+        try:
+            # Check inventory availability
+            inventory_start = time.time()
+            inventory_result = check_inventory(order_data["items"])
+            span.set_attribute("inventory.check_time_ms",
+                             (time.time() - inventory_start) * 1000)
+            span.set_attribute("inventory.all_available", inventory_result["all_available"])
+
+            if not inventory_result["all_available"]:
+                span.set_attribute("inventory.unavailable_items",
+                                 inventory_result["unavailable_items"])
+                raise ValueError("Some items are out of stock")
+
+            # Process payment via Stripe
+            payment_start = time.time()
+            stripe.api_key = "sk_test_..."
+            payment_intent = stripe.PaymentIntent.create(
+                amount=int(Decimal(order_data["total"]) * 100),  # Convert to cents
+                currency="usd",
+                payment_method=order_data["payment_method_id"],
+                confirm=True
+            )
+
+            span.set_attributes({
+                "payment.processing_time_ms": (time.time() - payment_start) * 1000,
+                "payment.transaction_id": payment_intent.id,
+                "payment.success": payment_intent.status == "succeeded"
+            })
+
+            # Create fulfillment record
+            fulfillment = create_fulfillment(order_data["order_id"])
+            span.set_attributes({
+                "fulfillment.id": fulfillment["id"],
+                "fulfillment.warehouse": fulfillment["warehouse"],
+                "fulfillment.shipping_method": fulfillment["shipping_method"],
+                "fulfillment.estimated_delivery": fulfillment["estimated_delivery"].isoformat()
+            })
+
+            return {
+                "success": True,
+                "order_id": order_data["order_id"],
+                "payment_id": payment_intent.id,
+                "fulfillment_id": fulfillment["id"]
+            }
+
+        except Exception as e:
+            span.status = "error"
+            span.set_attribute("error.message", str(e))
+            span.set_attribute("order.success", False)
+            raise
+```
+
 **How the Trace Works Together:**
 The client application span tracks the initial order request, while the server span handles order processing and fulfillment. The distributed trace provides visibility into the entire purchase flow, allowing you to:
 
@@ -389,11 +748,14 @@ The client application span tracks the initial order request, while the server s
 **Solution:** Comprehensive tracking of job lifecycle across queue management, processing stages, and worker performance.
 
 **Client Application Instrumentation:**
-```python
+
+```python {tabTitle:Transaction Mode (Default)}
 # Celery task submission
 import sentry_sdk
 from celery import shared_task
 import time
+import psutil
+import os
 from datetime import datetime, timedelta
 
 def submit_processing_job(data_file_path, priority="medium"):
@@ -406,20 +768,20 @@ def submit_processing_job(data_file_path, priority="medium"):
                 kwargs={"priority": priority},
                 queue="data_processing"
             )
-            
+
             span.set_data("job.id", task.id)
             span.set_data("job.submission_success", True)
-            
+
             # Start monitoring task progress
             monitoring_result = setup_task_monitoring(task.id)
             span.set_data("monitor.callback_url", monitoring_result["callback_url"])
-            
+
             return {
                 "job_id": task.id,
                 "status": "submitted",
                 "monitoring_url": monitoring_result["status_url"]
             }
-            
+
         except Exception as e:
             span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
             span.set_data("job.submission_success", False)
@@ -436,34 +798,34 @@ def process_data_file(file_path, priority="medium"):
             # Processing implementation with stage tracking
             span.set_data("processing.current_stage", "parse")
             data = parse_data_file(file_path)
-            
+
             span.set_data("processing.current_stage", "transform")
             transformed_data = transform_data(data)
-            
+
             span.set_data("processing.current_stage", "validate")
             validation_result = validate_data(transformed_data)
             span.set_data("validation.errors_count", len(validation_result["errors"]))
-            
+
             span.set_data("processing.current_stage", "export")
             export_result = export_processed_data(transformed_data)
-            
+
             # Record resource utilization
             span.set_data("resource.cpu_percent", psutil.cpu_percent())
-            span.set_data("resource.memory_used_mb", 
+            span.set_data("resource.memory_used_mb",
                              psutil.Process().memory_info().rss / (1024 * 1024))
-            
+
             # Update final job outcome
             span.set_data("outcome.status", "completed")
             span.set_data("outcome.records_processed", len(data))
-            span.set_data("outcome.output_size_bytes", 
+            span.set_data("outcome.output_size_bytes",
                              os.path.getsize(export_result["output_path"]))
-            
+
             return {
                 "success": True,
                 "records_processed": len(data),
                 "output_path": export_result["output_path"]
             }
-            
+
         except Exception as e:
             span.set_status(sentry_sdk.consts.SPANSTATUS.INTERNAL_ERROR)
             span.set_data("outcome.status", "failed")
@@ -472,6 +834,94 @@ def process_data_file(file_path, priority="medium"):
             raise
 ```
 
+```python {tabTitle:Stream Mode}
+# Celery task submission
+import sentry_sdk
+from celery import shared_task
+import time
+import psutil
+import os
+from datetime import datetime, timedelta
+
+def submit_processing_job(data_file_path, priority="medium"):
+    with sentry_sdk.traces.start_span(
+        name="Process Job",
+        attributes={"sentry.op": "job"},
+    ) as span:
+        # Submit job to Celery
+        try:
+            # Configure task and submit
+            task = process_data_file.apply_async(
+                args=[str(data_file_path)],
+                kwargs={"priority": priority},
+                queue="data_processing"
+            )
+
+            span.set_attribute("job.id", task.id)
+            span.set_attribute("job.submission_success", True)
+
+            # Start monitoring task progress
+            monitoring_result = setup_task_monitoring(task.id)
+            span.set_attribute("monitor.callback_url", monitoring_result["callback_url"])
+
+            return {
+                "job_id": task.id,
+                "status": "submitted",
+                "monitoring_url": monitoring_result["status_url"]
+            }
+
+        except Exception as e:
+            span.status = "error"
+            span.set_attribute("job.submission_success", False)
+            span.set_attribute("error.message", str(e))
+            raise
+
+@shared_task(name="tasks.process_data_file")
+def process_data_file(file_path, priority="medium"):
+    with sentry_sdk.traces.start_span(
+        name="Process Data",
+        attributes={"sentry.op": "process"},
+    ) as span:
+        try:
+            # Processing implementation with stage tracking
+            span.set_attribute("processing.current_stage", "parse")
+            data = parse_data_file(file_path)
+
+            span.set_attribute("processing.current_stage", "transform")
+            transformed_data = transform_data(data)
+
+            span.set_attribute("processing.current_stage", "validate")
+            validation_result = validate_data(transformed_data)
+            span.set_attribute("validation.errors_count", len(validation_result["errors"]))
+
+            span.set_attribute("processing.current_stage", "export")
+            export_result = export_processed_data(transformed_data)
+
+            # Record resource utilization
+            span.set_attribute("resource.cpu_percent", psutil.cpu_percent())
+            span.set_attribute("resource.memory_used_mb",
+                             psutil.Process().memory_info().rss / (1024 * 1024))
+
+            # Update final job outcome
+            span.set_attributes({
+                "outcome.status": "completed",
+                "outcome.records_processed": len(data),
+                "outcome.output_size_bytes": os.path.getsize(export_result["output_path"])
+            })
+
+            return {
+                "success": True,
+                "records_processed": len(data),
+                "output_path": export_result["output_path"]
+            }
+
+        except Exception as e:
+            span.status = "error"
+            span.set_attribute("outcome.status", "failed")
+            span.set_attribute("error.message", str(e))
+            raise
+```
+
 **How the Trace Works Together:**
 This example shows both the job submission and processing as a single trace, which is common in Celery/distributed task patterns. The spans track the entire job lifecycle, enabling you to:
 
diff --git a/docs/platforms/python/tracing/troubleshooting/index.mdx b/docs/platforms/python/tracing/troubleshooting/index.mdx
index b6c136095d57e..1e68ab304394f 100644
--- a/docs/platforms/python/tracing/troubleshooting/index.mdx
+++ b/docs/platforms/python/tracing/troubleshooting/index.mdx
@@ -91,3 +91,9 @@ with sentry_sdk.start_span(op="request", name="setup form") as span:
 
     # ...
 ```
+
+## Traces Miss Spans, High Memory Usage, or Data Loss After Crashes
+
+If you're hitting the 1,000-span limit, experiencing high memory usage from long-running processes, or losing span data when your process crashes, consider enabling stream mode. Stream mode sends spans to Sentry in batches as they finish rather than holding them in memory until the transaction ends.
+
+See <PlatformLink to="/tracing/new-spans/">New Spans</PlatformLink> for more information.
diff --git a/platform-includes/distributed-tracing/custom-instrumentation/python.mdx b/platform-includes/distributed-tracing/custom-instrumentation/python.mdx
index 8ac6dcfdea6e6..bf511e4998175 100644
--- a/platform-includes/distributed-tracing/custom-instrumentation/python.mdx
+++ b/platform-includes/distributed-tracing/custom-instrumentation/python.mdx
@@ -2,6 +2,12 @@ Distributed tracing works out of the box for [supported frameworks](/platforms/p
 
 This page describes how to manually propagate trace information into and out of your Python application. All you have to do is to make sure your application extracts incoming headers and to set those headers again when making an outgoing request within your application.
 
+<Alert>
+
+This page covers both transaction mode (default) and stream mode. See <PlatformLink to="/tracing/new-spans/">New Spans</PlatformLink> to learn more.
+
+</Alert>
+
 ## 1. Extract Incoming Tracing Information
 
 Incoming tracing information has to be extracted and stored in memory for later use. Sentry provides the `continue_trace()` function to help you with this. Incoming tracing information can come from different places:
@@ -10,9 +16,16 @@ Incoming tracing information has to be extracted and stored in memory for later
 - In a job queue, like Celery, it can be retrieved from meta or header variables.
 - You also can pick up tracing information from environment variables.
 
+<Alert>
+In **transaction mode**, `sentry_sdk.continue_trace()` returns a transaction, but does not start it. To start the transaction, use `start_transaction()`.
+
+In **stream mode**, `sentry_sdk.traces.continue_trace()` replaces `sentry_sdk.continue_trace()` and is not a context manager. Instead, it sets the propagation context, and the next span created with `sentry_sdk.traces.start_span()` picks it up automatically.
+
+</Alert>
+
 Here's an example of how to extract and store incoming tracing information using `continue_trace()`:
 
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from my_project import get_incoming_headers_as_dict
 
@@ -23,14 +36,43 @@ with sentry_sdk.start_transaction(transaction):
     ...
 ```
 
-In this example, `get_incoming_headers_as_dict()` returns a dictionary that contains tracing information from HTTP headers, environment variables, or any other mechanism your project uses to communicate with the outside world.
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from my_project import get_incoming_headers_as_dict
+
+headers = get_incoming_headers_as_dict()
 
-Sentry's `continue_trace()` function will extract the given headers, try to find the `sentry-trace` and `baggage` headers, and store them in memory for later use.
+sentry_sdk.traces.continue_trace(headers)
 
-`continue_trace()` returns a transaction, but does not start it. To start the transaction, use `start_transaction()`.
+with sentry_sdk.traces.start_span(name="handle request"):
+    ...
+```
 
+In these examples, `get_incoming_headers_as_dict()` returns a dictionary that contains tracing information from HTTP headers, environment variables, or any other mechanism your project uses to communicate with the outside world.
+
+## 2. Start a New Trace
+
+<Alert level="warning">
+  This step only applies to stream mode and is not available in transaction
+  mode.
+</Alert>
+
+In stream mode, if you need to start a completely new trace unconnected to the current one, use `sentry_sdk.traces.new_trace()`. This is useful for background jobs or scheduled tasks where you want a clean trace boundary:
+
+```python
+import sentry_sdk
+
+with sentry_sdk.traces.start_span(name="span in trace 1"):
+    ...
+
+sentry_sdk.traces.new_trace()
+
+with sentry_sdk.traces.start_span(name="span in trace 2"):
+    # This span is the root of a new, separate trace
+    ...
+```
 
-## 2. Inject Tracing Information to Outgoing Requests
+## 3. Inject Tracing Information to Outgoing Requests
 
 For distributed tracing to work, the two headers `sentry-trace` and `baggage`, must be added to outgoing requests. If you pregenerate HTML on the server-side, you might want to take a look at [Inject Tracing Information into Rendered HTML](#inject-tracing-information-into-rendered-html), which describes how to pass on tracing information through HTML meta tags.
 
diff --git a/platform-includes/performance/traces-sampler-as-sampler/python.mdx b/platform-includes/performance/traces-sampler-as-sampler/python.mdx
index a8c2a6f44cca2..40f49bb0d7fbe 100644
--- a/platform-includes/performance/traces-sampler-as-sampler/python.mdx
+++ b/platform-includes/performance/traces-sampler-as-sampler/python.mdx
@@ -1,4 +1,4 @@
-```python
+```python {tabTitle:Transaction Mode (Default)}
 import sentry_sdk
 from sentry_sdk.types import SamplingContext
 
@@ -30,3 +30,37 @@ sentry_sdk.init(
     traces_sampler=traces_sampler,
 )
 ```
+
+```python {tabTitle:Stream Mode}
+import sentry_sdk
+from sentry_sdk.types import SamplingContext
+
+def traces_sampler(sampling_context: SamplingContext) -> float:
+    # Use the parent sampling decision if we have an incoming trace.
+    # Note: we strongly recommend respecting the parent sampling decision,
+    # as this ensures your traces will be complete!
+    parent_sampling_decision = sampling_context["span_context"]["parent_sampled"]
+    if parent_sampling_decision is not None:
+        return float(parent_sampling_decision)
+
+    # Examine provided sampling context along with anything in the
+    # global namespace to compute the sample rate for this span
+    if "...":
+        # These are important - take a big sample
+        return 0.5
+    elif "...":
+        # These are less important - only take 1%
+        return 0.01
+    elif "...":
+        # These aren't worth tracking - drop these spans
+        return 0
+
+    # Default sample rate
+    return 0.1
+
+sentry_sdk.init(
+    # ...
+    traces_sampler=traces_sampler,
+    _experiments={"trace_lifecycle": "stream"},
+)
+```