Middleware & Dependency Injection

Middleware lets you run logic before and after every task on a client, without touching individual task definitions. Common uses include injecting request IDs, enriching inputs with shared data, encrypting/decrypting payloads, and normalizing or augmenting outputs.

⚠️

This feature is experimental, and middleware hook signatures may change in future releases.

Hatchet’s Python SDK uses FastAPI-style dependency injection to run logic before tasks and inject the results as parameters. Dependencies are declared as functions and wired into tasks with Depends.

Middleware hooks are registered on the client with withMiddleware and are fully type-safe — TypeScript sees the union of fields from the task input type and any values returned by before hooks, and similarly for task outputs and after hooks.

The Ruby SDK is in early access, and may change. We'd love your feedback!

In Ruby, this pattern uses callable objects (lambdas/procs) passed as deps when defining tasks. Dependencies are evaluated before each task run and made available via ctx.deps.

Defining Middleware

Define your dependency functions — they receive the workflow input and context, and their return values are injected into the task as parameters.

examples/python/dependency_injection/worker.py

async def async_dep(input: EmptyModel, ctx: Context) -> str:
    return ASYNC_DEPENDENCY_VALUE


def sync_dep(input: EmptyModel, ctx: Context) -> str:
    return SYNC_DEPENDENCY_VALUE


@asynccontextmanager
async def async_cm_dep(
    input: EmptyModel, ctx: Context, async_dep: Annotated[str, Depends(async_dep)]
) -> AsyncGenerator[str, None]:
    try:
        yield ASYNC_CM_DEPENDENCY_VALUE + "_" + async_dep
    finally:
        pass


@contextmanager
def sync_cm_dep(
    input: EmptyModel, ctx: Context, sync_dep: Annotated[str, Depends(sync_dep)]
) -> Generator[str, None, None]:
    try:
        yield SYNC_CM_DEPENDENCY_VALUE + "_" + sync_dep
    finally:
        pass


@contextmanager
def base_cm_dep(input: EmptyModel, ctx: Context) -> Generator[str, None, None]:
    try:
        yield CHAINED_CM_VALUE
    finally:
        pass


def chained_dep(
    input: EmptyModel, ctx: Context, base_cm: Annotated[str, Depends(base_cm_dep)]
) -> str:
    return "chained_" + base_cm


@asynccontextmanager
async def base_async_cm_dep(
    input: EmptyModel, ctx: Context
) -> AsyncGenerator[str, None]:
    try:
        yield CHAINED_ASYNC_CM_VALUE
    finally:
        pass


async def chained_async_dep(
    input: EmptyModel,
    ctx: Context,
    base_async_cm: Annotated[str, Depends(base_async_cm_dep)],
) -> str:
    return "chained_" + base_async_cm

Create a client and attach middleware with before and after hooks.

before(input, ctx) runs before the task. Its return value replaces the task input.
after(output, ctx, input) runs after the task. Its return value replaces the task output.

examples/typescript/middleware/client.ts

import { HatchetClient, HatchetMiddleware } from '@hatchet/v1';

export type GlobalInputType = {
  first: number;
  second: number;
};

export type GlobalOutputType = {
  extra: number;
};

const myMiddleware = {
  before: (input, ctx) => {
    console.log('before', input.first);
    return { ...input, dependency: 'abc-123' };
  },
  after: (output, ctx, input) => {
    return { ...output, additionalData: 2 };
  },
} satisfies HatchetMiddleware<GlobalInputType, GlobalOutputType>;

export const hatchetWithMiddleware = HatchetClient.init<
  GlobalInputType,
  GlobalOutputType
>().withMiddleware(myMiddleware);

⚠️

Spread the original value if you want to keep it. The return value of each hook replaces the input (or output) entirely — it does not shallow-merge. If you omit ...input in a before hook, the original fields are lost. The same applies to ...output in an after hook.

// ✅ Keeps original fields and adds `requestId`
before: (input) => ({ ...input, requestId: crypto.randomUUID() })
 
