Great question. In Prefect 3.x there isn’t a special DI container — you share dependencies using normal Python patterns, with a few Prefect-native helpers. The “right” pattern depends on whether your tasks run in the same process (default) or on distributed workers. Quick guidance - Same process (default ConcurrentTaskRunner): you can share an actual object (e.g., client/connection) created at flow scope and used by tasks. - Distributed/multi-process: share configuration/credentials, then create the object inside each task (optionally with per-process caching). Don’t try to pass non-serializable clients across process boundaries. Patterns and examples 1) Same process: create once at flow scope and use in tasks - Easiest way when all tasks run in the same process. Works with threads; ensure the client is thread-safe.

from prefect import flow, task
import httpx

@flow
def my_flow():
    client = httpx.Client(base_url="<https://api.example.com>")

    @task
    def get_users():
        return client.get("/users").json()

    @task
    def get_posts():
        return client.get("/posts").json()

    try:
        users = get_users()
        posts = get_posts()
        return {"users": users, "posts": posts}
    finally:
        client.close()

Alternative: pass the object explicitly as a parameter to tasks (works in same process only if the object is serializable; most clients are not). 2) Distributed/multi-process: share config, build per task with a cached factory - Persist config/creds with a Block, then create clients inside tasks. Use a module-level LRU cache so each worker/process builds the client once.

from functools import lru_cache
from prefect import flow, task
from prefect.blocks.core import Block
from pydantic import SecretStr
import httpx

class MyAPI(Block):
    base_url: str
    api_key: SecretStr

@lru_cache(maxsize=None)
def get_client(base_url: str, api_key: str):
    return httpx.Client(base_url=base_url, headers={"Authorization": f"Bearer {api_key}"})

@task
def call_api(block_name: str, path: str):
    cfg = MyAPI.load(block_name)
    client = get_client(cfg.base_url, cfg.api_key.get_secret_value())
    return client.get(path).json()

@flow
def etl(api_block="my-api"):
    users = call_api(api_block, "/users")
    posts = call_api(api_block, "/posts")
    return {"users": users, "posts": posts}

- Each worker process will build the client once and reuse it for all tasks that hit the same key. Docs: Blocks concepts, Create custom blocks 3) Cache expensive setup results (not the client object itself) - Use task caching for serializable, reusable artifacts (e.g., auth tokens, schemas, metadata, connection strings).

from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import INPUTS

@task(cache_policy=INPUTS, cache_expiration=timedelta(hours=1), persist_result=True)
def get_token(tenant: str) -> str:
    # make an auth call
    return "token-for-" + tenant

@task
def work(tenant: str):
    token = get_token(tenant)  # cached across runs
    # build a client per task using token, then do work
    return "ok"

@flow
def main(tenant="prod"):
    return work(tenant)

Docs: Caching concepts, How to cache workflow steps, Advanced caching 4) Same process: ContextVar for thread-safe access - Useful if you don’t want to plumb dependencies through every function. Not for distributed scenarios. ``` from contextvars import ContextVar from prefect import flow, task import sqlite3 db_var: ContextVar[sqlite3.Connection | None] = ContextVar("db", default=None) @task def use_db(): conn = db_var.get() cur = conn.cursor() cur.execute("select 1") return cur.fetchone() @flow def f():

conn = sqlite3.connect("memory") token = db_var.set(conn) try: return use_db() finally: db_var.reset(token) conn.close()

5) Manage lifecycle with a context manager
- Especially helpful for connections that must be closed, while still sharing within a flow run.

from contextlib import contextmanager from prefect import flow, task @contextmanager def client_ctx(): c = make_client() try: yield c finally: c.close() @task def t1(c): ... @task def t2(c): ... @flow def f(): with client_ctx() as c: a = t1(c) b = t2(c) return a, b ``` When to use what - Same-process only: create at flow scope and share via closure/args or ContextVar. - Distributed or remote workers: store config/creds in a Block, then build clients inside the task. Use a module-level LRU cache to avoid rebuilding per task within the same worker. - Expensive setup that’s serializable: cache the setup result with task caching and pass it into downstream tasks. Related docs - Blocks: Concepts, Custom blocks - Caching: Concepts, How-to, Advanced - Flows/Tasks: Flows, Tasks A couple of gotchas - Don’t try to cache or pass non-serializable client objects across processes; cache only serializable data (tokens, config). - Ensure thread-safety if you share a single client across concurrently running tasks. - Prefer building clients close to where they’re used, and rely on pooling/LRU for performance. If you can share a bit about your execution model (same machine vs. distributed workers) and what dependency you want to share (DB engine, HTTP client, SDK, etc.), I can tailor an optimal pattern and snippet for your setup.

