Skip to content

Provider

SynchronousMultiTracingProcessor

Bases: TracingProcessor

Forwards all calls to a list of TracingProcessors, in order of registration.

Source code in src/agents/tracing/provider.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class SynchronousMultiTracingProcessor(TracingProcessor):
    """
    Forwards all calls to a list of TracingProcessors, in order of registration.
    """

    def __init__(self):
        # Using a tuple to avoid race conditions when iterating over processors
        self._processors: tuple[TracingProcessor, ...] = ()
        self._lock = threading.Lock()

    def add_tracing_processor(self, tracing_processor: TracingProcessor):
        """
        Add a processor to the list of processors. Each processor will receive all traces/spans.
        """
        with self._lock:
            self._processors += (tracing_processor,)

    def set_processors(self, processors: list[TracingProcessor]):
        """
        Set the list of processors. This will replace the current list of processors.
        """
        with self._lock:
            self._processors = tuple(processors)

    def on_trace_start(self, trace: Trace) -> None:
        """
        Called when a trace is started.
        """
        for processor in self._processors:
            try:
                processor.on_trace_start(trace)
            except Exception as e:
                logger.error(f"Error in trace processor {processor} during on_trace_start: {e}")

    def on_trace_end(self, trace: Trace) -> None:
        """
        Called when a trace is finished.
        """
        for processor in self._processors:
            try:
                processor.on_trace_end(trace)
            except Exception as e:
                logger.error(f"Error in trace processor {processor} during on_trace_end: {e}")

    def on_span_start(self, span: Span[Any]) -> None:
        """
        Called when a span is started.
        """
        for processor in self._processors:
            try:
                processor.on_span_start(span)
            except Exception as e:
                logger.error(f"Error in trace processor {processor} during on_span_start: {e}")

    def on_span_end(self, span: Span[Any]) -> None:
        """
        Called when a span is finished.
        """
        for processor in self._processors:
            try:
                processor.on_span_end(span)
            except Exception as e:
                logger.error(f"Error in trace processor {processor} during on_span_end: {e}")

    def shutdown(self) -> None:
        """
        Called when the application stops.
        """
        for processor in self._processors:
            logger.debug(f"Shutting down trace processor {processor}")
            try:
                processor.shutdown()
            except Exception as e:
                logger.error(f"Error shutting down trace processor {processor}: {e}")

    def force_flush(self):
        """
        Force the processors to flush their buffers.
        """
        for processor in self._processors:
            try:
                processor.force_flush()
            except Exception as e:
                logger.error(f"Error flushing trace processor {processor}: {e}")

add_tracing_processor 

add_tracing_processor(tracing_processor: TracingProcessor)

Add a processor to the list of processors. Each processor will receive all traces/spans.

Source code in src/agents/tracing/provider.py 
27
28
29
30
31
32
def add_tracing_processor(self, tracing_processor: TracingProcessor):
    """
    Add a processor to the list of processors. Each processor will receive all traces/spans.
    """
    with self._lock:
        self._processors += (tracing_processor,)

set_processors 

set_processors(processors: list[TracingProcessor])

Set the list of processors. This will replace the current list of processors.

Source code in src/agents/tracing/provider.py 
34
35
36
37
38
39
def set_processors(self, processors: list[TracingProcessor]):
    """
    Set the list of processors. This will replace the current list of processors.
    """
    with self._lock:
        self._processors = tuple(processors)

on_trace_start 

on_trace_start(trace: Trace) -> None

Called when a trace is started.

Source code in src/agents/tracing/provider.py 
41
42
43
44
45
46
47
48
49
def on_trace_start(self, trace: Trace) -> None:
    """
    Called when a trace is started.
    """
    for processor in self._processors:
        try:
            processor.on_trace_start(trace)
        except Exception as e:
            logger.error(f"Error in trace processor {processor} during on_trace_start: {e}")

on_trace_end 

on_trace_end(trace: Trace) -> None

Called when a trace is finished.

Source code in src/agents/tracing/provider.py 
51
52
53
54
55
56
57
58
59
def on_trace_end(self, trace: Trace) -> None:
    """
    Called when a trace is finished.
    """
    for processor in self._processors:
        try:
            processor.on_trace_end(trace)
        except Exception as e:
            logger.error(f"Error in trace processor {processor} during on_trace_end: {e}")

