diff --git a/burr/integrations/langfuse.py b/burr/integrations/langfuse.py
new file mode 100644
index 000000000..20e506f68
--- /dev/null
+++ b/burr/integrations/langfuse.py
@@ -0,0 +1,414 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import json
+import logging
+from collections import abc
+from contextvars import ContextVar
+from typing import Any, Dict, List, Optional, Tuple
+
+from burr.integrations.base import require_plugin
+
+logger = logging.getLogger(__name__)
+
+try:
+    from langfuse import Langfuse
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import ReadableSpan
+    from opentelemetry.trace import get_current_span
+except ImportError as e:
+    require_plugin(
+        e,
+        "langfuse",
+    )
+
+try:
+    # propagate_attributes copies session/user attributes onto all child spans, including spans from third-party OTel instrumentation.
+    # available on langfuse>=3.9; on older versions we fall back to setting the attributes on Burr's own spans only (see _set_trace_attributes).
+    from langfuse import propagate_attributes
+except ImportError:
+    propagate_attributes = None
+
+try:
+    # Available on langfuse>=4 -- v4 only exports LLM-relevant spans by default,
+    # so we extend the default filter to also export Burr spans.
+    # On langfuse v3 all spans on the global tracer provider are exported,
+    # so no filter is necessary.
+    from langfuse import is_default_export_span
+except ImportError:
+    is_default_export_span = None
+
+from burr.core import Action, State, serde
+from burr.integrations.opentelemetry import OpenTelemetryBridge
+from burr.lifecycle.base import ExecuteMethod
+from burr.visibility import ActionSpan
+
+BURR_TRACER_NAME = "burr.integrations.langfuse"
+
+# Langfuse-recognized OpenTelemetry span attributes
+# https://langfuse.com/integrations/native/opentelemetry (attribute mapping)
+_LANGFUSE_OBSERVATION_INPUT = "langfuse.observation.input"
+_LANGFUSE_OBSERVATION_OUTPUT = "langfuse.observation.output"
+_LANGFUSE_OBSERVATION_METADATA_PREFIX = "langfuse.observation.metadata."
+_LANGFUSE_SESSION_ID = "langfuse.session.id"
+_LANGFUSE_USER_ID = "langfuse.user.id"
+# Langfuse v3 uses the OpenTelemetry semantic convention attribute names.
+_LANGFUSE_V3_SESSION_ID = "session.id"
+_LANGFUSE_V3_USER_ID = "user.id"
+
+# Attribute namespaces we pass through unprefixed so users can deliberately log Langfuse-mapped or GenAI semantic convention attributes.
+_PASSTHROUGH_ATTRIBUTE_PREFIXES = ("langfuse.", "gen_ai.")
+
+
+def burr_span_export_filter(span: "ReadableSpan") -> bool:
+    """
+    You only need this if you construct the :py:class:`Langfuse <langfuse.Langfuse>`
+    client yourself -- :py:class:`LangfuseBridge` applies it automatically when it
+    creates the client for you:
+
+    .. code-block:: python
+
+        from langfuse import Langfuse
+        from burr.integrations.langfuse import LangfuseBridge, burr_span_export_filter
+
+        client = Langfuse(should_export_span=burr_span_export_filter)
+        app = ApplicationBuilder().with_hooks(LangfuseBridge(langfuse_client=client))...
+    """
+    scope = span.instrumentation_scope
+    if scope is not None and (
+        scope.name == BURR_TRACER_NAME or scope.name.startswith(BURR_TRACER_NAME + ".")
+    ):
+        return True
+    if is_default_export_span is not None:
+        return is_default_export_span(span)
+    return True
+
+
+def _serialize_for_langfuse(value: Any) -> str:
+    # Serializes a value to a JSON string for the ``langfuse.observation.input``/ ``.output`` attributes, falling back to ``str`` on failure."""
+    try:
+        return json.dumps(serde.serialize(value))
+    except Exception as e:
+        logger.warning(f"Failed to serialize value for Langfuse: {e}")
+        return str(value)
+
+
+_OTEL_ATTRIBUTE_PRIMITIVES = (str, bool, int, float)
+
+
+def _convert_attribute(value: Any) -> Any:
+    """
+    Converts a logged attribute value to a valid OpenTelemetry attribute value.
+    OTel attributes only accept primitives and flat sequences of primitives, so
+    anything else (dicts, lists of dicts, ...) is JSON-serialized -- Langfuse
+    renders JSON strings in metadata.
+    """
+    if isinstance(value, _OTEL_ATTRIBUTE_PRIMITIVES):
+        return value
+    if isinstance(value, abc.Sequence):
+        primitive_types = {type(item) for item in value}
+        if len(primitive_types) <= 1 and all(
+            isinstance(item, _OTEL_ATTRIBUTE_PRIMITIVES) for item in value
+        ):
+            return list(value)
+    return _serialize_for_langfuse(value)
+
+
+class LangfuseBridge(OpenTelemetryBridge):
+    """
+    Adapter to log Burr application execution to `Langfuse <https://langfuse.com>`_.
+
+    1. Each application execution call (``run``/``step``/``iterate``/``stream_result``/...)
+       opens a root span, which defines the Langfuse trace
+    2. Each step opens a span, capturing the action's inputs/read state as the
+       observation input and its result/written state as the observation output
+    3. Each span opened through Burr's tracing API (``__tracer``) opens a span
+    4. Attributes logged through ``__tracer.log_attribute(s)`` are captured as
+       observation metadata
+
+    Burr's ``app_id`` maps to the Langfuse session (so multiple execution calls of the
+    same application group together), and the ``partition_key`` maps to the Langfuse
+    user - both can be overridden.
+
+    Basic usage - credentials are read from the standard ``LANGFUSE_PUBLIC_KEY``,
+    ``LANGFUSE_SECRET_KEY``, and ``LANGFUSE_HOST`` environment variables:
+
+    .. code-block:: python
+
+        from burr.integrations.langfuse import LangfuseBridge
+
+        app = (
+            ApplicationBuilder()
+            .with_graph(graph)
+            .with_entrypoint("prompt")
+            .with_hooks(LangfuseBridge())
+            .build()
+        )
+        app.run(halt_after=["response"])  # logs one trace to Langfuse
+
+    You can also pass credentials explicitly, or pass a pre-constructed client:
+
+    .. code-block:: python
+
+        LangfuseBridge(public_key="pk-lf-...", secret_key="sk-lf-...", host="...")
+        # or, equivalently
+        LangfuseBridge(langfuse_client=my_langfuse_client)
+
+    The underlying client is available as ``bridge.langfuse_client`` -- for example to
+    ``flush()`` in short-lived scripts, or to score traces.
+
+    .. note::
+
+        With langfuse v4+, spans are filtered before export, and only LLM-relevant
+        spans are exported by default. If you construct the ``Langfuse`` client
+        yourself, pass ``should_export_span=burr_span_export_filter`` to its
+        constructor so Burr spans are exported (see
+        :py:func:`burr_span_export_filter`). When ``LangfuseBridge`` constructs
+        the client for you, this is applied automatically.
+    """
+
+    def __init__(
+        self,
+        langfuse_client: Optional["Langfuse"] = None,
+        *,
+        session_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        capture_state: bool = True,
+        tracer: Optional["trace.Tracer"] = None,
+        tracer_provider: Optional["trace.TracerProvider"] = None,
+        **langfuse_kwargs: Any,
+    ):
+        """
+        Initializes the Langfuse bridge.
+
+        :param langfuse_client: A pre-constructed Langfuse client to use. If not passed,
+            one is created from ``langfuse_kwargs`` (falling back to the standard
+            ``LANGFUSE_*`` environment variables), with the Burr span export filter applied.
+        :param session_id: Langfuse session ID to group traces under. Defaults to the
+            Burr ``app_id``.
+        :param user_id: Langfuse user ID to attach to traces. Defaults to the Burr
+            ``partition_key`` (if set).
+        :param capture_state: Whether to capture state/inputs/results as observation
+            input/output. Set to False if your state contains data you do not want
+            sent to Langfuse.
+        :param tracer: OpenTelemetry tracer to use -- for testing/advanced use. Defaults
+            to a tracer named ``burr.integrations.langfuse`` from the global provider.
+        :param tracer_provider: OpenTelemetry tracer provider to use for both the
+            Langfuse client and Burr spans. When passing a pre-constructed client that
+            uses a custom provider, pass the same provider here.
+        :param langfuse_kwargs: Keyword arguments forwarded to the
+            :py:class:`Langfuse <langfuse.Langfuse>` constructor (e.g. ``public_key``,
+            ``secret_key``, ``host``). Only valid if ``langfuse_client`` is not passed.
+        """
+        if tracer is not None and tracer_provider is not None:
+            raise ValueError("Only pass one of tracer or tracer_provider, not both.")
+        if langfuse_client is not None and langfuse_kwargs:
+            raise ValueError(
+                f"Only pass one of langfuse_client or langfuse constructor kwargs, not both. "
+                f"Got: langfuse_client={langfuse_client} and kwargs={list(langfuse_kwargs)}"
+            )
+        if langfuse_client is None:
+            if is_default_export_span is not None:
+                langfuse_kwargs.setdefault("should_export_span", burr_span_export_filter)
+            if tracer_provider is not None:
+                langfuse_kwargs["tracer_provider"] = tracer_provider
+            langfuse_client = Langfuse(**langfuse_kwargs)
+        self.langfuse_client = langfuse_client
+        self.session_id = session_id
+        self.user_id = user_id
+        self.capture_state = capture_state
+        self._propagation_context_stack: ContextVar[Optional[List[Any]]] = ContextVar(
+            f"langfuse_propagation_context_stack_{id(self)}", default=None
+        )
+        # Note: the Langfuse client registers its span processor on the global
+        # tracer provider on construction, so we grab the tracer afterwards.
+        if tracer is None:
+            tracer = (
+                tracer_provider.get_tracer(BURR_TRACER_NAME)
+                if tracer_provider is not None
+                else trace.get_tracer(BURR_TRACER_NAME)
+            )
+        super().__init__(tracer=tracer)
+
+    def _trace_attributes(
+        self, app_id: str, partition_key: Optional[str]
+    ) -> Tuple[Optional[str], Optional[str]]:
+        session_id = self.session_id if self.session_id is not None else app_id
+        user_id = self.user_id if self.user_id is not None else partition_key
+        return session_id, user_id
+
+    def _set_trace_attributes(self, span: "trace.Span", app_id: str, partition_key: Optional[str]):
+        # sets Langfuse trace-level attributes (session/user) on a span
+        session_id, user_id = self._trace_attributes(app_id, partition_key)
+        if session_id is not None:
+            span.set_attribute(_LANGFUSE_SESSION_ID, session_id)
+            span.set_attribute(_LANGFUSE_V3_SESSION_ID, session_id)
+        if user_id is not None:
+            span.set_attribute(_LANGFUSE_USER_ID, user_id)
+            span.set_attribute(_LANGFUSE_V3_USER_ID, user_id)
+
+    def _enter_trace_attribute_context(self, app_id: str, partition_key: Optional[str]) -> None:
+        if propagate_attributes is None:  # langfuse < 3.9
+            return
+        session_id, user_id = self._trace_attributes(app_id, partition_key)
+        propagation_context = propagate_attributes(session_id=session_id, user_id=user_id)
+        propagation_context.__enter__()
+        stack = (self._propagation_context_stack.get() or [])[:]
+        stack.append(propagation_context)
+        self._propagation_context_stack.set(stack)
+
+    def _exit_trace_attribute_context(self) -> None:
+        if propagate_attributes is None:  # nothing was entered
+            return
+        stack = (self._propagation_context_stack.get() or [])[:]
+        if not stack:
+            logger.warning("No Langfuse trace attribute context to exit")
+            return
+        propagation_context = stack.pop()
+        self._propagation_context_stack.set(stack)
+        propagation_context.__exit__(None, None, None)
+
+    def pre_run_execute_call(
+        self,
+        *,
+        app_id: str,
+        partition_key: str,
+        state: "State",
+        method: ExecuteMethod,
+        **future_kwargs: Any,
+    ):
+        super().pre_run_execute_call(method=method, **future_kwargs)
+        span = get_current_span()
+        # trace-level attributes must be present on the root span
+        self._set_trace_attributes(span, app_id, partition_key)
+        span.set_attribute(_LANGFUSE_OBSERVATION_METADATA_PREFIX + "burr.app_id", app_id)
+        if partition_key is not None:
+            span.set_attribute(
+                _LANGFUSE_OBSERVATION_METADATA_PREFIX + "burr.partition_key", partition_key
+            )
+        if self.capture_state:
+            span.set_attribute(
+                _LANGFUSE_OBSERVATION_INPUT, _serialize_for_langfuse(state.get_all())
+            )
+        # Langfuse's propagation context copies session/user attributes to all child
+        # spans, including spans emitted by third-party OpenTelemetry instrumentation.
+        self._enter_trace_attribute_context(app_id, partition_key)
+
+    def post_run_execute_call(
+        self,
+        *,
+        state: "State",
+        exception: Optional[Exception],
+        **future_kwargs: Any,
+    ):
+        try:
+            if self.capture_state and exception is None:
+                span = get_current_span()
+                span.set_attribute(
+                    _LANGFUSE_OBSERVATION_OUTPUT, _serialize_for_langfuse(state.get_all())
+                )
+        finally:
+            try:
+                self._exit_trace_attribute_context()
+            finally:
+                super().post_run_execute_call(exception=exception, **future_kwargs)
+
+    def pre_run_step(
+        self,
+        *,
+        app_id: str,
+        partition_key: str,
+        sequence_id: int,
+        state: "State",
+        action: "Action",
+        inputs: Dict[str, Any],
+        **future_kwargs: Any,
+    ):
+        super().pre_run_step(action=action, **future_kwargs)
+        span = get_current_span()
+        # set session/user on every span so Langfuse session/user aggregations cover all observations, not just the trace root:
+        # https://langfuse.com/integrations/native/opentelemetry
+        self._set_trace_attributes(span, app_id, partition_key)
+        span.set_attribute(_LANGFUSE_OBSERVATION_METADATA_PREFIX + "burr.sequence_id", sequence_id)
+        if self.capture_state:
+            span.set_attribute(
+                _LANGFUSE_OBSERVATION_INPUT,
+                _serialize_for_langfuse(
+                    {
+                        "inputs": {
+                            key: value for key, value in inputs.items() if not key.startswith("__")
+                        },
+                        "state": {key: state.get(key) for key in action.reads},
+                    }
+                ),
+            )
+
+    def post_run_step(
+        self,
+        *,
+        state: "State",
+        action: "Action",
+        result: Optional[Dict[str, Any]],
+        exception: Exception,
+        **future_kwargs: Any,
+    ):
+        if self.capture_state and exception is None:
+            span = get_current_span()
+            span.set_attribute(
+                _LANGFUSE_OBSERVATION_OUTPUT,
+                _serialize_for_langfuse(
+                    {
+                        "result": result,
+                        "state": {key: state.get(key) for key in action.writes},
+                    }
+                ),
+            )
+        super().post_run_step(exception=exception, **future_kwargs)
+
+    def pre_start_span(
+        self,
+        *,
+        span: "ActionSpan",
+        app_id: str,
+        partition_key: Optional[str],
+        **future_kwargs: Any,
+    ):
+        super().pre_start_span(span=span, **future_kwargs)
+        self._set_trace_attributes(get_current_span(), app_id, partition_key)
+
+    def do_log_attributes(
+        self,
+        *,
+        attributes: Dict[str, Any],
+        **future_kwargs: Any,
+    ):
+        otel_span = get_current_span()
+        if otel_span is None:
+            logger.warning(
+                "Attempted to log attributes from the tracker outside of a span, ignoring"
+            )
+            return
+        otel_span.set_attributes(
+            {
+                (
+                    key
+                    if key.startswith(_PASSTHROUGH_ATTRIBUTE_PREFIXES)
+                    else _LANGFUSE_OBSERVATION_METADATA_PREFIX + key
+                ): _convert_attribute(value)
+                for key, value in attributes.items()
+            }
+        )
diff --git a/docs/reference/integrations/index.rst b/docs/reference/integrations/index.rst
index 562a8cd1b..e5b288363 100644
--- a/docs/reference/integrations/index.rst
+++ b/docs/reference/integrations/index.rst
@@ -30,6 +30,7 @@ Integrations -- we will be adding more
    streamlit
    opentelemetry
    traceloop
