Azure · Harsheet Shah (harsheet-shah) · Sep 10, 2026 · Sep 10, 2026
@@ -4,6 +4,14 @@
 
 ### Features Added
 
+- Added `flush_spans_async`, a non-blocking variant of `flush_spans` that
+  offloads the blocking `TracerProvider.force_flush` to a worker thread so it
+  does not stall the asyncio event loop, and `schedule_flush_spans`, a
+  fire-and-forget helper that flushes in the background without delaying the
+  caller. Background flushes are coalesced: at most one runs at a time and
+  concurrent requests collapse into a single follow-up flush, so the work does
+  not grow with the request rate.
+
 ### Breaking Changes
 
 ### Bugs Fixed

@@ -56,12 +56,18 @@ namespace azure.ai.agentserver.core
     def azure.ai.agentserver.core.resolve_state_subdir(name: str) -> Path: ...
 
 
+    def azure.ai.agentserver.core.schedule_flush_spans(timeout_millis: int = 5000) -> None: ...
+
+
     def azure.ai.agentserver.core.set_current_span(span: Any) -> Any: ...
 
 
     def azure.ai.agentserver.core.set_request_context(context: FoundryAgentRequestContext) -> Token[FoundryAgentRequestContext]: ...
 
 
+    async def azure.ai.agentserver.core.flush_spans_async:async(timeout_millis: int = 5000) -> None: ...
+
+
     async def azure.ai.agentserver.core.trace_stream:async(iterator: AsyncIterable[StreamContent], span: Any) -> AsyncIterator[StreamContent]: ...
 
 

@@ -30,7 +30,9 @@
     detach_context,
     end_span,
     flush_spans,
+    flush_spans_async,
     record_error,
+    schedule_flush_spans,
     set_current_span,
     trace_stream,
 )
@@ -52,6 +54,8 @@
     "end_span",
     "experimental",
     "flush_spans",
+    "flush_spans_async",
+    "schedule_flush_spans",
     "get_request_context",
     "record_error",
     "read_request_id",