// ❌ Replaces input entirely — task only receives { requestId }
before: (input) => ({ requestId: crypto.randomUUID() })

Chaining Middleware

You can chain multiple .withMiddleware() calls to run hooks in sequence. Each before hook receives the return value of the previous before hook (or the original input for the first hook), and each after hook receives the return value of the previous after hook.

examples/typescript/middleware/client.ts

const firstMiddleware = {
  before: (input, ctx) => {
    console.log('before', input.first);
    return { ...input, dependency: 'abc-123' };
  },
  after: (output, ctx, input) => {
    return { ...output, firstExtra: 3 };
  },
} satisfies HatchetMiddleware<GlobalInputType>;

const secondMiddleware = {
  before: (input, ctx) => {
    console.log('before', input.dependency); // available from previous middleware
    return { ...input, anotherDep: true };
  },
  after: (output, ctx, input) => {
    return { ...output, secondExtra: 4 };
  },
} satisfies HatchetMiddleware<GlobalInputType & { dependency: string }>;

export const hatchetWithMiddlewareChaining = HatchetClient.init<GlobalInputType>()
  .withMiddleware(firstMiddleware)
  .withMiddleware(secondMiddleware);

The Ruby SDK is in early access, and may change. We'd love your feedback!

Define your dependencies as callable objects (lambdas). They receive the input, context, and optionally a hash of previously resolved dependencies for chaining.

examples/ruby/dependency_injection/worker.rb

sync_dep = ->(_input, _ctx) { SYNC_DEPENDENCY_VALUE }
async_dep = ->(_input, _ctx) { ASYNC_DEPENDENCY_VALUE }

sync_cm_dep = lambda { |_input, _ctx, deps|
  "#{SYNC_CM_DEPENDENCY_VALUE}_#{deps[:sync_dep]}"
}

async_cm_dep = lambda { |_input, _ctx, deps|
  "#{ASYNC_CM_DEPENDENCY_VALUE}_#{deps[:async_dep]}"
}

chained_dep = ->(_input, _ctx, deps) { "chained_#{CHAINED_CM_VALUE}" }
chained_async_dep = ->(_input, _ctx, deps) { "chained_#{CHAINED_ASYNC_CM_VALUE}" }

How Middleware Executes

Dependencies are resolved before each task execution. Each dependency function receives the original workflow input and the task context, and its return value is injected as a named parameter to the task function.

When a task runs, the worker applies middleware hooks in this order:

Before hooks run in registration order

Each before hook receives the current input and the task Context. Its return value replaces the input for the next hook (or the task itself). Returning undefined (or void) skips replacement and passes the input through unchanged.

The task function executes

The task receives the final input after all before hooks have run.

After hooks run in registration order

Each after hook receives the current output, the task Context, and the final input. Its return value replaces the output for the next hook (or the final result). Returning undefined skips replacement.

Both before and after hooks can be async — return a Promise and it will be awaited before proceeding.

If a middleware hook throws an error, the task run fails with that error. There is no built-in error recovery within middleware — use try/catch inside your hooks if you need graceful fallback.

The `ctx` Parameter

The second parameter of both before and after hooks is the task Context object. This gives middleware access to:

ctx.workflowRunId — the ID of the current workflow run
ctx.stepRunId — the ID of the current step run
ctx.log() — emit structured logs visible in the Hatchet dashboard
ctx.cancel() — cancel the current run from within middleware

Global Types vs Middleware Types

There are two ways extra fields end up on a task’s input:

Mechanism	Set via	Required at call site?	Available at runtime?
Global input type	`HatchetClient.init<T>()`	Yes — callers must provide these fields	Yes
Middleware before hook	`.withMiddleware({ before })`	No — injected automatically by the worker	Yes

Global input types (T in init<T>()) represent fields that callers must supply when triggering a task. This is useful when you know every task must always receive certain parameters — for example, a userId for authentication or a tenantId for multi-tenant routing. By declaring these as the global type, TypeScript enforces that every caller provides them.

