AsyncUserContextMiddleware Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware_typing/test_middleware_typing.py lines 316–331

class AsyncUserContextMiddleware(AgentMiddleware[AgentState[Any], UserContext, Any]):
    """Async middleware with correctly typed ContextT."""

    async def abefore_model(
        self, state: AgentState[Any], runtime: Runtime[UserContext]
    ) -> dict[str, Any] | None:
        _user_name: str = runtime.context["user_name"]
        return None

    async def awrap_model_call(
        self,
        request: ModelRequest[UserContext],
        handler: Callable[[ModelRequest[UserContext]], Awaitable[ModelResponse[Any]]],
    ) -> ModelResponse[Any]:
        _user_id: str = request.runtime.context["user_id"]
        return await handler(request)