@@ -33,6 +33,7 @@
 OpenTelemetry is a required dependency — these functions always create
 real spans.  Azure Monitor export is optional (auto-configured by the distro).
 """
+import asyncio  # pylint: disable=do-not-import-asyncio
 from collections.abc import AsyncIterable, AsyncIterator  # pylint: disable=import-error
 from contextlib import contextmanager, nullcontext
 import logging
@@ -583,6 +584,96 @@ def flush_spans(timeout_millis: int = 5000) -> None:
             logger.debug("TracerProvider.force_flush() failed", exc_info=True)
 
 
+# A single coalesced background flush runs at a time.  ``force_flush`` drains
+# the provider *globally*, so concurrent per-request flushes would be redundant
+# work.  Instead of spawning one task per request (which lets ``_bg_flush_task``
+# / the executor queue grow without bound under load), requests that arrive
+# while a flush is in flight set ``_bg_flush_pending``; the running task then
+# performs exactly one follow-up flush afterwards to capture spans produced
+# during the active flush.  This bounds in-flight background work to a single
+# task regardless of request rate.  The module-level reference also keeps the
+# task alive (asyncio only holds weak references to tasks).  Access is confined
+# to the event-loop thread, so no lock is required.
+_bg_flush_task: "Optional[asyncio.Task[None]]" = None
+_bg_flush_pending: bool = False
+
+
+async def flush_spans_async(timeout_millis: int = 5000) -> None:
+    """Non-blocking variant of :func:`flush_spans`.
+
+    ``TracerProvider.force_flush`` blocks the calling thread until the exporter
+    drains its queue.  On the request hot path -- which runs inside an ``async``
+    handler -- that blocks the asyncio event loop, serialising every concurrent
+    request behind a single export (head-of-line blocking).  Offload the
+    blocking call to the default thread pool so the event loop stays free to
+    send the response and service other requests concurrently.
+
+    No-op when the OTel SDK is not installed or the provider does not support
+    ``force_flush``.
+
+    :param timeout_millis: Maximum time to wait for the flush, in milliseconds.
+        Defaults to 5000 (5 seconds).
+    :type timeout_millis: int
+    """
+    provider = trace.get_tracer_provider()
+    flush = getattr(provider, "force_flush", None)
+    if flush is None:
+        return
+    try:
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(None, flush, timeout_millis)
+    except Exception:  # pylint: disable=broad-exception-caught
+        logger.debug("TracerProvider.force_flush() (async) failed", exc_info=True)
+
+
+async def _coalesced_flush(timeout_millis: int) -> None:
+    """Run a background flush, then one more pass per pending coalesced request.
+
+    Because ``force_flush`` drains the provider globally, a single trailing
+    flush captures the spans of every request that arrived while a flush was
+    already running -- no need for a task (or export) per request.
+    """
+    global _bg_flush_pending  # pylint: disable=global-statement
+    await flush_spans_async(timeout_millis)
+    while _bg_flush_pending:
+        _bg_flush_pending = False
+        await flush_spans_async(timeout_millis)
+
+
+def schedule_flush_spans(timeout_millis: int = 5000) -> None:
+    """Schedule a coalesced background span flush and return immediately.
+
+    Unlike :func:`flush_spans` / :func:`flush_spans_async`, this does not delay
+    the caller (i.e. the HTTP response) by the export duration.  At most one
+    background flush task runs at a time: calls made while a flush is in flight
+    are coalesced into a single follow-up flush rather than spawning a task per
+    request, so neither the retained task reference nor the executor queue grows
+    with the request rate.  Falls back to a synchronous flush when no event loop
+    is running.
+
+    .. note::
+       Only safe when the hosting platform guarantees a brief drain window
+       before it suspends/freezes the process after sending a response;
+       otherwise the final request's spans may be lost.
+
+    :param timeout_millis: Maximum time to wait for the flush, in milliseconds.
+        Defaults to 5000 (5 seconds).
+    :type timeout_millis: int
+    """
+    global _bg_flush_task, _bg_flush_pending  # pylint: disable=global-statement
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        flush_spans(timeout_millis)
+        return
+    if _bg_flush_task is not None and not _bg_flush_task.done():
+        # A flush is already draining the provider globally; record that more
+        # spans arrived so the running task performs one more pass afterwards.
+        _bg_flush_pending = True
+        return
+    _bg_flush_task = loop.create_task(_coalesced_flush(timeout_millis))
+
+
 def record_error(span: Any, exc: BaseException) -> None:
     """Record an exception and ERROR status on a span.
 

@@ -3,7 +3,9 @@
 # ---------------------------------------------------------
 """Tests for tracing configuration — not invocation spans (those live in the invocations package)."""
 
+import asyncio
 import os
+import pytest
 from functools import partial
 from threading import Event, Thread
 from typing import Any, Optional
@@ -25,6 +27,7 @@
     _BaggageLogRecordProcessor,
     _FoundryEnrichmentSpanProcessor,
 )
+from azure.ai.agentserver.core import _tracing
 
 
 class _CollectorExporter(SpanExporter):
@@ -938,3 +941,120 @@ def test_does_not_overwrite_existing_log_attributes(self) -> None:
         assert attrs["gen_ai.agent.name"] == "existing-name"
         assert attrs["gen_ai.agent.version"] == "0.0.1"
         assert attrs["microsoft.session.id"] == "existing-session"