Middleware before hooks, on the other hand, inject fields that are computed at runtime (e.g. request IDs, decrypted secrets, fetched config) and are not required from callers.

type RequiredContext = { userId: string; orgId: string };
 
const client = HatchetClient.init<RequiredContext>()
  .withMiddleware({
    before: (input) => ({
      ...input,
      resolvedAt: Date.now(),        // injected, not required from caller
      permissions: lookupPerms(input.userId), // derived from global type
    }),
  });
 
// Callers MUST provide userId and orgId — TypeScript enforces this
await myTask.run({ userId: 'usr_123', orgId: 'org_456', /* ...task fields */ });

The Ruby SDK is in early access, and may change. We'd love your feedback!

Dependencies are resolved in the order they are declared in the deps hash. Each dependency function can optionally receive already-resolved dependencies as its third argument, enabling chaining.

Using Middleware in Tasks

Inject dependencies into your tasks using Depends and type annotations. The dependency results are passed directly as function parameters.

examples/python/dependency_injection/worker.py

@hatchet.task()
async def async_task_with_dependencies(
    _i: EmptyModel,
    ctx: Context,
    async_dep: Annotated[str, Depends(async_dep)],
    sync_dep: Annotated[str, Depends(sync_dep)],
    async_cm_dep: Annotated[str, Depends(async_cm_dep)],
    sync_cm_dep: Annotated[str, Depends(sync_cm_dep)],
    chained_dep: Annotated[str, Depends(chained_dep)],
    chained_async_dep: Annotated[str, Depends(chained_async_dep)],
) -> Output:
    return Output(
        sync_dep=sync_dep,
        async_dep=async_dep,
        async_cm_dep=async_cm_dep,
        sync_cm_dep=sync_cm_dep,
        chained_dep=chained_dep,
        chained_async_dep=chained_async_dep,
    )

⚠️

Your dependency functions must take two positional arguments: the workflow input and the Context (the same as any other task).

Tasks created from a middleware-enabled client automatically receive the merged input and output types. There is no extra configuration needed on the task itself.

examples/typescript/middleware/workflow.ts

import { hatchetWithMiddleware } from './client';

type TaskInput = {
  message: string;
};

type TaskOutput = {
  message: string;
};

export const taskWithMiddleware = hatchetWithMiddleware.task<TaskInput, TaskOutput>({
  name: 'task-with-middleware',
  fn: (input, _ctx) => {
    console.log('task', input.message); // string  (from TaskInput)
    console.log('task', input.first); // number  (from GlobalInputType)
    console.log('task', input.second); // number  (from GlobalInputType)
    console.log('task', input.dependency); // string  (from Pre Middleware)
    return {
      message: input.message,
      extra: 1,
    };
  },
});
// !!

The task’s input type is the intersection of TaskInput, GlobalInputType, and the return type of the before middleware hook. The task’s return type must satisfy TaskOutput and GlobalOutputType, while the caller receives the intersection of those with the after middleware return type.

The Ruby SDK is in early access, and may change. We'd love your feedback!

Pass a deps hash when defining a task. The resolved dependency values are available inside the task block via ctx.deps.

examples/ruby/dependency_injection/worker.rb

ASYNC_TASK_WITH_DEPS = HATCHET.task(
  name: "async_task_with_dependencies",
  deps: {
    sync_dep: sync_dep,
    async_dep: async_dep,
    sync_cm_dep: sync_cm_dep,
    async_cm_dep: async_cm_dep,
    chained_dep: chained_dep,
    chained_async_dep: chained_async_dep
  }
) do |input, ctx|
  {
    "sync_dep" => ctx.deps[:sync_dep],
    "async_dep" => ctx.deps[:async_dep],
    "async_cm_dep" => ctx.deps[:async_cm_dep],
    "sync_cm_dep" => ctx.deps[:sync_cm_dep],
    "chained_dep" => ctx.deps[:chained_dep],
    "chained_async_dep" => ctx.deps[:chained_async_dep]
  }
end

