temporalio.worker.ActivityInboundInterceptor

class documentation

class ActivityInboundInterceptor: (source)

Known subclasses: temporalio.contrib.openai_agents._trace_interceptor._ContextPropagationActivityInboundInterceptor, temporalio.contrib.opentelemetry._interceptor._TracingActivityInboundInterceptor, temporalio.contrib.opentelemetry._otel_interceptor._TracingActivityInboundInterceptor, temporalio.worker._activity._ActivityInboundImpl

Constructor: ActivityInboundInterceptor(next)

View In Hierarchy

Inbound interceptor to wrap outbound creation and activity execution.

This should be extended by any activity inbound interceptors.

Method	`__init__`	Create the inbound interceptor.
Async Method	`execute_activity`	Called to invoke the activity.
Method	`init`	Initialize with an outbound interceptor.
Instance Variable	`next`	Undocumented

def __init__(self, next: ActivityInboundInterceptor): (source) ¶

overridden in temporalio.contrib.openai_agents._trace_interceptor._ContextPropagationActivityInboundInterceptor, temporalio.contrib.opentelemetry._interceptor._TracingActivityInboundInterceptor, temporalio.contrib.opentelemetry._otel_interceptor._TracingActivityInboundInterceptor, temporalio.worker._activity._ActivityInboundImpl

Create the inbound interceptor.

Parameters
next:`ActivityInboundInterceptor`	The next interceptor in the chain. The default implementation of all calls is to delegate to the next interceptor.

async def execute_activity(self, input: ExecuteActivityInput) -> Any: (source) ¶

Called to invoke the activity.

def init(self, outbound: ActivityOutboundInterceptor): (source) ¶

overridden in temporalio.worker._activity._ActivityInboundImpl

Initialize with an outbound interceptor.

To add a custom outbound interceptor, wrap the given interceptor before sending to the next init call.

next = (source) ¶

Undocumented