on_span_start 

on_span_start(span: Span[Any]) -> None

Called when a span is started.

Source code in src/agents/tracing/provider.py 
61
62
63
64
65
66
67
68
69
def on_span_start(self, span: Span[Any]) -> None:
    """
    Called when a span is started.
    """
    for processor in self._processors:
        try:
            processor.on_span_start(span)
        except Exception as e:
            logger.error(f"Error in trace processor {processor} during on_span_start: {e}")

on_span_end 

on_span_end(span: Span[Any]) -> None

Called when a span is finished.

Source code in src/agents/tracing/provider.py 
71
72
73
74
75
76
77
78
79
def on_span_end(self, span: Span[Any]) -> None:
    """
    Called when a span is finished.
    """
    for processor in self._processors:
        try:
            processor.on_span_end(span)
        except Exception as e:
            logger.error(f"Error in trace processor {processor} during on_span_end: {e}")

shutdown 

shutdown() -> None

Called when the application stops.

Source code in src/agents/tracing/provider.py 
81
82
83
84
85
86
87
88
89
90
def shutdown(self) -> None:
    """
    Called when the application stops.
    """
    for processor in self._processors:
        logger.debug(f"Shutting down trace processor {processor}")
        try:
            processor.shutdown()
        except Exception as e:
            logger.error(f"Error shutting down trace processor {processor}: {e}")

force_flush 

force_flush()

Force the processors to flush their buffers.

Source code in src/agents/tracing/provider.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
def force_flush(self):
    """
    Force the processors to flush their buffers.
    """
    for processor in self._processors:
        try:
            processor.force_flush()
        except Exception as e:
            logger.error(f"Error flushing trace processor {processor}: {e}")

TraceProvider

Bases: ABC

Interface for creating traces and spans.

Source code in src/agents/tracing/provider.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class TraceProvider(ABC):
    """Interface for creating traces and spans."""

    @abstractmethod
    def register_processor(self, processor: TracingProcessor) -> None:
        """Add a processor that will receive all traces and spans."""

    @abstractmethod
    def set_processors(self, processors: list[TracingProcessor]) -> None:
        """Replace the list of processors with ``processors``."""

    @abstractmethod
    def get_current_trace(self) -> Trace | None:
        """Return the currently active trace, if any."""

    @abstractmethod
    def get_current_span(self) -> Span[Any] | None:
        """Return the currently active span, if any."""

    @abstractmethod
    def set_disabled(self, disabled: bool) -> None:
        """Enable or disable tracing globally."""

    @abstractmethod
    def time_iso(self) -> str:
        """Return the current time in ISO 8601 format."""

    @abstractmethod
    def gen_trace_id(self) -> str:
        """Generate a new trace identifier."""

    @abstractmethod
    def gen_span_id(self) -> str:
        """Generate a new span identifier."""

    @abstractmethod
    def gen_group_id(self) -> str:
        """Generate a new group identifier."""

    @abstractmethod
    def create_trace(
        self,
        name: str,
        trace_id: str | None = None,
        group_id: str | None = None,
        metadata: dict[str, Any] | None = None,
        disabled: bool = False,
    ) -> Trace:
        """Create a new trace."""

    @abstractmethod
    def create_span(
        self,
        span_data: TSpanData,
        span_id: str | None = None,
        parent: Trace | Span[Any] | None = None,
        disabled: bool = False,
    ) -> Span[TSpanData]:
        """Create a new span."""

    @abstractmethod
    def shutdown(self) -> None:
        """Clean up any resources used by the provider."""

register_processor abstractmethod 

register_processor(processor: TracingProcessor) -> None

Add a processor that will receive all traces and spans.

Source code in src/agents/tracing/provider.py 
106
107
108
@abstractmethod
def register_processor(self, processor: TracingProcessor) -> None:
    """Add a processor that will receive all traces and spans."""

set_processors abstractmethod 

set_processors(processors: list[TracingProcessor]) -> None

Replace the list of processors with processors.

Source code in src/agents/tracing/provider.py 
110
111
112
@abstractmethod
def set_processors(self, processors: list[TracingProcessor]) -> None:
    """Replace the list of processors with ``processors``."""