SYNC_TASK_WITH_DEPS = HATCHET.task(
  name: "sync_task_with_dependencies",
  deps: {
    sync_dep: sync_dep,
    async_dep: async_dep,
    sync_cm_dep: sync_cm_dep,
    async_cm_dep: async_cm_dep,
    chained_dep: chained_dep,
    chained_async_dep: chained_async_dep
  }
) do |input, ctx|
  {
    "sync_dep" => ctx.deps[:sync_dep],
    "async_dep" => ctx.deps[:async_dep],
    "async_cm_dep" => ctx.deps[:async_cm_dep],
    "sync_cm_dep" => ctx.deps[:sync_cm_dep],
    "chained_dep" => ctx.deps[:chained_dep],
    "chained_async_dep" => ctx.deps[:chained_async_dep]
  }
end

DURABLE_ASYNC_TASK_WITH_DEPS = HATCHET.durable_task(
  name: "durable_async_task_with_dependencies",
  deps: {
    sync_dep: sync_dep,
    async_dep: async_dep,
    sync_cm_dep: sync_cm_dep,
    async_cm_dep: async_cm_dep,
    chained_dep: chained_dep,
    chained_async_dep: chained_async_dep
  }
) do |input, ctx|
  {
    "sync_dep" => ctx.deps[:sync_dep],
    "async_dep" => ctx.deps[:async_dep],
    "async_cm_dep" => ctx.deps[:async_cm_dep],
    "sync_cm_dep" => ctx.deps[:sync_cm_dep],
    "chained_dep" => ctx.deps[:chained_dep],
    "chained_async_dep" => ctx.deps[:chained_async_dep]
  }
end

DURABLE_SYNC_TASK_WITH_DEPS = HATCHET.durable_task(
  name: "durable_sync_task_with_dependencies",
  deps: {
    sync_dep: sync_dep,
    async_dep: async_dep,
    sync_cm_dep: sync_cm_dep,
    async_cm_dep: async_cm_dep,
    chained_dep: chained_dep,
    chained_async_dep: chained_async_dep
  }
) do |input, ctx|
  {
    "sync_dep" => ctx.deps[:sync_dep],
    "async_dep" => ctx.deps[:async_dep],
    "async_cm_dep" => ctx.deps[:async_cm_dep],
    "sync_cm_dep" => ctx.deps[:sync_cm_dep],
    "chained_dep" => ctx.deps[:chained_dep],
    "chained_async_dep" => ctx.deps[:chained_async_dep]
  }
end

DI_WORKFLOW = HATCHET.workflow(name: "dependency-injection-workflow")

# Workflow tasks with dependencies follow the same pattern
DI_WORKFLOW.task(:wf_task_with_dependencies) do |input, ctx|
  {
    "sync_dep" => SYNC_DEPENDENCY_VALUE,
    "async_dep" => ASYNC_DEPENDENCY_VALUE
  }
end

Running a Worker

No special worker configuration is needed — dependencies are evaluated automatically each time a task runs.

Workers are created from the same middleware-enabled client. No special setup is required — the middleware hooks are applied automatically when tasks execute.

examples/typescript/middleware/worker.ts

import { hatchetWithMiddleware } from './client';
import { taskWithMiddleware } from './workflow';

async function main() {
  const worker = await hatchetWithMiddleware.worker('task-with-middleware', {
    workflows: [taskWithMiddleware],
  });

  await worker.start();
}

if (require.main === module) {
  main();
}

Practical Examples

The examples below show TypeScript middleware for common production patterns. Each can be adapted to the Python dependency injection model by extracting the same logic into a dependency function.

End-to-End Encryption

Encrypt sensitive input fields before they reach the Hatchet server, and decrypt the output on the way back. This ensures plaintext data never leaves your worker process.

examples/typescript/middleware/recipes.ts

import { HatchetClient, HatchetMiddleware } from '@hatchet/v1';
import { randomUUID, createCipheriv, createDecipheriv, randomBytes } from 'crypto';

The before hook decrypts incoming data so your task function works with plaintext. The after hook encrypts the output before it is stored. The encryption key never leaves the worker environment.

