Temporal tracing with Future AGI

pip install "temporalio[opentelemetry]==1.34.0" opentelemetry-exporter-otlp-proto-http fi-instrumentation-otel traceai-openai openai

For a local test, start a Temporal dev server with temporal server start-dev (the Temporal CLI is free and needs no account). Point TEMPORAL_ADDRESS at your own cluster otherwise.

OpenTelemetryPlugin does not choose an endpoint, headers or a project. It reads the global tracer provider, which must come from Temporal’s create_tracer_provider. That function takes no endpoint or header parameters (create_tracer_provider(sampler, resource, shutdown_on_exit, active_span_processor, id_generator, span_limits)), so you pass the project as the resource and add the exporter as a span processor. This is tracing.py in the example:

# file: tracing.py
import os

from fi_instrumentation.otel import BatchSpanProcessor
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http import Compression
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from temporalio.contrib.opentelemetry import ReplaySafeTracerProvider, create_tracer_provider

...

def setup_tracing(
    instrument_openai: bool = True, exporter: Optional[SpanExporter] = None
) -> ReplaySafeTracerProvider:
    if exporter is None:
        base_url = os.environ.get("FI_BASE_URL", "https://api.futureagi.com").rstrip("/")
        exporter = OTLPSpanExporter(
            endpoint=f"{base_url}/tracer/v1/traces",
            headers={"X-Api-Key": os.environ["FI_API_KEY"], "X-Secret-Key": os.environ["FI_SECRET_KEY"]},
            compression=Compression.NoCompression,  # optional: Future AGI cloud and self-hosted v1.56.0+ also accept gzip
        )
    provider = create_tracer_provider(
        resource=Resource.create(
            {"project_name": os.environ.get("FI_PROJECT_NAME", "temporal-agent"), "project_type": "observe"}
        )
    )
    provider.add_span_processor(WorkflowIdAsSession())
    provider.add_span_processor(
        BatchSpanProcessor(
            span_exporter=SafeSpanExporter(
                exporter, keep_error_text=os.environ.get("KEEP_TEMPORAL_ERROR_TEXT", "").lower() == "true"
            )
        )
    )
    # OpenTelemetryPlugin reads the global provider, and refuses one that is not replay-safe.
    trace.set_tracer_provider(provider)
    if instrument_openai:
        from traceai_openai import OpenAIInstrumentor

        OpenAIInstrumentor().instrument(tracer_provider=provider)
    return provider

• project_name is a resource attribute, and it is required: a batch without it is dropped. service.name is not read as a project name, and a span attribute called project_name does not count.
• Both key headers are required. X-Api-Key and X-Secret-Key go on the exporter and nowhere else, never on a span.
• No gzip. The exporter is built with Compression.NoCompression, which works on every Future AGI release. Future AGI cloud and self-hosted v1.56.0 and later also accept gzip; earlier self-hosted releases answered it with HTTP 400.
• set_tracer_provider before Client.connect. Without it the worker refuses to start with When using OpenTelemetryPlugin, the global trace provider must be a ReplaySafeTracerProvider.
• fi_instrumentation.otel.BatchSpanProcessor is OpenTelemetry’s batch processor with traceAI’s usual rule that a span that ended without a status is stored as OK. You can use OpenTelemetry’s own BatchSpanProcessor instead.
• WorkflowIdAsSession and SafeSpanExporter are two small classes in the same file: the first makes the Temporal workflow id the Future AGI session of every span; the second removes error text (see Privacy).

LLM calls and tool calls belong in activities. Workflow code is replayed and has to stay deterministic. This is the shape of activities.py:

# file: activities.py
client = AsyncOpenAI(max_retries=0)


@activity.defn
async def llm_plan(model: str, messages: list[dict]) -> dict:
    return await _chat(model, messages, with_tools=True)


@activity.defn
async def get_weather(arguments: str, flaky: bool) -> str:
    """The tool. With flaky=True the first attempt fails, so Temporal retries the activity."""
    args = json.loads(arguments or "{}")
    with _tracer().start_as_current_span("get_weather", fi_span_kind="tool") as span:
        span.set_input(json.dumps(args))
        if flaky and activity.info().attempt == 1:
            raise TimeoutError(f"weather backend timed out for {args.get('city')}")
        result = f"{args.get('city', 'unknown')}: 21C and sunny"
        span.set_output(result)
        return result

Temporal retries a failed activity itself, so the OpenAI client is built with max_retries=0; otherwise one Temporal attempt could hide several model calls. Spans created inside async or synchronous activities are children of their RunActivity span.

Workflow code must not construct or import the exporter, the tracer module or the model SDK. It refers to the activities by name. workflows.py imports only temporalio and the standard library:

# file: workflows.py
from dataclasses import dataclass
from datetime import timedelta

from temporalio import workflow
from temporalio.common import RetryPolicy

...

@workflow.defn
class WeatherAgentWorkflow:
    @workflow.run
    async def run(self, req: AgentRequest) -> str:
        retry = RetryPolicy(initial_interval=timedelta(milliseconds=200), maximum_attempts=2)
        plan = await workflow.execute_activity(
            "llm_plan", args=[req.model, messages], start_to_close_timeout=timedelta(seconds=60), retry_policy=retry
        )

The plugin lets the opentelemetry package through the workflow sandbox, so you can also create your own spans inside workflow code with opentelemetry.trace.get_tracer(...). Nothing else tracing-related belongs there.

Both processes call setup_tracing() first, then connect a client with the plugin. A worker built from that client uses the plugin too; do not add it to the worker a second time. This is worker.py:

# file: worker.py
provider = setup_tracing()
# The plugin goes on the client. A worker built from that client gets it too.
client = await Client.connect(
    os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
    namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
    plugins=[OpenTelemetryPlugin(add_temporal_spans=True)],
)
...
async with worker:
    print("worker ready", flush=True)
    await stop.wait()
provider.force_flush()  # spans of the last activities and workflows are still batched
provider.shutdown()

and starter.py, which flushes before it exits:

# file: starter.py
provider = setup_tracing(instrument_openai=False)
client = await Client.connect(
    os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
    namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
    plugins=[OpenTelemetryPlugin(add_temporal_spans=True)],
)
...
finally:
    # A starter exits right after the call. Without this the StartWorkflow span is lost.
    provider.force_flush()
    provider.shutdown()

add_temporal_spans=True is what creates the StartWorkflow, RunWorkflow, StartActivity and RunActivity spans. Its default is False, and then the plugin only carries context: your own spans still nest correctly, but there are no Temporal spans in the trace.

export FI_API_KEY=... FI_SECRET_KEY=... FI_PROJECT_NAME=temporal-agent
export FI_BASE_URL=https://api.futureagi.com         # or your self-hosted Future AGI
export OPENAI_API_KEY=...                            # and OPENAI_BASE_URL for an OpenAI-compatible endpoint
python worker.py                                     # terminal 1
python starter.py "What is the weather in Lisbon today?"                 # terminal 2
python starter.py "What is the weather in Lisbon today?" --flaky         # the tool fails once and is retried

The trace shows up in Observe under the project from FI_PROJECT_NAME. Use the same FI_PROJECT_NAME in the starter and in the worker (see Troubleshooting).