+   langfuse
    langchain
    pydantic
    haystack
diff --git a/docs/reference/integrations/langfuse.rst b/docs/reference/integrations/langfuse.rst
new file mode 100644
index 000000000..fed6e2902
--- /dev/null
+++ b/docs/reference/integrations/langfuse.rst
@@ -0,0 +1,73 @@
+..
+   Licensed to the Apache Software Foundation (ASF) under one
+   or more contributor license agreements.  See the NOTICE file
+   distributed with this work for additional information
+   regarding copyright ownership.  The ASF licenses this file
+   to you under the Apache License, Version 2.0 (the
+   "License"); you may not use this file except in compliance
+   with the License.  You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing,
+   software distributed under the License is distributed on an
+   "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+   KIND, either express or implied.  See the License for the
+   specific language governing permissions and limitations
+   under the License.
+
+
+.. _langfuseintegrationref:
+
+--------
+Langfuse
+--------
+
+`Langfuse <https://langfuse.com>`_ is an open-source LLM engineering platform with
+tracing/observability capabilities. Burr integrates with it through the
+OpenTelemetry-native Langfuse Python SDK, building on the
+:ref:`opentelemetry integration <opentelintegrationref>`.
+
+Install the integration:
+
+.. code-block:: bash
+
+    pip install "apache-burr[langfuse]"
+
+Then add the bridge as a hook -- credentials are read from the standard
+``LANGFUSE_PUBLIC_KEY``, ``LANGFUSE_SECRET_KEY``, and ``LANGFUSE_HOST``
+environment variables:
+
+.. code-block:: python
+
+    from burr.core import ApplicationBuilder
+    from burr.integrations.langfuse import LangfuseBridge
+
+    app = (
+        ApplicationBuilder()
+        .with_graph(graph)
+        .with_entrypoint("prompt")
+        .with_hooks(LangfuseBridge())
+        .build()
+    )
+    app.run(halt_after=["response"])  # logs one trace to Langfuse
+
+Each application execution call becomes a Langfuse trace, each step becomes a span
+(with state/inputs/results captured as observation input/output), and spans opened
+through Burr's :ref:`tracing API <opentelref>` become nested spans. Any additional
+OpenTelemetry LLM instrumentation (e.g. ``opentelemetry-instrumentation-openai``)
+appears nested within the corresponding Burr step.
+
+See the following resources for more information:
+
+- `Example in the repository <https://github.com/apache/burr/tree/main/examples/integrations/langfuse>`_
+- `Langfuse OpenTelemetry docs <https://langfuse.com/integrations/native/opentelemetry>`_
+
+Reference for the various useful methods:
+
+.. autoclass:: burr.integrations.langfuse.LangfuseBridge
+    :members:
+
+    .. automethod:: __init__
+
+.. autofunction:: burr.integrations.langfuse.burr_span_export_filter
diff --git a/examples/integrations/langfuse/README.md b/examples/integrations/langfuse/README.md
new file mode 100644
index 000000000..d8dec2988
--- /dev/null
+++ b/examples/integrations/langfuse/README.md
@@ -0,0 +1,72 @@
+<!--
+     Licensed to the Apache Software Foundation (ASF) under one
+     or more contributor license agreements.  See the NOTICE file
+     distributed with this work for additional information
+     regarding copyright ownership.  The ASF licenses this file
+     to you under the Apache License, Version 2.0 (the
+     "License"); you may not use this file except in compliance
+     with the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing,
+     software distributed under the License is distributed on an
+     "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+     KIND, either express or implied.  See the License for the
+     specific language governing permissions and limitations
+     under the License.
+-->
+
+# Langfuse + Burr
+
+This shows how to trace a Burr application to [Langfuse](https://langfuse.com)
+using the `LangfuseBridge` hook.
+
+What gets logged:
+
+1. One Langfuse **trace** per application execution call (`run`/`step`/`iterate`/`stream_result`/...)
+2. One **span** per step, with the action's inputs/read state as observation input and
+   its result/written state as observation output
+3. One **span** per span opened through Burr's tracing API (`__tracer`)
+4. Attributes logged via `__tracer.log_attribute(s)` as observation **metadata**
+5. Burr's `app_id` maps to the Langfuse **session** and `partition_key` to the
+   Langfuse **user** (both overridable)
+
+If you also install an OpenTelemetry LLM instrumentor (e.g.
+`opentelemetry-instrumentation-openai`), LLM calls show up as generations nested
+inside the corresponding Burr steps, with prompts/completions/token usage.
+
+## Running
+
+```bash
+pip install "apache-burr[langfuse]" openai opentelemetry-instrumentation-openai
+
+export LANGFUSE_PUBLIC_KEY="pk-lf-..."
+export LANGFUSE_SECRET_KEY="sk-lf-..."
+export LANGFUSE_HOST="https://cloud.langfuse.com"  # or your self-hosted URL
+export OPENAI_API_KEY="sk-..."
+
+python application.py
+```
+
+Then open your Langfuse project -- you will see one trace per `.run()` call, with
+the step spans, tracer spans, and (if instrumented) OpenAI generations nested inside.
+
+See [application.py](./application.py) for the full code, and the
+[integration docs](https://burr.apache.org/reference/integrations/langfuse/) for
+configuration options (custom session/user IDs, disabling state capture, passing
+your own `Langfuse` client).
+
+## Note on langfuse v4+
+
+Langfuse SDK v4+ only exports LLM-relevant spans by default. `LangfuseBridge`
+handles this automatically when it constructs the client. If you construct the
+`Langfuse` client yourself, pass the provided filter:
+
+```python
+from langfuse import Langfuse
+from burr.integrations.langfuse import LangfuseBridge, burr_span_export_filter
+
+client = Langfuse(should_export_span=burr_span_export_filter)
+bridge = LangfuseBridge(langfuse_client=client)
+```
diff --git a/examples/integrations/langfuse/__init__.py b/examples/integrations/langfuse/__init__.py
new file mode 100644
index 000000000..13a83393a
--- /dev/null
+++ b/examples/integrations/langfuse/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/examples/integrations/langfuse/application.py b/examples/integrations/langfuse/application.py
new file mode 100644
index 000000000..9c5ab7023
--- /dev/null
+++ b/examples/integrations/langfuse/application.py
@@ -0,0 +1,110 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# This is a simple chatbot to trace Langfuse
+
+"""
+Requires the following environment variables (see https://langfuse.com):
+
+- LANGFUSE_PUBLIC_KEY
+- LANGFUSE_SECRET_KEY
+- LANGFUSE_HOST (e.g. https://cloud.langfuse.com)
+- OPENAI_API_KEY
+"""
+
+from typing import Optional, Tuple
+
+import openai
+
+from burr.core import Application, ApplicationBuilder, State
+from burr.core.action import action
+from burr.integrations.langfuse import LangfuseBridge
+from burr.visibility import TracerFactory, trace
+
+try:
+    # Optional: with the openai instrumentor installed, LLM calls (prompts, completions, token usage) show up as generations nested inside Burr steps.
+    # pip install opentelemetry-instrumentation-openai
+    from opentelemetry.instrumentation.openai import OpenAIInstrumentor
+
+    OpenAIInstrumentor().instrument()
+except ImportError:
+    pass
+
+
+@action(reads=[], writes=["chat_history", "prompt"])
+def process_prompt(state: State, prompt: str) -> Tuple[dict, State]:
+    result = {"chat_item": {"role": "user", "content": prompt}}
+    return result, state.append(chat_history=result["chat_item"]).update(prompt=prompt)
+
+
+@trace()
+def _query_openai(prompt: str, chat_history: Optional[list] = None) -> str:
+    client = openai.Client()
+    result = client.chat.completions.create(
+        model="gpt-4o-mini",
+        messages=[
+            {"role": "system", "content": "You are a helpful assistant"},
+            *(chat_history or []),
+            {"role": "user", "content": prompt},
+        ],
+    )
+    return result.choices[0].message.content
+
+
+@action(reads=["prompt", "chat_history"], writes=["chat_history"])
+def chat_response(state: State, __tracer: TracerFactory) -> Tuple[dict, State]:
+    with __tracer("prepare_history"):
+        chat_history = state["chat_history"].copy()
+        __tracer.log_attributes(history_length=len(chat_history))
+    content = _query_openai(prompt=state["prompt"], chat_history=chat_history[:-1])
+    result = {"chat_item": {"role": "assistant", "content": content}}
+    return result, state.append(chat_history=result["chat_item"])
+
+
+def application(
+    app_id: Optional[str] = None,
+    user_id: Optional[str] = None,
+    bridge: Optional[LangfuseBridge] = None,
+) -> Application:
+    return (
+        ApplicationBuilder()
+        .with_actions(process_prompt, chat_response)
+        .with_transitions(
+            ("process_prompt", "chat_response"),
+            ("chat_response", "process_prompt"),
+        )
+        .with_state(chat_history=[])
+        .with_entrypoint("process_prompt")
+        .with_identifiers(app_id=app_id, partition_key=user_id)
+        # reads LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY/LANGFUSE_HOST from the environment.
+        # app_id maps to the Langfuse session, partition_key to the Langfuse user.
+        .with_hooks(bridge if bridge is not None else LangfuseBridge())
+        .build()
+    )
+
+
+if __name__ == "__main__":
+    bridge = LangfuseBridge()
+    app = application(user_id="example-user", bridge=bridge)
+    for user_prompt in ["What is Burr?", "And what is Langfuse?"]:
+        # each .run() call logs one trace to Langfuse
+        action_, result, state = app.run(
+            halt_after=["chat_response"], inputs={"prompt": user_prompt}
+        )
+        print(state["chat_history"][-1]["content"])
+    # flush before exiting a short-lived script -- the client batches exports
+    bridge.langfuse_client.flush()
diff --git a/pyproject.toml b/pyproject.toml
index 29f271e6b..6a17ee988 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -100,6 +100,7 @@ tests = [
   "apache-burr[pymongo]",
   "apache-burr[redis]",
   "apache-burr[opentelemetry]",
+  "apache-burr[langfuse]",
   "apache-burr[haystack]",
   "apache-burr[ray]"
 ]
@@ -222,6 +223,11 @@ opentelemetry = [
   "opentelemetry-sdk",
 ]
 
+langfuse = [
+  "langfuse>=3.0.0",
+  "apache-burr[opentelemetry]",
+]
+
 ray = [
   "ray[default]"
 ]
diff --git a/tests/integrations/test_burr_langfuse.py b/tests/integrations/test_burr_langfuse.py
new file mode 100644
index 000000000..418a1a99c
--- /dev/null
+++ b/tests/integrations/test_burr_langfuse.py
@@ -0,0 +1,278 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import json
+from contextlib import contextmanager
+from contextvars import ContextVar
+from typing import Tuple
+from unittest.mock import Mock
+
+import pytest
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.sdk.util.instrumentation import InstrumentationScope
+
+from burr.core import ApplicationBuilder, State, action
+from burr.integrations.langfuse import BURR_TRACER_NAME, LangfuseBridge, burr_span_export_filter
+from burr.visibility import TracerFactory
+
+
+@action(reads=[], writes=["count"])
+def counter_action(state: State, increment: int, __tracer: TracerFactory) -> Tuple[dict, State]:
+    with __tracer("inner_work"):
+        __tracer.log_attributes(
+            custom_attr="custom_value",
+            complex_attr=[{"role": "user", "content": "hi"}],
+            mixed_attr=[1, "two"],
+        )
+        result = {"count": state.get("count", 0) + increment}
+    return result, state.update(**result)
+
+
+@action(reads=["count"], writes=["done"])
+def finish_action(state: State) -> Tuple[dict, State]:
+    result = {"done": True}
+    return result, state.update(**result)
+
+
+@pytest.fixture
+def span_capture():
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    tracer = provider.get_tracer(BURR_TRACER_NAME)
+    return exporter, tracer
+
+
+def _build_app(bridge: LangfuseBridge):
+    return (
+        ApplicationBuilder()
+        .with_actions(counter_action, finish_action)
+        .with_transitions(("counter_action", "finish_action"))
+        .with_entrypoint("counter_action")
+        .with_state(count=0)
+        .with_identifiers(app_id="test-app-id", partition_key="test-user")
+        .with_hooks(bridge)
+        .build()
+    )
+
+
+def test_langfuse_bridge_span_structure_and_attributes(span_capture):
+    exporter, tracer = span_capture
+    bridge = LangfuseBridge(langfuse_client=Mock(), tracer=tracer)
+    app = _build_app(bridge)
+    app.run(halt_after=["finish_action"], inputs={"increment": 5})
+
+    spans = exporter.get_finished_spans()
+    spans_by_name = {span.name: span for span in spans}
+    # 1 trace root (run) + 2 steps + 1 tracer span
+    assert set(spans_by_name) == {"run", "counter_action", "finish_action", "inner_work"}
+
+    # all spans belong to a single trace, rooted at the execute call
+    root = spans_by_name["run"]
+    assert root.parent is None
+    assert all(span.context.trace_id == root.context.trace_id for span in spans)
+    assert spans_by_name["counter_action"].parent.span_id == root.context.span_id
+    assert spans_by_name["finish_action"].parent.span_id == root.context.span_id
+    assert (
+        spans_by_name["inner_work"].parent.span_id
+        == spans_by_name["counter_action"].context.span_id
+    )
+
+    # session/user attributes: app_id -> session, partition_key -> user
+    for name in ["run", "counter_action", "finish_action"]:
+        assert spans_by_name[name].attributes["langfuse.session.id"] == "test-app-id"
+        assert spans_by_name[name].attributes["langfuse.user.id"] == "test-user"
+        assert spans_by_name[name].attributes["session.id"] == "test-app-id"
+        assert spans_by_name[name].attributes["user.id"] == "test-user"
+
+    # step observation input/output
+    step_input = json.loads(
+        spans_by_name["counter_action"].attributes["langfuse.observation.input"]
+    )
+    assert step_input["inputs"] == {"increment": 5}
+    step_output = json.loads(
+        spans_by_name["counter_action"].attributes["langfuse.observation.output"]
+    )
+    assert step_output["state"] == {"count": 5}
+
+    # trace-level input/output from application state
+    trace_output = json.loads(root.attributes["langfuse.observation.output"])
+    assert trace_output["count"] == 5
+    assert trace_output["done"] is True
+
+    # attributes logged via __tracer land in Langfuse observation metadata
+    inner = spans_by_name["inner_work"]
+    assert inner.attributes["langfuse.observation.metadata.custom_attr"] == "custom_value"
+    # complex values (e.g. lists of dicts) are JSON-serialized, since OTel
+    # attributes only accept primitives and flat sequences of primitives
+    assert json.loads(inner.attributes["langfuse.observation.metadata.complex_attr"]) == [
+        {"role": "user", "content": "hi"}
+    ]
+    assert json.loads(inner.attributes["langfuse.observation.metadata.mixed_attr"]) == [1, "two"]
+
+
+def test_langfuse_bridge_capture_state_false(span_capture):
+    exporter, tracer = span_capture
+    bridge = LangfuseBridge(langfuse_client=Mock(), tracer=tracer, capture_state=False)
+    app = _build_app(bridge)
+    app.run(halt_after=["finish_action"], inputs={"increment": 5})
+
+    for span in exporter.get_finished_spans():
+        assert "langfuse.observation.input" not in span.attributes
+        assert "langfuse.observation.output" not in span.attributes
+
+
+def test_langfuse_bridge_session_and_user_overrides(span_capture):
+    exporter, tracer = span_capture
+    bridge = LangfuseBridge(
+        langfuse_client=Mock(),
+        tracer=tracer,
+        session_id="override-session",
+        user_id="override-user",
+    )
+    app = _build_app(bridge)
+    app.run(halt_after=["finish_action"], inputs={"increment": 1})
+
+    root = {span.name: span for span in exporter.get_finished_spans()}["run"]
+    assert root.attributes["langfuse.session.id"] == "override-session"
+    assert root.attributes["langfuse.user.id"] == "override-user"
+
+
+def test_langfuse_bridge_records_exceptions(span_capture):
+    exporter, tracer = span_capture
+
+    @action(reads=[], writes=[])
+    def failing_action(state: State) -> Tuple[dict, State]:
+        raise RuntimeError("boom")
+
+    bridge = LangfuseBridge(langfuse_client=Mock(), tracer=tracer)
+    app = (
+        ApplicationBuilder()
+        .with_actions(failing_action)
+        .with_transitions()
+        .with_entrypoint("failing_action")
+        .with_state()
+        .with_identifiers(app_id="test-app-id")
+        .with_hooks(bridge)
+        .build()
+    )
+    with pytest.raises(RuntimeError):
+        app.run(halt_after=["failing_action"])
+
+    spans_by_name = {span.name: span for span in exporter.get_finished_spans()}
+    assert not spans_by_name["failing_action"].status.is_ok
+
+
+def test_langfuse_bridge_rejects_client_and_kwargs():
+    with pytest.raises(ValueError):
+        LangfuseBridge(langfuse_client=Mock(), public_key="pk-lf-123")
+
+
+def test_langfuse_bridge_uses_custom_provider_and_propagates_trace_attributes(monkeypatch):
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    propagation_attributes = ContextVar("propagation_attributes", default=None)
+    langfuse_constructor = Mock(return_value=Mock())
+
+    @contextmanager
+    def capture_propagation_attributes(**attributes):
+        token = propagation_attributes.set(attributes)
+        try:
+            yield
+        finally:
+            propagation_attributes.reset(token)
+
+    monkeypatch.setattr(
+        "burr.integrations.langfuse.propagate_attributes", capture_propagation_attributes
+    )
+    monkeypatch.setattr("burr.integrations.langfuse.Langfuse", langfuse_constructor)
+
+    @action(reads=[], writes=["propagation"])
+    def propagated_action(state: State) -> Tuple[dict, State]:
+        result = {"propagation": propagation_attributes.get()}
+        return result, state.update(**result)
+
+    bridge = LangfuseBridge(
+        tracer_provider=provider,
+        public_key="pk-lf-test",
+        secret_key="sk-lf-test",
+    )
+    app = (
+        ApplicationBuilder()
+        .with_actions(propagated_action)
+        .with_transitions()
+        .with_entrypoint("propagated_action")
+        .with_identifiers(app_id="test-app-id", partition_key="test-user")
+        .with_hooks(bridge)
+        .build()
+    )
+    _, result, _ = app.run(halt_after=["propagated_action"])
+
+    assert {span.name for span in exporter.get_finished_spans()} == {
+        "run",
+        "propagated_action",
+    }
+    assert result["propagation"] == {
+        "session_id": "test-app-id",
+        "user_id": "test-user",
+    }
+    assert langfuse_constructor.call_args.kwargs["tracer_provider"] is provider
+    assert propagation_attributes.get() is None
+
+
+def test_langfuse_bridge_falls_back_without_propagate_attributes(span_capture, monkeypatch):
+    # langfuse < 3.9 has no propagate_attributes, and the bridge should still work with session/user set on Burr's own spans
+    exporter, tracer = span_capture
+    monkeypatch.setattr("burr.integrations.langfuse.propagate_attributes", None)
+    bridge = LangfuseBridge(langfuse_client=Mock(), tracer=tracer)
+    app = _build_app(bridge)
+    app.run(halt_after=["finish_action"], inputs={"increment": 2})
+
+    spans_by_name = {span.name: span for span in exporter.get_finished_spans()}
+    assert set(spans_by_name) == {"run", "counter_action", "finish_action", "inner_work"}
+    assert spans_by_name["run"].attributes["langfuse.session.id"] == "test-app-id"
+    assert spans_by_name["run"].attributes["langfuse.user.id"] == "test-user"
+
+
+def test_langfuse_bridge_rejects_tracer_and_provider(span_capture):
+    _, tracer = span_capture
+    with pytest.raises(ValueError):
+        LangfuseBridge(
+            langfuse_client=Mock(),
+            tracer=tracer,
+            tracer_provider=TracerProvider(),
+        )
+
+
+def _make_readable_span(scope_name, attributes=None) -> ReadableSpan:
+    return ReadableSpan(
+        name="test",
+        instrumentation_scope=InstrumentationScope(name=scope_name),
+        attributes=attributes or {},
+    )
+
+
+def test_burr_span_export_filter():
+    assert burr_span_export_filter(_make_readable_span(BURR_TRACER_NAME))
+    assert burr_span_export_filter(_make_readable_span(BURR_TRACER_NAME + ".sub"))
+    # spans exported by langfuse's default filter still pass
+    assert burr_span_export_filter(
+        _make_readable_span("some.other.scope", {"gen_ai.system": "openai"})
+    )