Offloading Large Payloads to S3

When task inputs or outputs exceed Hatchet’s payload size limit (or you simply want to keep large blobs out of the control plane), upload them to S3 and pass a signed URL instead.

examples/typescript/middleware/recipes.ts

import { S3Client, PutObjectCommand, GetObjectCommand } from '@aws-sdk/client-s3';
import { getSignedUrl } from '@aws-sdk/s3-request-presigner';

const ALGORITHM = 'aes-256-gcm';
const KEY = Buffer.from(process.env.ENCRYPTION_KEY!, 'hex');

type EncryptedEnvelope = { ciphertext: string; iv: string; tag: string };

function encrypt(plaintext: string): EncryptedEnvelope {
  const iv = randomBytes(16);
  const cipher = createCipheriv(ALGORITHM, KEY, iv);
  const encrypted = Buffer.concat([cipher.update(plaintext, 'utf8'), cipher.final()]);
  return {
    ciphertext: encrypted.toString('base64'),
    iv: iv.toString('base64'),
    tag: cipher.getAuthTag().toString('base64'),
  };
}

function decrypt(ciphertext: string, iv: string, tag: string): string {
  const decipher = createDecipheriv(ALGORITHM, KEY, Buffer.from(iv, 'base64'));
  decipher.setAuthTag(Buffer.from(tag, 'base64'));
  return decipher.update(ciphertext, 'base64', 'utf8') + decipher.final('utf8');
}

type EncryptedInput = { encrypted?: EncryptedEnvelope };

const e2eEncryption: HatchetMiddleware<EncryptedInput> = {
  before: (input) => {
    if (!input.encrypted) return input;
    const { ciphertext, iv, tag } = input.encrypted;
    const decrypted = JSON.parse(decrypt(ciphertext, iv, tag));
    return { ...input, ...decrypted, encrypted: undefined };
  },
  after: (output) => {
    const payload = JSON.stringify(output);
    return { encrypted: encrypt(payload) };
  },
};

const encryptionClient = HatchetClient.init<EncryptedInput>().withMiddleware(e2eEncryption);

const s3 = new S3Client({ region: process.env.AWS_REGION });
const BUCKET = process.env.S3_BUCKET!;
const PAYLOAD_THRESHOLD = 256 * 1024; // 256 KB

async function uploadToS3(data: unknown): Promise<string> {
  const key = `hatchet-payloads/${randomUUID()}.json`;
  await s3.send(
    new PutObjectCommand({
      Bucket: BUCKET,
      Key: key,
      Body: JSON.stringify(data),
      ContentType: 'application/json',
    })
  );
  return getSignedUrl(s3, new GetObjectCommand({ Bucket: BUCKET, Key: key }), {
    expiresIn: 3600,
  });
}

async function downloadFromS3(url: string): Promise<unknown> {
  const res = await fetch(url);
  return res.json();
}

type S3Input = { s3Url?: string };

const s3Offload: HatchetMiddleware<S3Input> = {
  before: async (input) => {
    if (input.s3Url) {
      const restored = (await downloadFromS3(input.s3Url)) as Record<string, any>;
      return { ...restored, s3Url: undefined };
    }
    return input;
  },
  after: async (output) => {
    const serialized = JSON.stringify(output);
    if (serialized.length > PAYLOAD_THRESHOLD) {
      const url = await uploadToS3(output);
      return { s3Url: url };
    }
    return output;
  },
};

const s3Client = HatchetClient.init<S3Input>().withMiddleware(s3Offload);

⚠️

The caller is responsible for uploading oversized inputs to S3 before triggering the task. The before hook only handles the download side. You can use the same uploadToS3 helper on the caller side to upload the input and pass { __s3Url: url } as the task input.

FAQ

What is Hatchet middleware and how does it differ from Express middleware?

Hatchet middleware runs inside the worker process around each task invocation — not on an HTTP request path. A before hook transforms input before the task runs, and an after hook transforms output after. Unlike Express middleware, there is no next() function; hooks return their result directly and the runner chains them automatically.