get_current_trace abstractmethod 

get_current_trace() -> Trace | None

Return the currently active trace, if any.

Source code in src/agents/tracing/provider.py 
114
115
116
@abstractmethod
def get_current_trace(self) -> Trace | None:
    """Return the currently active trace, if any."""

get_current_span abstractmethod 

get_current_span() -> Span[Any] | None

Return the currently active span, if any.

Source code in src/agents/tracing/provider.py 
118
119
120
@abstractmethod
def get_current_span(self) -> Span[Any] | None:
    """Return the currently active span, if any."""

set_disabled abstractmethod 

set_disabled(disabled: bool) -> None

Enable or disable tracing globally.

Source code in src/agents/tracing/provider.py 
122
123
124
@abstractmethod
def set_disabled(self, disabled: bool) -> None:
    """Enable or disable tracing globally."""

time_iso abstractmethod 

time_iso() -> str

Return the current time in ISO 8601 format.

Source code in src/agents/tracing/provider.py 
126
127
128
@abstractmethod
def time_iso(self) -> str:
    """Return the current time in ISO 8601 format."""

gen_trace_id abstractmethod 

gen_trace_id() -> str

Generate a new trace identifier.

Source code in src/agents/tracing/provider.py 
130
131
132
@abstractmethod
def gen_trace_id(self) -> str:
    """Generate a new trace identifier."""

gen_span_id abstractmethod 

gen_span_id() -> str

Generate a new span identifier.

Source code in src/agents/tracing/provider.py 
134
135
136
@abstractmethod
def gen_span_id(self) -> str:
    """Generate a new span identifier."""

gen_group_id abstractmethod 

gen_group_id() -> str

Generate a new group identifier.

Source code in src/agents/tracing/provider.py 
138
139
140
@abstractmethod
def gen_group_id(self) -> str:
    """Generate a new group identifier."""

create_trace abstractmethod 