Got it — you’re injecting

Reader

and

Datastore

interfaces and want to run steps like read → store → mark as Prefect tasks while sharing those dependencies. Here are safe, idiomatic ways to adapt that pattern in Prefect 3.x. Option A — Same process: create once at flow scope and use in tasks (closure) - Best if you’re using the default task runner (same process, threaded). You create your concrete `Reader`/`Datastore` once and share them across tasks via closures.

from typing import Protocol, Sequence
from prefect import flow, task

# Interfaces
class Reader(Protocol):
    def read(self) -> list[bytes]: ...
    def mark_emails_as_read(self) -> None: ...

class Datastore(Protocol):
    def store_file(self, data: list[bytes]) -> None: ...

class EmailTask:
    def __init__(self, reader: Reader, datastore: Datastore):
        self.reader = reader
        self.datastore = datastore

    def read_mails(self) -> list[bytes]:
        return self.reader.read()

    def store_mails(self, mails: list[bytes]) -> None:
        self.datastore.store_file(mails)

    def mark_emails_as_read(self) -> None:
        self.reader.mark_emails_as_read()

@flow
def email_flow(reader: Reader, datastore: Datastore):
    svc = EmailTask(reader, datastore)

    @task
    def read():
        return svc.read_mails()

    @task
    def store(mails: list[bytes]):
        svc.store_mails(mails)

    @task
    def mark():
        svc.mark_emails_as_read()

    mails = read()
    store(mails)
    mark()

Why this works - Tasks run in the same process by default, so the closure safely shares the same objects. - No pickling/serialization of `reader`/`datastore` required. - Be sure your concrete client implementations are thread‑safe if tasks run concurrently. Option B — Distributed/multi-process: build per-task, reuse with LRU cache - If tasks might run in different processes/containers, don’t pass live clients. Persist only config/credentials (ideally with a Block), then create the client in each process with an LRU cache so it’s built once per worker.

from functools import lru_cache
from typing import Protocol
from prefect import flow, task
from prefect.blocks.core import Block
from pydantic import SecretStr

# Interfaces
class Reader(Protocol):
    def read(self) -> list[bytes]: ...
    def mark_emails_as_read(self) -> None: ...

class Datastore(Protocol):
    def store_file(self, data: list[bytes]) -> None: ...

# Store only config, not live clients
class EmailConfig(Block):
    imap_host: str
    imap_user: str
    imap_password: SecretStr
    bucket: str

# Factories (keyed by primitives -> one client per process/worker)
@lru_cache(maxsize=None)
def get_reader(host: str, user: str, password: str) -> Reader:
    # return a concrete Reader, e.g., ImapReader(host, user, password)
    ...

@lru_cache(maxsize=None)
def get_datastore(bucket: str) -> Datastore:
    # return a concrete Datastore, e.g., S3Datastore(bucket)
    ...

@task
def read_mails(cfg_name: str) -> list[bytes]:
    cfg = EmailConfig.load(cfg_name)
    reader = get_reader(cfg.imap_host, cfg.imap_user, cfg.imap_password.get_secret_value())
    return reader.read()

@task
def store_mails(cfg_name: str, mails: list[bytes]) -> None:
    cfg = EmailConfig.load(cfg_name)
    store = get_datastore(cfg.bucket)
    store.store_file(mails)

@task
def mark_emails_as_read(cfg_name: str) -> None:
    cfg = EmailConfig.load(cfg_name)
    reader = get_reader(cfg.imap_host, cfg.imap_user, cfg.imap_password.get_secret_value())
    reader.mark_emails_as_read()

@flow
def email_flow(cfg_block_name: str = "email-config"):
    mails = read_mails(cfg_block_name)
    store_mails(cfg_block_name, mails)
    mark_emails_as_read(cfg_block_name)

Why this works - Only serializable config crosses process boundaries. - Each worker process creates and then reuses its own clients via

@lru_cache

. Optional: cache expensive but serializable setup

- If you have heavy but serializable setup (e.g., auth token), cache it in a task and pass to downstream tasks.

from datetime import timedelta
from prefect import task
from prefect.cache_policies import INPUTS

@task(persist_result=True, cache_policy=INPUTS, cache_expiration=timedelta(hours=1))
def get_auth_token(user: str) -> str:
    # slow auth request
    return "token..."

A couple of practical tips - Prefer functions for tasks; bound methods can be harder to serialize/reuse across modules. - If you do decorate class methods as tasks, the instance (

self

) must be picklable to run out-of-process. That’s usually not true for live clients. - If tasks run concurrently, confirm your `Reader`/`Datastore` implementations are thread‑safe; otherwise use locking or create separate instances per task. Docs for deeper dives - Blocks: Concepts, Custom blocks - Caching: Concepts, How-to, Advanced - Flows and Tasks: Flows, Tasks Want me to tailor this to your concrete `Reader`/`Datastore` (e.g., IMAP + S3, Gmail API + GCS)? Also, are you sticking with the default runner or planning to distribute tasks to separate workers? That determines whether Option A or B is the better fit.

Short answer: Yes, you can decorate the methods of your EmailTask as Prefect tasks — Prefect 3.x explicitly supports instance, class, and static methods as tasks. What to know before you do it - Local vs distributed: If tasks run in the same process (default), binding instance methods works great. If tasks may run on remote workers or separate processes, Prefect will serialize the whole instance. That means: - Your instance must be cloudpickle-serializable. - Avoid open sockets/DB connections/file handles on the instance. - Keep instance state small and importable (not defined in main). - Concurrency: If multiple tasks use the same instance concurrently, ensure the underlying clients are thread-safe or guard access. - Caching/names: Prefect exposes self to cache keys and task_run_name templates, so different instances naturally get separate caches and you can reference self in names. Patterns A) Instance methods as tasks (same-process friendly)

from prefect import task, flow

class EmailTask:
    def __init__(self, reader, datastore):
        self.reader = reader
        self.datastore = datastore

    @task(task_run_name="read-mails-{self.__class__.__name__}")
    def read_mails(self) -> list[bytes]:
        return self.reader.read()

    @task
    def store_mails(self, mails: list[bytes]) -> None:
        self.datastore.store_file(mails)

    @task
    def mark_emails_as_read(self) -> None:
        self.reader.mark_emails_as_read()

@flow
def email_flow(reader, datastore):
    svc = EmailTask(reader, datastore)
    mails = svc.read_mails()
    svc.store_mails(mails)
    svc.mark_emails_as_read()

B) Safer for distributed workers: static/class methods or free functions - Keep tasks stateless and pass only primitives/config. Build clients inside tasks (with LRU cache) or via Blocks.

from functools import lru_cache
from prefect import task, flow
from prefect.blocks.core import Block
from pydantic import SecretStr

class EmailConfig(Block):
    imap_host: str
    user: str
    password: SecretStr
    bucket: str

@lru_cache(maxsize=None)
def get_reader(host, user, password):
    # return concrete Reader
    ...

@lru_cache(maxsize=None)
def get_store(bucket):
    # return concrete Datastore
    ...

class EmailTasks:
    @staticmethod
    @task
    def read(cfg_name: str) -> list[bytes]:
        cfg = EmailConfig.load(cfg_name)
        r = get_reader(cfg.imap_host, cfg.user, cfg.password.get_secret_value())
        return r.read()

    @staticmethod
    @task
    def store(cfg_name: str, mails: list[bytes]) -> None:
        cfg = EmailConfig.load(cfg_name)
        s = get_store(cfg.bucket)
        s.store_file(mails)

    @staticmethod
    @task
    def mark_read(cfg_name: str) -> None:
        cfg = EmailConfig.load(cfg_name)
        r = get_reader(cfg.imap_host, cfg.user, cfg.password.get_secret_value())
        r.mark_emails_as_read()

@flow
def email_flow(cfg_block="email-config"):
    mails = EmailTasks.read(cfg_block)
    EmailTasks.store(cfg_block, mails)
    EmailTasks.mark_read(cfg_block)

C) Hybrid: instance methods but pass only serializable state - If you like the OO style, construct the service with small, serializable config (not live clients), and build clients within the methods.

class EmailService:
    def __init__(self, host: str, user: str, password: str, bucket: str):
        self.host, self.user, self.password, self.bucket = host, user, password, bucket

    def _reader(self):
        return get_reader(self.host, self.user, self.password)

    def _store(self):
        return get_store(self.bucket)

    @task
    def read(self) -> list[bytes]:
        return self._reader().read()

    @task
    def store(self, mails: list[bytes]) -> None:
        self._store().store_file(mails)

    @task
    def mark(self) -> None:
        self._reader().mark_emails_as_read()

Rules of thumb - OK to make methods tasks if you run in one process and your instance is small/serializable.

- Prefer static/class methods or free functions for portability to workers. - Put only config/credentials in the instance; create clients per-process with an LRU cache. - If you need shared clients in one process, create them at flow scope and call inner tasks that close over them. If you share what your Reader/Datastore concretely are (e.g., IMAP + S3) and whether you plan remote workers, I can tailor which pattern is safest and provide a ready-to-run snippet.

You can make class methods Prefect tasks — that works at runtime — but your error is from Pyright’s type checker not understanding Prefect’s descriptor that binds

self

automatically. What’s happening -

@prefect.task

returns a

Task

object. Prefect’s descriptor injects

self

at runtime, so

svc.connect_to_email_reader()

works when you run it. - Pyright doesn’t “see” that injected

self

, so it thinks you’re calling a function that still requires a positional

self

argument and reports “No overloads for __call__”. Ways to fix or silence the type error - Keep instance method, silence Pyright on the call site

svc.connect_to_email_reader()  # pyright: ignore[reportCallIssue]

This is simplest if you’re staying in a single process. - Call via the class and pass the instance explicitly (type checker-friendly)

EmailIngestionTasks.connect_to_email_reader(svc)

Prefect supports this calling style too. - Make the task a staticmethod or classmethod (avoids self entirely)

class EmailIngestionTasks:
      def __init__(self, reader, datastore):
          self.reader = reader
          self.datastore = datastore

      @staticmethod
      @prefect.task
      def connect_to_email_reader(host: str, user: str):  # pass what you need
          # build reader here or call a factory
          ...

- Wrap the instance call in a top-level task (clean typing, portable)

@prefect.task
  def connect_to_email_reader_task(reader) -> None:
      reader.connect()

  # then in your flow
  connect_to_email_reader_task(svc.reader)

- Hybrid: keep OO, but pass only serializable config and build clients inside the task (best for distributed workers). This also avoids pickling live clients on the instance. Your minimal example (works at runtime; Pyright may complain unless you use one of the fixes above)

from prefect import task, flow

class EmailIngestionTasks:
    def __init__(self, email_reader, datastore):
        self.reader = email_reader
        self.datastore = datastore

    @task(log_prints=True, retries=2, retry_delay_seconds=10)
    def connect_to_email_reader(self) -> None:
        self.reader.connect()

@flow
def run_flow(reader, datastore):
    svc = EmailIngestionTasks(reader, datastore)
    svc.connect_to_email_reader()  # pyright: ignore[reportCallIssue]

Notes - If you plan to run tasks on remote workers, don’t keep live connections/clients on the instance — they must be picklable. Prefer passing config/credentials and creating clients inside tasks (with LRU caching). - Docs on tasks: Tasks concepts If you share how you’re invoking this (local vs. distributed) and what

EmailReader/Datastore

are, I can suggest the cleanest pattern and typings for your setup.