Can I use middleware with both tasks and workflows?

Yes. Middleware is registered on the HatchetClient instance, so it applies to every task created from that client — whether the task is a standalone client.task() or part of a multi-step client.workflow(). Each step in a workflow will have middleware applied independently.

Does middleware run on the server or on the worker?

Middleware runs entirely on the worker. The Hatchet server never sees or executes your middleware code. This is what makes patterns like end-to-end encryption possible — plaintext data stays within your infrastructure.

What happens if my middleware throws an error?

If a before or after hook throws (or returns a rejected Promise), the task run fails with that error. There is no automatic retry of middleware itself, but the task’s configured retry policy will still apply, re-running the task (and its middleware) from scratch.

Can I use async/await in middleware hooks?

Yes. Both before and after hooks can be synchronous or asynchronous. If a hook returns a Promise, the worker will await it before proceeding to the next hook or the task function.

How do I share state between `before` and `after` hooks?

The after hook receives the task input (after before hooks have run) as its third argument. Add fields in before (e.g. startedAt, traceId) and read them from input in after. There is no separate shared context object — the input itself is the carrier.

Does middleware apply to child tasks spawned via fanout?

Middleware is scoped to the client instance. If a child task is defined on the same middleware-enabled client, its middleware will run when that child task executes. If the child task uses a different client instance, only that client’s middleware (if any) applies.

Can I selectively skip middleware for certain tasks?

Middleware applies to all tasks on a given client. To skip middleware for specific tasks, create a second client without middleware and define those tasks on it. This is a deliberate design choice — middleware is a cross-cutting concern, and selective opt-out is handled at the client boundary.

Is there a performance overhead to using middleware?

Middleware hooks are plain JavaScript functions that run in-process on the worker. The overhead is the execution time of your hook code. For lightweight operations (adding a field, logging), the overhead is negligible. For heavier operations (network calls like S3 uploads or decryption), the task’s total duration will include that time, so keep hooks as efficient as possible.

What is the difference between global types and middleware types in TypeScript?

Global types (HatchetClient.init<GlobalInput, GlobalOutput>()) define fields that callers must provide when triggering a task. Middleware types (inferred from withMiddleware return values) define fields that are injected at runtime by the worker. Both end up on the task’s input type, but only global types appear in the caller-facing run() signature.

Can I use middleware for rate limiting or authentication?

Yes. A before hook can check rate limits, validate API keys, or verify JWTs before the task runs. If the check fails, throw an error to abort the task. However, for rate limiting specifically, consider using Hatchet’s built-in rate limiting feature, which operates at the scheduling layer and is more efficient than in-worker checks.

How do I test middleware in isolation?

Middleware hooks are plain functions — you can unit-test them directly by calling them with mock input and a mock context object. For integration tests, the e2e test pattern of creating a client, attaching middleware, defining a task, starting a worker, and asserting on the result works well. See the middleware example on GitHub for a complete test setup.

Streaming SDK Improvements

We use cookies

Middleware & Dependency Injection

Defining Middleware

Chaining Middleware

How Middleware Executes

Before hooks run in registration order

The task function executes

After hooks run in registration order

The ctx Parameter

Global Types vs Middleware Types

Using Middleware in Tasks

Running a Worker

Practical Examples

End-to-End Encryption

Offloading Large Payloads to S3

FAQ

What is Hatchet middleware and how does it differ from Express middleware?

Can I use middleware with both tasks and workflows?

Does middleware run on the server or on the worker?

What happens if my middleware throws an error?

Can I use async/await in middleware hooks?

How do I share state between before and after hooks?

Does middleware apply to child tasks spawned via fanout?

Can I selectively skip middleware for certain tasks?

Is there a performance overhead to using middleware?

What is the difference between global types and middleware types in TypeScript?

Can I use middleware for rate limiting or authentication?

How do I test middleware in isolation?

The `ctx` Parameter

How do I share state between `before` and `after` hooks?