+
+
+# ------------------------------------------------------------------ #
+# flush_spans_async / schedule_flush_spans (non-blocking flush)
+# ------------------------------------------------------------------ #
+
+
+class _BlockingFlushProvider:
+    """Fake TracerProvider whose force_flush blocks until released.
+
+    Records each timeout it is called with so tests can assert pass-through
+    and coalescing behaviour.
+    """
+
+    def __init__(self, block_first: bool = False) -> None:
+        self.calls: list = []
+        self._block_first = block_first
+        self.started = Event()
+        self.release = Event()
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        first = len(self.calls) == 0
+        self.calls.append(timeout_millis)
+        if first:
+            self.started.set()
+            if self._block_first:
+                self.release.wait(5)
+        return True
+
+
+async def _wait_for(flag: Event, *, timeout_s: float = 5.0) -> None:
+    """Yield to the event loop until *flag* is set (proves the loop is free)."""
+    deadline = asyncio.get_running_loop().time() + timeout_s
+    while not flag.is_set():
+        assert asyncio.get_running_loop().time() < deadline, "flag never set"
+        await asyncio.sleep(0.001)
+
+
+class TestFlushSpansAsync:
+    """`flush_spans_async` offloads the blocking export off the event loop."""
+
+    @pytest.mark.asyncio
+    async def test_does_not_block_event_loop_and_passes_timeout(self) -> None:
+        provider = _BlockingFlushProvider(block_first=True)
+        with mock.patch.object(_tracing.trace, "get_tracer_provider", return_value=provider):
+            flush_task = asyncio.create_task(_tracing.flush_spans_async(1234))
+            # If force_flush ran inline it would block here; instead this
+            # coroutine keeps running while the export blocks in a worker thread.
+            await _wait_for(provider.started)
+            assert not flush_task.done(), "flush should still be draining off-loop"
+            provider.release.set()
+            await flush_task
+        assert provider.calls == [1234]  # timeout argument forwarded
+
+    @pytest.mark.asyncio
+    async def test_swallows_exceptions(self) -> None:
+        class _BoomProvider:
+            def force_flush(self, timeout_millis: int = 30000) -> bool:
+                raise RuntimeError("boom")
+
+        with mock.patch.object(_tracing.trace, "get_tracer_provider", return_value=_BoomProvider()):
+            await _tracing.flush_spans_async()  # must not raise
+
+    @pytest.mark.asyncio
+    async def test_noop_when_provider_has_no_force_flush(self) -> None:
+        class _NoFlushProvider:
+            pass
+
+        with mock.patch.object(_tracing.trace, "get_tracer_provider", return_value=_NoFlushProvider()):
+            await _tracing.flush_spans_async()  # must not raise
+
+
+class TestScheduleFlushSpans:
+    """`schedule_flush_spans` runs a single coalesced background flush."""
+
+    def setup_method(self) -> None:
+        _tracing._bg_flush_task = None
+        _tracing._bg_flush_pending = False
+
+    def teardown_method(self) -> None:
+        _tracing._bg_flush_task = None
+        _tracing._bg_flush_pending = False
+
+    def test_falls_back_to_sync_without_running_loop(self) -> None:
+        with mock.patch.object(_tracing, "flush_spans") as m_sync:
+            _tracing.schedule_flush_spans(777)
+        m_sync.assert_called_once_with(777)
+        assert _tracing._bg_flush_task is None
+
+    @pytest.mark.asyncio
+    async def test_coalesces_concurrent_requests_into_one_followup(self) -> None:
+        provider = _BlockingFlushProvider(block_first=True)
+        with mock.patch.object(_tracing.trace, "get_tracer_provider", return_value=provider):
+            # First call spawns exactly one background flush task.
+            _tracing.schedule_flush_spans()
+            task = _tracing._bg_flush_task
+            assert task is not None
+            await _wait_for(provider.started)
+            # 50 requests arrive while the flush is in flight. They must NOT
+            # each spawn a task (bounded); they coalesce into one pending pass.
+            for _ in range(50):
+                _tracing.schedule_flush_spans()
+            assert _tracing._bg_flush_task is task
+            assert _tracing._bg_flush_pending is True
+            provider.release.set()
+            await task
+        # Initial flush + one coalesced follow-up == 2 exports, not 51.
+        assert len(provider.calls) == 2
+        assert _tracing._bg_flush_pending is False
+
+    @pytest.mark.asyncio
+    async def test_single_request_flushes_once(self) -> None:
+        provider = _BlockingFlushProvider(block_first=False)
+        with mock.patch.object(_tracing.trace, "get_tracer_provider", return_value=provider):
+            _tracing.schedule_flush_spans()
+            await _tracing._bg_flush_task
+        assert len(provider.calls) == 1
@@ -4,6 +4,16 @@
 
 ### Bugs Fixed
 
