EnvVarSecrets: add multi-tenant context support (ContextVar / pipeline-run context)

[tstadel]

Is your feature request related to a problem?

EnvVarSecrets currently resolves secret values exclusively via os.environ, which is a process-global namespace. This makes it effectively impossible to safely host multiple tenants or multiple concurrent pipeline configurations within a single process — there is no way to supply per-request or per-tenant secret values without mutating the global environment, which introduces race conditions and security risks.

This is a meaningful gap for anyone running Haystack as a hosted service (e.g. a multi-pipeline API server, or any multi-tenant SaaS product built on Haystack).

Describe the solution you'd like

Two levels of improvement, ordered by implementation complexity:

1. Quick win — contextvars.ContextVar support

Augment EnvVarSecrets.resolve() to check a ContextVar registry before falling back to os.environ. Callers can then set per-coroutine/per-thread context before invoking a pipeline:

from haystack.core.secrets import env_var_context

token = env_var_context.set({"OPENAI_API_KEY": "sk-tenant-xyz"})
try:
    pipeline.run(...)
finally:
    env_var_context.reset(token)

This is low-overhead and safe for both asyncio and thread-pool concurrency models since ContextVar is isolated per task/thread context.

However ContextVar scopes are by far not easy to understand and manage especially when supporting both async and sync contexts, often found in asgi web frameworks such as in fastAPI. Proper setup for all edge-cases is not straightforward and comes with costs and caveats. Furthermore debugging can become a pain requiring deep insights into low-level frameworks such as anyio, or asyncio.

2. Clean solution — first-class pipeline run context

Introduce a SecretsContext (or a general PipelineRunContext) object that can be passed directly to pipeline.run():

pipeline.run(
    inputs={...},
    context=PipelineRunContext(
        secrets={"OPENAI_API_KEY": "sk-tenant-xyz"}
    )
)

EnvVarSecrets (and any other secret resolver) would receive this context and prefer it over os.environ. This is the most ergonomic and explicit API, keeps secrets out of global state entirely, and makes the data flow auditable.

Downsides of this approach would be that some components would need to be rewritten, i.e. moving secret values resolution from init and warmup to run().

3. Clean solution but less effort — first-class pipeline init context

Introduce a SecretsContext (or a general PipelineContext) object that can be passed directly to Pipeline():

pipeline = Pipeline(
    context=PipelineRunContext(
        secrets={"OPENAI_API_KEY": "sk-tenant-xyz"}
    )
)
pipeline.run(inputs={...},)

EnvVarSecrets (and any other secret resolver) would receive this context and prefer it over os.environ. This is the best balance between ergonomic, explicit API, and keeping components as is (e.g. resolving secret values at component init and warmup()).

Describe alternatives you've considered

• Monkey-patching os.environ per request — unsafe under concurrency.
• One process per tenant — operationally expensive, defeats the purpose of a shared runtime.
• Custom Secret subclass per tenant — works today but requires callers to re-instantiate entire pipeline component graphs per tenant, which is wasteful.

Additional context

The key design constraint here is that EnvVarSecrets must remain the standard way to configure secrets on components — introducing a separate Secret type that tenants or hosting layers have to swap in would create a two-tier system where pipeline definitions need to differ between environments. That breaks the promise that a pipeline configured by one person can be handed off to a hosting service unchanged.

Put differently: where secrets come from at runtime should be a concern of the hosting layer, not the pipeline author. A pipeline configurator should be able to write EnvVarSecrets("OPENAI_API_KEY") and trust that whoever runs it will supply the value through whatever mechanism is appropriate — os.environ in local dev, a ContextVar or PipelineRunContext in a hosted multi-tenant service. The resolution strategy should be an invisible infrastructure detail, not something baked into the pipeline definition.

This also means the lookup order in EnvVarSecrets.resolve() should reflect that layering: pipeline run context → ContextVar → os.environ, so that more specific scopes naturally override broader ones without any action required from the pipeline author.

• EnvVarSecrets is defined in haystack/core/secrets.py
• The ContextVar approach requires no public API changes and could ship as a non-breaking minor addition
• The PipelineRunContext approach would require a small signature change to Pipeline.run() but is fully backwards-compatible with a default of None
• This is a prerequisite for safely building multi-tenant hosted Haystack services without process-per-tenant isolation

Would you be willing to submit a PR? Yes / open to discussion on the preferred approach first.

[tstadel]

Tracing settings currently controlled by environment variables such as enabling content tracing would also be nice to have as pipeline run setting. E.g. one wants to replay a single request with content tracing for debugging purposes.

[tstadel]

Adding some concrete context on this

Concrete use case: AI error resolution assistant

We're building an AI-powered error resolution assistant on top of Haystack. Each user authenticates with their own platform API key to connect to a remote MCP server, which executes tools on their behalf — so User A should only ever access their own workspaces and pipelines, and User B theirs. This is a hard multi-tenancy requirement, not just a configuration convenience.

Right now, the only safe workaround we're investigating is running one worker process per request and patching EnvVarSecrets to support ContextVars https://github.com/deepset-ai/haystack-runtime/pull/628. That unblocks the assistant functionally, but it limits the way we can scale the e.g. system (one request per worker only).

Broader context: serverless and multi-org execution

While there is ongoing work for running pipelines with different dependencies in parallel which could also unblock the assistant use-case, this feature would allow us to manage process boundaries at the right granularity (org-level, workspace-level, or pipeline-level) rather than being forced into one-process-per-request as the only isolation mechanism.

Impersonation / per-tenant credential injection is also a recurring topic for enterprise customers, so this isn't purely an internal concern.

[julian-risch]

Hi @tstadel Sebastian and I talked about this issue today and you're welcome to open a PR to Haystack main branch (or v3 branch if there is no urgency) implementing your idea "first-class pipeline init context".
Please also have a look at this related issue: #11312

[anakin87]

As per Julian's suggestion, I looked into this (with some LLM assistance).

Using context vars seems a good idea to me, to implement a minimal mechanism.

In the following proposal, override is applied at Pipeline init.

In haystack/utils/auth.py

# AI-generated code
_env_overrides: contextvars.ContextVar[dict[str, str]] = contextvars.ContextVar(
    "haystack_env_overrides", default={}
)

@contextlib.contextmanager
def env_overrides(values: dict[str, str]):
    """Override env-var lookups for EnvVarSecret within this scope."""
    merged = {**_env_overrides.get(), **values}
    token = _env_overrides.set(merged)
    try:
        yield
    finally:
        _env_overrides.reset(token)

In EnvVarSecret.resolve_value():

  def resolve_value(self) -> Any | None:
      overrides = _env_overrides.get()
      for env_var in self._env_vars:
          value = overrides.get(env_var) or os.getenv(env_var)
          if value is not None:
              return value
      if self._strict:
          raise ValueError(f"None of the following authentication environment variables are set: {self._env_vars}")
      return None

Example usage:

from haystack import Pipeline
from haystack.utils.auth import env_overrides

PIPELINE_YAML = open("pipeline.yaml").read()

# could be cached
def get_pipeline_for(tenant_id: str) -> Pipeline:
    creds = vault.fetch(tenant_id)
    with env_overrides(creds):
        return Pipeline.loads(PIPELINE_YAML)

def handle_request(tenant_id: str, inputs: dict):
    pipeline = get_pipeline_for(tenant_id)
    return pipeline.run(inputs)

Pipeline authors keep writing their pipelines unchanged; secrets resolution strategy is an infrastructure detail.

Pros I see:

• Small, non-breaking change
• No changes to Pipeline, Component, ...
• if we want to add a more structured API later, it can be a wrapper of this

For more advanced use cases, it's probably worth taking a look into proper gateways (Portkey, LiteLLM AI gateway).

[SyedShahmeerAli12]

Hey, would love to take a stab at this the ContextVar-based env_overrides() approach outlined above looks solid and the change is contained to auth.py + tests. Can I open a PR for it?

[anakin87]

@SyedShahmeerAli12 this is something that we prefer to handle internally. Thanks for your collaboration!

[tstadel]

@anakin87 I like your approach. That's also what we basically tested on platform side. It's definitely valuable, but it comes with important caveats as mentioned in the issue description. Here's an example:

When running a streaming api using a synchronous generator for streaming, e.g.

def stream_pipeline(request):
    with env_overrides(request.env):
        p = Pipeline()
        ....
        for chunk in chunks:
           yield chunk

@app.post("/stream")
def stream(request):
    return StreamingResponse(stream_pipeline(request))

As a consequence

_env_overrides.reset(token)

breaks as fastAPI's StreamingResponse runs each (iterator) iteration in a separate context on a worker thread. So as long as you consume the env only within the first iteration, in theory it's fine, but reset will still raise on the last iteration.
This could be solved by using

contextlib.suppress(ValueError):
    _env_overrides.reset(token)

But what's worse:

• you won't get any feedback on the broken context if you want to consume the env after the first yield
• this behavior is not even document in fastAPI/starlette
• people might spend a good bunch of time debugging lower level frameworks (fastapi, starlett, anyio, asyncio) to fully understand the consequences

[anakin87]

My impression is that we could overcome the error you are describing as follows, using the override at pipeline initialization and keeping yield outside the override scope:

def stream_pipeline(request):
    with env_overrides(request.env):
        p = Pipeline()
    ...
    for chunk in chunks:
       yield chunk

@app.post("/stream")
def stream(request):
    return StreamingResponse(stream_pipeline(request))

We could also separate building from streaming:

@app.post("/stream")
def stream(request):
  p = get_pipeline_for(request.tenant_id)
  return StreamingResponse(p.run(...))

Caveats:

• I don't know the platform ecosystem, so this might not be feasible
• The main limit of this proposal is using overrides only at pipeline initialization, not at run

(This is to better understand the use case, and see if a minimal change can solve it...)

[tstadel]

@anakin87 I agree, we can have it as pipeline initialization feature only. I took a bit of time to investigate how Option 3 would look like and it turns out to be extremely complex if we don't want to run into the same or even worse issues:

• E.g. we'd need to override default value secrets during from_dict to get around ContextVars
• When creating a pipeline from code (without from_dict) we currently have no means to inject any pipeline context into components, as components are being created outside of the pipeline and during add_component it's already too late as most components resolve secret values during component init

Given that, I guess ContextVars is the most practical way for now. Would you like to go forward with the implementation?

[redbotster]

The lookup order you've proposed -- pipeline run context → ContextVar → os.environ -- is the right layering. The key benefit is that pipeline authors stay oblivious to deployment topology; the hosting layer decides resolution strategy at runtime.

One additional consideration for multi-tenant hosted deployments: the secret values themselves should never transit through the pipeline's data plane. The pattern that avoids this is brokered access -- the hosting layer hands the pipeline a short-lived token that references a secret, and the resolution step does a vault lookup rather than passing the raw key. This way a logging misconfiguration or exception dump can't accidentally expose a tenant's credential.

We built exactly this pattern into 1Claw Vault for agent pipelines (HSM-backed, scoped to agent identity). The PipelineRunContext shape you're proposing is a clean integration point for that kind of broker -- just wanted to surface it as a design constraint worth keeping open. Docs at docs.1claw.xyz if useful.

[tstadel]

@redbotster thanks for adding your point of view, very valid points.
@anakin87 I did a bit of research and I wouldn't recommend to go with init-only secrets. While it's true that most components only ever call resolve_value() during init, there are exceptions even in the main package:

• SASEvaluator, DALLEImageGenerator, SentenceTransformersDiversityRanker, SentenceTransformersSimilarityRanker, TransformersSimilarityRanker, ExtractiveReader call it during warmup()
• HuggingFaceTEIRanker calls it during run() and run_async()
• SearchApiWebSearch, SerperDevWebSearch call it during run() and run_async() and __init__()

So, full support for runtime secret value resolution seems to be a bigger topic that might make sense to move to the next major version. We can still go for ContextVar support in v2 with appropriate comments and maybe raising a warning whenever we detect errors during context cleanup. WDYT?