create_trace(
    name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace

Create a new trace.

Source code in src/agents/tracing/provider.py 
142
143
144
145
146
147
148
149
150
151
@abstractmethod
def create_trace(
    self,
    name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace:
    """Create a new trace."""

create_span abstractmethod 

create_span(
    span_data: TSpanData,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TSpanData]

Create a new span.

Source code in src/agents/tracing/provider.py 
153
154
155
156
157
158
159
160
161
@abstractmethod
def create_span(
    self,
    span_data: TSpanData,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TSpanData]:
    """Create a new span."""

shutdown abstractmethod 

shutdown() -> None

Clean up any resources used by the provider.

Source code in src/agents/tracing/provider.py 
163
164
165
@abstractmethod
def shutdown(self) -> None:
    """Clean up any resources used by the provider."""

DefaultTraceProvider

Bases: TraceProvider

Source code in src/agents/tracing/provider.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class DefaultTraceProvider(TraceProvider):
    def __init__(self) -> None:
        self._multi_processor = SynchronousMultiTracingProcessor()
        self._disabled = os.environ.get("OPENAI_AGENTS_DISABLE_TRACING", "false").lower() in (
            "true",
            "1",
        )

    def register_processor(self, processor: TracingProcessor):
        """
        Add a processor to the list of processors. Each processor will receive all traces/spans.
        """
        self._multi_processor.add_tracing_processor(processor)

    def set_processors(self, processors: list[TracingProcessor]):
        """
        Set the list of processors. This will replace the current list of processors.
        """
        self._multi_processor.set_processors(processors)

    def get_current_trace(self) -> Trace | None:
        """
        Returns the currently active trace, if any.
        """
        return Scope.get_current_trace()

    def get_current_span(self) -> Span[Any] | None:
        """
        Returns the currently active span, if any.
        """
        return Scope.get_current_span()

    def set_disabled(self, disabled: bool) -> None:
        """
        Set whether tracing is disabled.
        """
        self._disabled = disabled

    def time_iso(self) -> str:
        """Return the current time in ISO 8601 format."""
        return datetime.now(timezone.utc).isoformat()

    def gen_trace_id(self) -> str:
        """Generate a new trace ID."""
        return f"trace_{uuid.uuid4().hex}"

    def gen_span_id(self) -> str:
        """Generate a new span ID."""
        return f"span_{uuid.uuid4().hex[:24]}"

    def gen_group_id(self) -> str:
        """Generate a new group ID."""
        return f"group_{uuid.uuid4().hex[:24]}"

    def create_trace(
        self,
        name: str,
        trace_id: str | None = None,
        group_id: str | None = None,
        metadata: dict[str, Any] | None = None,
        disabled: bool = False,
    ) -> Trace:
        """
        Create a new trace.
        """
        if self._disabled or disabled:
            logger.debug(f"Tracing is disabled. Not creating trace {name}")
            return NoOpTrace()

        trace_id = trace_id or self.gen_trace_id()

        logger.debug(f"Creating trace {name} with id {trace_id}")

        return TraceImpl(
            name=name,
            trace_id=trace_id,
            group_id=group_id,
            metadata=metadata,
            processor=self._multi_processor,
        )

    def create_span(
        self,
        span_data: TSpanData,
        span_id: str | None = None,
        parent: Trace | Span[Any] | None = None,
        disabled: bool = False,
    ) -> Span[TSpanData]:
        """
        Create a new span.
        """
        if self._disabled or disabled:
            logger.debug(f"Tracing is disabled. Not creating span {span_data}")
            return NoOpSpan(span_data)

        if not parent:
            current_span = Scope.get_current_span()
            current_trace = Scope.get_current_trace()
            if current_trace is None:
                logger.error(
                    "No active trace. Make sure to start a trace with `trace()` first "
                    "Returning NoOpSpan."
                )
                return NoOpSpan(span_data)
            elif isinstance(current_trace, NoOpTrace) or isinstance(current_span, NoOpSpan):
                logger.debug(
                    f"Parent {current_span} or {current_trace} is no-op, returning NoOpSpan"
                )
                return NoOpSpan(span_data)

            parent_id = current_span.span_id if current_span else None
            trace_id = current_trace.trace_id

        elif isinstance(parent, Trace):
            if isinstance(parent, NoOpTrace):
                logger.debug(f"Parent {parent} is no-op, returning NoOpSpan")
                return NoOpSpan(span_data)
            trace_id = parent.trace_id
            parent_id = None
        elif isinstance(parent, Span):
            if isinstance(parent, NoOpSpan):
                logger.debug(f"Parent {parent} is no-op, returning NoOpSpan")
                return NoOpSpan(span_data)
            parent_id = parent.span_id
            trace_id = parent.trace_id

        logger.debug(f"Creating span {span_data} with id {span_id}")

        return SpanImpl(
            trace_id=trace_id,
            span_id=span_id or self.gen_span_id(),
            parent_id=parent_id,
            processor=self._multi_processor,
            span_data=span_data,
        )

    def shutdown(self) -> None:
        if self._disabled:
            return

        try:
            logger.debug("Shutting down trace provider")
            self._multi_processor.shutdown()
        except Exception as e:
            logger.error(f"Error shutting down trace provider: {e}")

register_processor 

register_processor(processor: TracingProcessor)

Add a processor to the list of processors. Each processor will receive all traces/spans.

Source code in src/agents/tracing/provider.py 
176
177
178
179
180
def register_processor(self, processor: TracingProcessor):
    """
    Add a processor to the list of processors. Each processor will receive all traces/spans.
    """
    self._multi_processor.add_tracing_processor(processor)

set_processors 

set_processors(processors: list[TracingProcessor])

Set the list of processors. This will replace the current list of processors.

Source code in src/agents/tracing/provider.py 
182
183
184
185
186
def set_processors(self, processors: list[TracingProcessor]):
    """
    Set the list of processors. This will replace the current list of processors.
    """
    self._multi_processor.set_processors(processors)

get_current_trace 

get_current_trace() -> Trace | None

Returns the currently active trace, if any.

Source code in src/agents/tracing/provider.py 
188
189
190
191
192
def get_current_trace(self) -> Trace | None:
    """
    Returns the currently active trace, if any.
    """
    return Scope.get_current_trace()

get_current_span 

get_current_span() -> Span[Any] | None

Returns the currently active span, if any.

Source code in src/agents/tracing/provider.py 
194
195
196
197
198
def get_current_span(self) -> Span[Any] | None:
    """
    Returns the currently active span, if any.
    """
    return Scope.get_current_span()

set_disabled 

set_disabled(disabled: bool) -> None

Set whether tracing is disabled.

Source code in src/agents/tracing/provider.py 
200
201
202
203
204
def set_disabled(self, disabled: bool) -> None:
    """
    Set whether tracing is disabled.
    """
    self._disabled = disabled

time_iso 

time_iso() -> str

Return the current time in ISO 8601 format.

Source code in src/agents/tracing/provider.py 
206
207
208
def time_iso(self) -> str:
    """Return the current time in ISO 8601 format."""
    return datetime.now(timezone.utc).isoformat()

gen_trace_id 

gen_trace_id() -> str

Generate a new trace ID.

Source code in src/agents/tracing/provider.py 
210
211
212
def gen_trace_id(self) -> str:
    """Generate a new trace ID."""
    return f"trace_{uuid.uuid4().hex}"

gen_span_id 

gen_span_id() -> str

Generate a new span ID.

Source code in src/agents/tracing/provider.py 
214
215
216
def gen_span_id(self) -> str:
    """Generate a new span ID."""
    return f"span_{uuid.uuid4().hex[:24]}"

gen_group_id 

gen_group_id() -> str

Generate a new group ID.

Source code in src/agents/tracing/provider.py 
218
219
220
def gen_group_id(self) -> str:
    """Generate a new group ID."""
    return f"group_{uuid.uuid4().hex[:24]}"

create_trace 

create_trace(
    name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace

Create a new trace.

Source code in src/agents/tracing/provider.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def create_trace(
    self,
    name: str,
    trace_id: str | None = None,
    group_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    disabled: bool = False,
) -> Trace:
    """
    Create a new trace.
    """
    if self._disabled or disabled:
        logger.debug(f"Tracing is disabled. Not creating trace {name}")
        return NoOpTrace()

    trace_id = trace_id or self.gen_trace_id()

    logger.debug(f"Creating trace {name} with id {trace_id}")

    return TraceImpl(
        name=name,
        trace_id=trace_id,
        group_id=group_id,
        metadata=metadata,
        processor=self._multi_processor,
    )

create_span 

create_span(
    span_data: TSpanData,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TSpanData]

Create a new span.

Source code in src/agents/tracing/provider.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def create_span(
    self,
    span_data: TSpanData,
    span_id: str | None = None,
    parent: Trace | Span[Any] | None = None,
    disabled: bool = False,
) -> Span[TSpanData]:
    """
    Create a new span.
    """
    if self._disabled or disabled:
        logger.debug(f"Tracing is disabled. Not creating span {span_data}")
        return NoOpSpan(span_data)

    if not parent:
        current_span = Scope.get_current_span()
        current_trace = Scope.get_current_trace()
        if current_trace is None:
            logger.error(
                "No active trace. Make sure to start a trace with `trace()` first "
                "Returning NoOpSpan."
            )
            return NoOpSpan(span_data)
        elif isinstance(current_trace, NoOpTrace) or isinstance(current_span, NoOpSpan):
            logger.debug(
                f"Parent {current_span} or {current_trace} is no-op, returning NoOpSpan"
            )
            return NoOpSpan(span_data)

        parent_id = current_span.span_id if current_span else None
        trace_id = current_trace.trace_id

    elif isinstance(parent, Trace):
        if isinstance(parent, NoOpTrace):
            logger.debug(f"Parent {parent} is no-op, returning NoOpSpan")
            return NoOpSpan(span_data)
        trace_id = parent.trace_id
        parent_id = None
    elif isinstance(parent, Span):
        if isinstance(parent, NoOpSpan):
            logger.debug(f"Parent {parent} is no-op, returning NoOpSpan")
            return NoOpSpan(span_data)
        parent_id = parent.span_id
        trace_id = parent.trace_id

    logger.debug(f"Creating span {span_data} with id {span_id}")

    return SpanImpl(
        trace_id=trace_id,
        span_id=span_id or self.gen_span_id(),
        parent_id=parent_id,
        processor=self._multi_processor,
        span_data=span_data,
    )