Nexus Python SDK - synchronous operation handler
Use the @nexusrpc.handler.sync_operation decorator to create synchronous Nexus Operation handlers. A synchronous operation handler must return within 10 seconds. Implementations should be reliable to avoid tripping the circuit breaker. Example: @nexusrpc.handler.sync_operation async def my_sync_operation(self, ctx: nexusrpc.handler.StartOperationContext, input: MyInput) -> MyOutput: return MyOutput(message=f"Hello {input.name} from sync operation!")
Nexus Python SDK - asynchronous operation handler with workflow
Use the @nexus.workflow_run_operation decorator to expose a Workflow as an asynchronous Nexus operation. This is the easiest way to create an asynchronous operation handler. The decorator automatically handles operation lifecycle and result delivery. Workflow IDs should typically be business-meaningful IDs used for deduplication.
Nexus Python SDK - register handler in Worker
Register a Nexus Service handler in a Worker by passing it to the nexus_service_handlers parameter when creating the Worker instance. Example: worker = Worker(client, task_queue=TASK_QUEUE, workflows=[WorkflowStartedByNexusOperation], nexus_service_handlers=[MyNexusServiceHandler()])
Nexus Python SDK - multiple workflow arguments from single operation input
Use ctx.start_workflow() with the args parameter as a list to map a single Nexus Operation input to multiple Workflow arguments. Example: return await ctx.start_workflow(HelloHandlerWorkflow.run, args=[input.name, input.language], id=f"hello-multi-args-{input.name}-{input.language}")
Nexus Python SDK - use client from within sync handler
Common pattern in sync handlers is to use the Temporal Client to Signal, Query, or Update a Workflow. Use Signal-With-Start or Update-With-Start to ensure Workflow is started. All calls must complete within the Nexus request timeout (10 seconds). Use nexus.client() to get the Worker's Client. Example shows using get_workflow_handle_for() to get a workflow handle.
Nexus Python SDK - workflow run operation context
Asynchronous operation handlers use nexus.WorkflowRunOperationContext. Call ctx.start_workflow() to start a Workflow and return a nexus.WorkflowHandle[OutputType]. The handle provides access to the running workflow.
Nexus Python SDK - operation handler reliability
Handlers should be reliable since the circuit breaker trips after 5 consecutive retryable errors (for example: worker timeouts), blocking all Operations from the caller to that Endpoint. Synchronous operations should only be used when execution is highly reliable, has predictably low latency, and finishes within 10-second handler deadline. Asynchronous operations should be used when latency or availability is uncertain or work exceeds handler deadline.
Nexus Python SDK - nexus.client and nexus.info
Use nexus.client() to get the Client that the Worker was initialized with from within a Nexus operation handler. Use nexus.info() to access information about the currently-executing Nexus Operation including its Task Queue.
Implement Nexus Operation handler with @nexus.workflow_run_operation decorator
Create an Operation handler class decorated with @nexusrpc.handler.service_handler(service=ServiceClass). Use @nexus.workflow_run_operation decorator on async handler methods that take nexus.WorkflowRunOperationContext and input parameters. The handler can start a Workflow using ctx.start_workflow(WorkflowClass.run, args, id=workflow_id). The handler receives Nexus requests and bridges the service contract to the actual Workflow execution.
Nexus Operation context parameter types
Nexus Operation handlers receive nexus.WorkflowRunOperationContext as the first parameter after self. This context provides methods like start_workflow() to initiate Workflows from the Operation handler.
Implement Nexus Operation handler with WorkflowRunOperationHandler
Use nexus.serviceHandler() to create a handler for a Nexus Service. Use temporalNexus.WorkflowRunOperationHandler<InputType, OutputType> to create an asynchronous Nexus Operation that starts a Workflow. The handler receives context and input, and uses temporalNexus.startWorkflow() to start the workflow with specified workflowId and task queue.
Register Nexus Service handler in Worker
Pass the Nexus Service Handler to the Worker via the nexusServices parameter when creating the Worker. A Worker will only poll for and process incoming Nexus requests if Nexus Service Handlers are registered.
Nexus TypeScript quickstart example: complete handler implementation
import { randomUUID } from 'crypto';
import * as nexus from 'nexus-rpc';
import * as temporalNexus from '@temporalio/nexus';
import { sayHelloService, MyInput } from './service';
import { example } from './workflows';
export const sayHelloHandler = nexus.serviceHandler(sayHelloService, {
sayHello: new temporalNexus.WorkflowRunOperationHandler<MyInput, string>(
async (ctx, input: MyInput) => {
return await temporalNexus.startWorkflow(ctx, example, {
args: [input.name],
workflowId: "say-hello-nexus-" + randomUUID(),
});
},
),
});
Nexus TypeScript quickstart example: register handler in Worker
import { NativeConnection, Worker } from '@temporalio/worker';
import * as activities from './activities';
import { sayHelloHandler } from './handler';
async function run() {
const connection = await NativeConnection.connect({
address: 'localhost:7233',
});
try {
const worker = await Worker.create({
connection,
namespace: 'default',
taskQueue: 'hello-world',
workflowsPath: require.resolve('./workflows'),
activities,
nexusServices: [sayHelloHandler],
});
await worker.run();
} finally {
await connection.close();
}
}
run().catch((err) => {
console.error(err);
process.exit(1);
});