Multi-Runtime Agent Observability: One Fleet View Across LangChain, CrewAI, AutoGen, and Custom Runtimes

A LangChain callback may call a run a chain, a CrewAI event may describe an agent or task, and an OpenTelemetry trace may organize work as spans. Runtime-native tools are useful for understanding their own execution, but a fleet incident asks a broader question: are separate agents showing the same failure pattern, and when did it begin?

The goal of a shared fleet view is to line up lifecycle, duration, and failure signals while keeping each runtime’s original trace context available for deeper debugging. That mapping and its FleetPulse delivery contract are not published yet, so treat the common view described here as a product direction rather than a feature you can enable today.

LangChain exposes lifecycle callbacks through BaseCallbackHandler. Attach a handler to a Runnable invocation and use its run identifier and completion hook as the seam for a sanitized adapter.

from typing import Any
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler


class FleetPulseCallback(BaseCallbackHandler):
    def on_chain_end(
        self,
        outputs: dict[str, Any],
        *,
        run_id: UUID,
        **kwargs: Any,
    ) -> None:
        # Illustrative FleetPulse adapter placeholder; no shipped API is implied.
        pass


result = chain.invoke(
    {"request": "summarize status"},
    config={"callbacks": [FleetPulseCallback()]},
)

CrewAI’s BaseEventListener subscribes to typed lifecycle events on the crew event bus. Keep a listener instance loaded with the crew, then choose the event boundaries that matter to your operators.

from crewai.events import AgentExecutionCompletedEvent, BaseEventListener


class FleetPulseListener(BaseEventListener):
    def __init__(self):
        super().__init__()

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_execution_completed(source, event):
            # Illustrative FleetPulse adapter placeholder; forward selected, sanitized signals.
            pass


fleetpulse_listener = FleetPulseListener()

AutoGen’s AgentChat runtime is instrumented for OpenTelemetry. Configure a tracer provider and span processor for local inspection now; connecting a FleetPulse exporter is a placeholder until that exporter and its settings are documented.

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter


provider = TracerProvider(resource=Resource({"service.name": "support-agent"}))
# Console export is for local inspection; the FleetPulse exporter is illustrative.
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

For a custom worker, OpenTelemetry gives you direct control over the provider and trace boundaries. An OTLP exporter can send spans to a collector; the FleetPulse destination and authentication settings remain illustrative until a product contract is available.

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


provider = TracerProvider(
    resource=Resource.create({"service.name": "custom-agent-worker"})
)
# FleetPulse destination/configuration is illustrative until documented.
exporter = OTLPSpanExporter()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("agent.run"):
    run_agent()

A common view can help an on-call engineer compare failures and timing across runtimes, then follow the source trace into the framework that owns the work. Native hooks are the starting point; useful fleet context depends on a documented adapter, safe field selection, and a supported destination.

• Choose stable service and agent identities so runs can be grouped across deployments.
• Record lifecycle and error signals without forwarding prompts, responses, or secrets by default.
• Preserve each runtime’s trace identifier so a fleet-level alert can lead back to native details.