+- The per-request span flush in the Responses endpoint no longer blocks the
+  asyncio event loop. The synchronous `flush_spans()` call in the request
+  `finally` block ran `TracerProvider.force_flush` inline, which blocks the
+  event loop until the exporter drains and serialises concurrent requests
+  behind a single export (head-of-line blocking). It now awaits the
+  non-blocking `flush_spans_async()` by default. A new `AGENTSERVER_FLUSH_MODE`
+  environment variable selects the strategy: `async` (default, off the event
+  loop), `background` (return the response first, flush in the background --
+  requires a platform drain window), or `sync` (legacy blocking behaviour).
+
 - Scoped durable multi-turn task IDs with `FOUNDRY_AGENT_SESSION_GUID` when
   available, preventing recreated same-name sessions from colliding with task
   tombstones. Existing pre-rollout active chains remain resumable through a

@@ -13,6 +13,7 @@
 import asyncio  # pylint: disable=do-not-import-asyncio
 import contextvars
 import logging
+import os
 import threading
 from typing import TYPE_CHECKING, Any, cast
 
@@ -24,7 +25,9 @@
 from azure.ai.agentserver.core import (  # pylint: disable=import-error,no-name-in-module
     FoundryAgentRequestContext,
     flush_spans,
+    flush_spans_async,
     reset_request_context,
+    schedule_flush_spans,
     set_request_context,
 )
 from azure.ai.agentserver.core.tasks import (
@@ -187,6 +190,46 @@ def _get_scope_request_id(request: Request) -> str | None:
 _streaming_var: contextvars.ContextVar[str] = contextvars.ContextVar("Streaming", default="")
 
 
+_FLUSH_MODE_ENV = "AGENTSERVER_FLUSH_MODE"
+_DEFAULT_FLUSH_MODE = "async"
+
+
+async def _flush_spans_for_mode(mode: str) -> None:
+    """Dispatch span flushing according to *mode* (see ``AGENTSERVER_FLUSH_MODE``).
+
+    ``force_flush`` blocks the calling thread until the exporter drains; doing
+    that inline on this ``async`` handler blocks the event loop and serialises
+    concurrent requests behind one export.  The mode selects the strategy:
+
+    * ``"async"`` (default) -> :func:`flush_spans_async`: off the event loop;
+      same durability, no head-of-line blocking under concurrency.
+    * ``"background"`` -> :func:`schedule_flush_spans`: return the response
+      first and flush in the background (lowest latency, but needs the platform
+      to grant a brief drain window before freezing).
+    * ``"sync"`` -> :func:`flush_spans`: legacy blocking behaviour.
+
+    Any unrecognised value falls back to the ``"async"`` default (fail safe:
+    never silently drop telemetry).
+
+    :param mode: The flush mode; matched case-insensitively.
+    :type mode: str
+    """
+    normalized = (mode or "").strip().lower()
+    if normalized == "sync":
+        flush_spans()
+    elif normalized == "background":
+        schedule_flush_spans()
+    else:
+        if normalized and normalized != _DEFAULT_FLUSH_MODE:
+            logger.warning(
+                "Unrecognised %s=%r; falling back to %r flush mode.",
+                _FLUSH_MODE_ENV,
+                mode,
+                _DEFAULT_FLUSH_MODE,
+            )
+        await flush_spans_async()
+
+
 class _ResponseLogFilter(logging.Filter):
     """Attach response-scope IDs to every log record from context vars.
 
@@ -949,11 +992,13 @@ async def _iter_with_context():  # type: ignore[return]
             _conversation_id_var.reset(cid_token)
             _streaming_var.reset(str_token)
             reset_request_context(platform_ctx_token)
-            # Flush pending spans before the response is sent.
-            # BatchSpanProcessor exports on a timer; in hosted sandboxes
-            # the platform may freeze the process after the HTTP response,
-            # losing any buffered spans (e.g. LangGraph per-node spans).
-            flush_spans()
+            # Flush pending spans before the process may be frozen.
+            # ``AGENTSERVER_FLUSH_MODE`` selects the strategy (see
+            # ``_flush_spans_for_mode``); the default keeps the flush off the
+            # event loop without dropping telemetry.
+            await _flush_spans_for_mode(
+                os.environ.get(_FLUSH_MODE_ENV, _DEFAULT_FLUSH_MODE)
+            )
             try:
                 _otel_context.detach(baggage_token)
             except ValueError: