AG-UI dependencies per request

[kadosh1000]

How to Initialize Dependencies Per Request in ag_ui?

Summary

I need to initialize different dependencies for each user/request when using to_ag_ui(), but the current implementation appears to share dependencies across all users.

Use Case

I'm building a multi-tenant application where each user should have their own isolated dependencies (e.g., database connections, user-specific context, API clients with different credentials). Currently, it seems like dependencies are initialized once and shared across all ag_ui sessions.

What I've Tried

from pydantic_ai import Agent
from pydantic_ai.dependencies import RunContext

# Current approach - but this creates shared dependencies
def get_user_db_connection(user_id: str):
    # This should be called per user, not shared
    return DatabaseConnection(user_id=user_id)

agent = Agent(
    'openai:gpt-4',
    deps_type=DatabaseConnection,
)

@agent.tool
def query_data(ctx: RunContext[DatabaseConnection], query: str) -> str:
    # ctx.deps is shared across users - not what we want
    return ctx.deps.execute(query)

# This creates a shared instance
app = agent.to_ag_ui()

Expected Behavior

Each user session should be able to initialize its own dependencies, possibly through:

• A dependency factory function that receives request context
• Per-session dependency initialization hooks
• Request-scoped dependency injection

Something like:

def dependency_factory(request_context):
    user_id = extract_user_from_request(request_context)
    return DatabaseConnection(user_id=user_id)

app = agent.to_ag_ui(deps_factory=dependency_factory)

Actual Behavior

Dependencies appear to be initialized once and shared across all users, making it impossible to have user-specific context.

Questions

1. Is there a current way to achieve per-request dependency initialization?
2. If not, would you be open to adding this functionality?
3. What would be the preferred API design for this feature?

Environment

• pydantic-ai version: 0.4.9
• Python version: 3.14

[DouweM]

@kadosh1000 Thanks for the report.

On question 1, that was a bit of an oversight. It's possible to do this by using a class that's currently private:

pydantic-ai/pydantic_ai_slim/pydantic_ai/ag_ui.py

Lines 188 to 218 in 0584724

	adapter = _Adapter(agent=agent)
	async def endpoint(request: Request) -> Response | StreamingResponse:
	"""Endpoint to run the agent with the provided input data."""
	accept = request.headers.get('accept', SSE_CONTENT_TYPE)
	try:
	input_data = RunAgentInput.model_validate(await request.json())
	except ValidationError as e: # pragma: no cover
	return Response(
	content=json.dumps(e.json()),
	media_type='application/json',
	status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
	)
	return StreamingResponse(
	adapter.run(
	input_data,
	accept,
	output_type=output_type,
	model=model,
	deps=deps,
	model_settings=model_settings,
	usage_limits=usage_limits,
	usage=usage,
	infer_name=infer_name,
	toolsets=toolsets,
	),
	media_type=SSE_CONTENT_TYPE,
	)
	self.router.add_route('/', endpoint, methods=['POST'], name='run_agent')

You can build your own pydantic_ai.ag_ui._Adapter for an agent, and then call adapter.run with request-specific deps inside your request handler. Can you give that a try please? If it works, we can easily make that class public and document it.

On question 2, can you please split that off into a separate issue? I'll respond there with some more thoughts.

[kadosh1000]

Thanks for the reply @DouweM , I saw that private class in the code, wondered if there's something I'm missing.

I don't think implementing pydantic_ai.ag_ui._Adapter will be the solution, as the issue seems to be with the endpoint function which calls adapter.run.

We have in mind implementing AGUIApp, so we can override the endpoint function, and thinking of several options to tackle the deps:

1. Set dependencies using a middleware - request.state.ag_ui_deps = <DEPS>

def middleware(request: Request):
    user_id = extract_user_from_request(request)
    request.state.ag_ui_deps = user_id

2. Create a DepsFactory which will get the Request and return the Deps to use

def dependency_factory(request: Request) -> AgentDepsT:
    user_id = extract_user_from_request(request_context)
    return userId

app = agent.to_ag_ui(deps_factory=dependency_factory)

We're leaning towards option 2, but it would be great to have your input as well

Just created #2382 for question 2 and edited this issue

[haithamshahin333]

+1 to this issue, also running into the same conceptual problem where this not a clear way to instantiate deps per request. I believe this is also similar to #2320.

@DouweM appreciate your thoughts on this, thank you!

[DouweM]

@haithamshahin333 I'll work on a solution right now and hope to release before end of day!

[haithamshahin333]

@haithamshahin333 I'll work on a solution right now and hope to release before end of day!

Thank you! Appreciate your work and help on this, I'll give it a try and come back with any feedback.