diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py
index 6f3f11c2f3..38ecc8da51 100644
--- a/sentry_sdk/_span_batcher.py
+++ b/sentry_sdk/_span_batcher.py
@@ -70,7 +70,15 @@ def add(self, span: "StreamedSpan") -> None:
     @staticmethod
     def _to_transport_format(item: "StreamedSpan") -> "Any":
         # TODO[span-first]
-        res: "dict[str, Any]" = {}
+        res: "dict[str, Any]" = {
+            "name": item.name,
+        }
+
+        if item._attributes:
+            res["attributes"] = {
+                k: serialize_attribute(v) for (k, v) in item._attributes.items()
+            }
+
         return res
 
     def _flush(self) -> None:
diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py
index c0681fa0a6..d81415ad15 100644
--- a/sentry_sdk/client.py
+++ b/sentry_sdk/client.py
@@ -69,6 +69,7 @@
     from sentry_sdk.scope import Scope
     from sentry_sdk.session import Session
     from sentry_sdk.spotlight import SpotlightClient
+    from sentry_sdk.traces import StreamedSpan
     from sentry_sdk.transport import Transport, Item
     from sentry_sdk._log_batcher import LogBatcher
     from sentry_sdk._metrics_batcher import MetricsBatcher
@@ -227,6 +228,9 @@ def _capture_log(self, log: "Log", scope: "Scope") -> None:
     def _capture_metric(self, metric: "Metric", scope: "Scope") -> None:
         pass
 
+    def _capture_span(self, span: "StreamedSpan", scope: "Scope") -> None:
+        pass
+
     def capture_session(self, *args: "Any", **kwargs: "Any") -> None:
         return None
 
@@ -920,7 +924,7 @@ def capture_event(
 
     def _capture_telemetry(
         self,
-        telemetry: "Optional[Union[Log, Metric]]",
+        telemetry: "Optional[Union[Log, Metric, StreamedSpan]]",
         ty: str,
         scope: "Scope",
     ) -> None:
@@ -947,6 +951,8 @@ def _capture_telemetry(
             batcher = self.log_batcher
         elif ty == "metric":
             batcher = self.metrics_batcher  # type: ignore
+        elif ty == "span":
+            batcher = self.span_batcher  # type: ignore
 
         if batcher is not None:
             batcher.add(telemetry)  # type: ignore
@@ -957,6 +963,9 @@ def _capture_log(self, log: "Optional[Log]", scope: "Scope") -> None:
     def _capture_metric(self, metric: "Optional[Metric]", scope: "Scope") -> None:
         self._capture_telemetry(metric, "metric", scope)
 
+    def _capture_span(self, span: "Optional[StreamedSpan]", scope: "Scope") -> None:
+        self._capture_telemetry(span, "span", scope)
+
     def capture_session(
         self,
         session: "Session",
diff --git a/sentry_sdk/scope.py b/sentry_sdk/scope.py
index acd82a57ef..3bc51c1af0 100644
--- a/sentry_sdk/scope.py
+++ b/sentry_sdk/scope.py
@@ -29,9 +29,11 @@
 from sentry_sdk.tracing_utils import (
     Baggage,
     has_tracing_enabled,
+    has_span_streaming_enabled,
     normalize_incoming_data,
     PropagationContext,
 )
+from sentry_sdk.traces import StreamedSpan
 from sentry_sdk.tracing import (
     BAGGAGE_HEADER_NAME,
     SENTRY_TRACE_HEADER_NAME,
@@ -1278,6 +1280,17 @@ def _capture_metric(self, metric: "Optional[Metric]") -> None:
 
         client._capture_metric(metric, scope=merged_scope)
 
+    def _capture_span(self, span: "Optional[StreamedSpan]") -> None:
+        if span is None:
+            return
+
+        client = self.get_client()
+        if not has_span_streaming_enabled(client.options):
+            return
+
+        merged_scope = self._merge_scopes()
+        client._capture_span(span, scope=merged_scope)
+
     def capture_message(
         self,
         message: str,
@@ -1522,16 +1535,25 @@ def _apply_flags_to_event(
             )
 
     def _apply_scope_attributes_to_telemetry(
-        self, telemetry: "Union[Log, Metric]"
+        self, telemetry: "Union[Log, Metric, StreamedSpan]"
     ) -> None:
+        # TODO: turn Logs, Metrics into actual classes
+        if isinstance(telemetry, dict):
+            attributes = telemetry["attributes"]
+        else:
+            attributes = telemetry._attributes
+
         for attribute, value in self._attributes.items():
-            if attribute not in telemetry["attributes"]:
-                telemetry["attributes"][attribute] = value
+            if attribute not in attributes:
+                attributes[attribute] = value
 
     def _apply_user_attributes_to_telemetry(
-        self, telemetry: "Union[Log, Metric]"
+        self, telemetry: "Union[Log, Metric, StreamedSpan]"
     ) -> None:
-        attributes = telemetry["attributes"]
+        if isinstance(telemetry, dict):
+            attributes = telemetry["attributes"]
+        else:
+            attributes = telemetry._attributes
 
         if not should_send_default_pii() or self._user is None:
             return
@@ -1651,16 +1673,19 @@ def apply_to_event(
         return event
 
     @_disable_capture
-    def apply_to_telemetry(self, telemetry: "Union[Log, Metric]") -> None:
+    def apply_to_telemetry(self, telemetry: "Union[Log, Metric, StreamedSpan]") -> None:
         # Attributes-based events and telemetry go through here (logs, metrics,
         # spansV2)
-        trace_context = self.get_trace_context()
-        trace_id = trace_context.get("trace_id")
-        if telemetry.get("trace_id") is None:
-            telemetry["trace_id"] = trace_id or "00000000-0000-0000-0000-000000000000"
-        span_id = trace_context.get("span_id")
-        if telemetry.get("span_id") is None and span_id:
-            telemetry["span_id"] = span_id
+        if not isinstance(telemetry, StreamedSpan):
+            trace_context = self.get_trace_context()
+            trace_id = trace_context.get("trace_id")
+            if telemetry.get("trace_id") is None:
+                telemetry["trace_id"] = (
+                    trace_id or "00000000-0000-0000-0000-000000000000"
+                )
+            span_id = trace_context.get("span_id")
+            if telemetry.get("span_id") is None and span_id:
+                telemetry["span_id"] = span_id
 
         self._apply_scope_attributes_to_telemetry(telemetry)
         self._apply_user_attributes_to_telemetry(telemetry)
diff --git a/sentry_sdk/traces.py b/sentry_sdk/traces.py
index d2f6549e83..1c58e4a1cb 100644
--- a/sentry_sdk/traces.py
+++ b/sentry_sdk/traces.py
@@ -8,8 +8,11 @@
 import uuid
 from typing import TYPE_CHECKING
 
+from sentry_sdk.utils import format_attribute
+
 if TYPE_CHECKING:
     from typing import Optional
+    from sentry_sdk._types import Attributes, AttributeValue
 
 
 class StreamedSpan:
@@ -22,15 +25,40 @@ class StreamedSpan:
     span implementation lives in tracing.Span.
     """
 
-    __slots__ = ("_trace_id",)
+    __slots__ = (
+        "name",
+        "_attributes",
+        "_trace_id",
+    )
 
     def __init__(
         self,
         *,
+        name: str,
+        attributes: "Optional[Attributes]" = None,
         trace_id: "Optional[str]" = None,
     ):
+        self.name: str = name
+        self._attributes: "Attributes" = attributes or {}
+
         self._trace_id = trace_id
 
+    def get_attributes(self) -> "Attributes":
+        return self._attributes
+
+    def set_attribute(self, key: str, value: "AttributeValue") -> None:
+        self._attributes[key] = format_attribute(value)
+
+    def set_attributes(self, attributes: "Attributes") -> None:
+        for key, value in attributes.items():
+            self.set_attribute(key, value)
+
+    def remove_attribute(self, key: str) -> None:
+        try:
+            del self._attributes[key]
+        except KeyError:
+            pass
+
     @property
     def trace_id(self) -> str:
         if not self._trace_id: