Dependency Injection

FastAPI-style Depends() handler injection, scanner, and solver. See the Dependency Injection guide for usage patterns.

Depends

Depends

Depends(
    dependency: Callable[..., T], *, use_cache: bool = True
) -> T

Marker placed as the default value of a handler/dependency parameter.

The return type is annotated as T so the IDE and type checker treat the parameter as the resolved dependency type, not as a :class:Dependency sentinel. At runtime the function returns a :class:Dependency object that the scanner picks up.

Parameters:

Name	Type	Description	Default
dependency	Callable[..., T]	A callable that produces the dependency. May be a plain function, an async function, a sync generator, or an async generator. Generators get teardown semantics — anything after yield runs after the handler finishes.	required
use_cache	bool	When True (default), repeated references to the same callable within one request return the cached value. Set to False for resources that must be fresh per use (e.g. random IDs, timestamps).	True

Returns:

Type	Description
T	Statically typed as T, the dependency's resolved value.
T	At runtime returns a :class:Dependency sentinel.

Source code in src/agenticapi/dependencies/depends.py

def Depends(  # noqa: N802, UP047 — name mirrors FastAPI's public API on purpose
    dependency: Callable[..., T],
    *,
    use_cache: bool = True,
) -> T:
    """Marker placed as the default value of a handler/dependency parameter.

    The return type is annotated as ``T`` so the IDE and type checker
    treat the parameter as the resolved dependency type, not as a
    :class:`Dependency` sentinel. At runtime the function returns a
    :class:`Dependency` object that the scanner picks up.

    Args:
        dependency: A callable that produces the dependency. May be a
            plain function, an async function, a sync generator, or an
            async generator. Generators get teardown semantics —
            anything after ``yield`` runs after the handler finishes.
        use_cache: When True (default), repeated references to the
            same callable within one request return the cached value.
            Set to False for resources that must be fresh per use
            (e.g. random IDs, timestamps).

    Returns:
        Statically typed as ``T``, the dependency's resolved value.
        At runtime returns a :class:`Dependency` sentinel.
    """
    return cast("T", Dependency(callable=dependency, use_cache=use_cache))

Dependency dataclass

A resolved dependency description.

Carries the user-supplied callable plus its options. The runtime solver consumes these to build the per-request injection plan.

Attributes:

Name	Type	Description
callable	Callable[..., Any]	The dependency provider. May be sync, async, a sync generator (yield-based teardown), or an async generator (async yield-based teardown).
use_cache	bool	When True (default), the same dependency callable yields the same value within one request. When False, the callable is invoked on every reference within the request.

Source code in src/agenticapi/dependencies/depends.py

@dataclass(frozen=True, slots=True)
class Dependency:
    """A resolved dependency description.

    Carries the user-supplied callable plus its options. The runtime
    solver consumes these to build the per-request injection plan.

    Attributes:
        callable: The dependency provider. May be sync, async, a
            sync generator (yield-based teardown), or an async
            generator (async yield-based teardown).
        use_cache: When True (default), the same dependency callable
            yields the same value within one request. When False, the
            callable is invoked on every reference within the request.
    """

    callable: Callable[..., Any]
    use_cache: bool = True

Scanner

The scanner inspects a handler's signature at registration time and produces an InjectionPlan describing how each parameter should be resolved.

scan_handler

scan_handler(handler: Callable[..., Any]) -> InjectionPlan

Scan a handler signature and return its :class:InjectionPlan.

Resolves string annotations via :func:typing.get_type_hints so from __future__ import annotations is fully supported. Falls back to raw inspect.Parameter.annotation strings when the handler imports something only under TYPE_CHECKING.

Parameters:

Name	Type	Description	Default
handler	Callable[..., Any]	The async or sync handler callable to scan.	required

Returns:

Name	Type	Description
An	InjectionPlan	class:InjectionPlan ready for the solver.

Source code in src/agenticapi/dependencies/scanner.py

def scan_handler(handler: Callable[..., Any]) -> InjectionPlan:
    """Scan a handler signature and return its :class:`InjectionPlan`.

    Resolves string annotations via :func:`typing.get_type_hints` so
    ``from __future__ import annotations`` is fully supported. Falls
    back to raw ``inspect.Parameter.annotation`` strings when the
    handler imports something only under ``TYPE_CHECKING``.

    Args:
        handler: The async or sync handler callable to scan.

    Returns:
        An :class:`InjectionPlan` ready for the solver.
    """
    sig = inspect.signature(handler)
    try:
        type_hints = get_type_hints(handler, include_extras=True)
    except Exception:
        type_hints = {}

    params: list[ParamPlan] = []
    legacy_positional_count = 0
    seen_annotated_intent = False
    seen_annotated_context = False
    intent_payload_schema: type[BaseModel] | None = None

    for index, (param_name, param) in enumerate(sig.parameters.items()):
        annotation = type_hints.get(param_name, param.annotation)
        default = param.default

        # 1) Depends(...) default value — highest precedence.
        if isinstance(default, Dependency):
            params.append(
                ParamPlan(
                    name=param_name,
                    kind=InjectionKind.DEPENDS,
                    dependency=default,
                    annotation=annotation,
                )
            )
            continue

        # 2) Built-in annotated injectors.
        if _is_intent_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.INTENT, annotation=annotation))
            seen_annotated_intent = True
            # Phase D4: capture the typed payload model from Intent[T]
            # so the framework can constrain LLM output to match.
            if intent_payload_schema is None:
                intent_payload_schema = _extract_intent_payload_schema(annotation)
            continue
        if _is_context_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.CONTEXT, annotation=annotation))
            seen_annotated_context = True
            continue
        if _is_agent_tasks_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.AGENT_TASKS, annotation=annotation))
            continue
        if _is_uploaded_files_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.UPLOADED_FILES, annotation=annotation))
            continue
        if _is_htmx_headers_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.HTMX_HEADERS, annotation=annotation))
            continue
        if _is_agent_stream_annotation(annotation):
            params.append(ParamPlan(name=param_name, kind=InjectionKind.AGENT_STREAM, annotation=annotation))
            continue

        # 3) Legacy positional fall-through.
        # Handlers historically accepted ``(intent, context)`` without
        # type annotations. We preserve that by treating the first two
        # unannotated parameters as positional intent/context slots,
        # but only if they haven't already been satisfied by an
        # annotated parameter above.
        if index < 2 and annotation is inspect.Parameter.empty:
            params.append(
                ParamPlan(
                    name=param_name,
                    kind=InjectionKind.POSITIONAL_LEGACY,
                    annotation=None,
                )
            )
            legacy_positional_count += 1
            continue

        # Unknown parameter with no resolver — leave it absent so the
        # call site supplies it (or fails with a clear TypeError).
        params.append(
            ParamPlan(
                name=param_name,
                kind=InjectionKind.POSITIONAL_LEGACY,
                annotation=annotation,
            )
        )

    # If neither Intent nor AgentContext was annotated AND we did not
    # collect any legacy positionals, fall back to legacy positional
    # mode for the first two params (preserves the (intent, context)
    # contract for handlers that use string annotations the type-hint
    # resolver couldn't load).
    if not seen_annotated_intent and not seen_annotated_context and legacy_positional_count == 0:
        # Mark the first up-to-two params as legacy positionals.
        new_params: list[ParamPlan] = []
        for i, plan in enumerate(params):
            if i < 2 and plan.kind not in {
                InjectionKind.AGENT_TASKS,
                InjectionKind.UPLOADED_FILES,
                InjectionKind.HTMX_HEADERS,
                InjectionKind.AGENT_STREAM,
                InjectionKind.DEPENDS,
            }:
                new_params.append(ParamPlan(name=plan.name, kind=InjectionKind.POSITIONAL_LEGACY, annotation=None))
                legacy_positional_count += 1
            else:
                new_params.append(plan)
        params = new_params

    return InjectionPlan(
        params=tuple(params),
        legacy_positional_count=legacy_positional_count,
        intent_payload_schema=intent_payload_schema,
    )

InjectionKind

Bases: StrEnum

Categorisation of how a single parameter is filled.

Source code in src/agenticapi/dependencies/scanner.py

class InjectionKind(StrEnum):
    """Categorisation of how a single parameter is filled."""

    INTENT = "intent"
    CONTEXT = "context"
    AGENT_TASKS = "agent_tasks"
    UPLOADED_FILES = "uploaded_files"
    HTMX_HEADERS = "htmx_headers"
    AGENT_STREAM = "agent_stream"
    DEPENDS = "depends"
    POSITIONAL_LEGACY = "positional_legacy"

InjectionPlan dataclass

Cached injection plan for a single handler.

Attributes:

Name	Type	Description
params	tuple[ParamPlan, ...]	Per-parameter resolution plans, in declaration order.
legacy_positional_count	int	How many of the leading parameters should receive intent/context positionally for handlers that don't annotate them.
intent_payload_schema	type[BaseModel] | None	Pydantic model parameter extracted from an Intent[T] annotation, or None for handlers that use bare Intent or no annotation. Forwarded to the IntentParser so the LLM is constrained to produce a payload matching T.

Source code in src/agenticapi/dependencies/scanner.py

@dataclass(frozen=True, slots=True)
class InjectionPlan:
    """Cached injection plan for a single handler.

    Attributes:
        params: Per-parameter resolution plans, in declaration order.
        legacy_positional_count: How many of the leading parameters
            should receive ``intent``/``context`` positionally for
            handlers that don't annotate them.
        intent_payload_schema: Pydantic model parameter extracted from
            an ``Intent[T]`` annotation, or ``None`` for handlers that
            use bare ``Intent`` or no annotation. Forwarded to the
            ``IntentParser`` so the LLM is constrained to produce a
            payload matching ``T``.
    """

    params: tuple[ParamPlan, ...] = field(default_factory=tuple)
    legacy_positional_count: int = 0
    intent_payload_schema: type[BaseModel] | None = None

ParamPlan dataclass

How a single handler parameter is resolved.

Attributes:

Name	Type	Description
name	str	The parameter's name in the handler signature.
kind	InjectionKind	Which injector handles this parameter.
dependency	Dependency | None	The user-supplied :class:Dependency for DEPENDS parameters; None otherwise.
annotation	Any	The resolved annotation for diagnostics.

Source code in src/agenticapi/dependencies/scanner.py

@dataclass(frozen=True, slots=True)
class ParamPlan:
    """How a single handler parameter is resolved.

    Attributes:
        name: The parameter's name in the handler signature.
        kind: Which injector handles this parameter.
        dependency: The user-supplied :class:`Dependency` for
            ``DEPENDS`` parameters; ``None`` otherwise.
        annotation: The resolved annotation for diagnostics.
    """

    name: str
    kind: InjectionKind
    dependency: Dependency | None = None
    annotation: Any = None

Solver

The solver applies an InjectionPlan to a live request, resolving all dependencies and producing a ResolvedHandlerCall that can be invoked through an AsyncExitStack.

solve async

solve(
    plan: InjectionPlan,
    *,
    intent: Intent[Any],
    context: AgentContext,
    files: dict[str, Any] | None,
    htmx_scope: dict[str, Any] | None,
    overrides: dict[Callable[..., Any], Callable[..., Any]],
    route_dependencies: list[Dependency] | None = None,
    agent_stream: Any | None = None,
) -> ResolvedHandlerCall

Resolve a handler's :class:InjectionPlan for one request.

Parameters:

Name	Type	Description	Default
plan	InjectionPlan	The handler's pre-computed injection plan.	required
intent	Intent[Any]	The parsed agent intent.	required
context	AgentContext	The active :class:AgentContext.	required
files	dict[str, Any] | None	Uploaded files keyed by form-field name (or None).	required
htmx_scope	dict[str, Any] | None	Raw ASGI scope for HTMX header parsing (or None).	required
overrides	dict[Callable[..., Any], Callable[..., Any]]	app.dependency_overrides map.	required
route_dependencies	list[Dependency] | None	Optional list of route-level dependencies (D6) to resolve for side effects before the handler runs. Their teardown is registered on the same exit stack.	None
agent_stream	Any | None	Optional :class:AgentStream instance to inject into handlers that declared an AgentStream parameter. Phase F1; None for non-streaming endpoints.	None

Returns:

Name	Type	Description
A	ResolvedHandlerCall	class:ResolvedHandlerCall whose exit_stack must be
	ResolvedHandlerCall	closed after the handler completes.

Source code in src/agenticapi/dependencies/solver.py

async def solve(
    plan: InjectionPlan,
    *,
    intent: Intent[Any],
    context: AgentContext,
    files: dict[str, Any] | None,
    htmx_scope: dict[str, Any] | None,
    overrides: dict[Callable[..., Any], Callable[..., Any]],
    route_dependencies: list[Dependency] | None = None,
    agent_stream: Any | None = None,
) -> ResolvedHandlerCall:
    """Resolve a handler's :class:`InjectionPlan` for one request.

    Args:
        plan: The handler's pre-computed injection plan.
        intent: The parsed agent intent.
        context: The active :class:`AgentContext`.
        files: Uploaded files keyed by form-field name (or ``None``).
        htmx_scope: Raw ASGI scope for HTMX header parsing (or ``None``).
        overrides: ``app.dependency_overrides`` map.
        route_dependencies: Optional list of route-level dependencies
            (D6) to resolve for side effects before the handler runs.
            Their teardown is registered on the same exit stack.
        agent_stream: Optional :class:`AgentStream` instance to inject
            into handlers that declared an ``AgentStream`` parameter.
            Phase F1; ``None`` for non-streaming endpoints.

    Returns:
        A :class:`ResolvedHandlerCall` whose ``exit_stack`` must be
        closed after the handler completes.
    """
    from agenticapi.interface.htmx import HtmxHeaders
    from agenticapi.interface.tasks import AgentTasks

    call = ResolvedHandlerCall()
    cache: dict[Callable[..., Any], Any] = {}
    positional_buffer: list[Any] = []

    # Phase D6: route-level dependencies run first, for side effects
    # only. Their return values are discarded but exceptions propagate
    # up the stack so e.g. an auth check can short-circuit the request
    # by raising AuthenticationError.
    for dep in route_dependencies or ():
        await _resolve_dependency(dep, overrides, cache, call.exit_stack, chain=[])

    for param in plan.params:
        if param.kind is InjectionKind.INTENT:
            call.kwargs[param.name] = intent
        elif param.kind is InjectionKind.CONTEXT:
            call.kwargs[param.name] = context
        elif param.kind is InjectionKind.AGENT_TASKS:
            tasks = AgentTasks()
            call.tasks = tasks
            call.kwargs[param.name] = tasks
        elif param.kind is InjectionKind.UPLOADED_FILES:
            call.kwargs[param.name] = files or {}
        elif param.kind is InjectionKind.HTMX_HEADERS:
            call.kwargs[param.name] = HtmxHeaders.from_scope(htmx_scope or {})
        elif param.kind is InjectionKind.AGENT_STREAM:
            # Phase F1: stream injection. The framework supplies the
            # stream when streaming is enabled on the endpoint.
            # Otherwise we leave it absent — the handler will get a
            # TypeError at call time which is the right diagnostic.
            if agent_stream is not None:
                call.kwargs[param.name] = agent_stream
        elif param.kind is InjectionKind.DEPENDS:
            if param.dependency is None:
                raise DependencyResolutionError(
                    f"Parameter '{param.name}' declared as Depends but has no dependency callable",
                    chain=[],
                )
            call.kwargs[param.name] = await _resolve_dependency(
                param.dependency, overrides, cache, call.exit_stack, chain=[]
            )
        elif param.kind is InjectionKind.POSITIONAL_LEGACY:
            # Fill positional intent / context slots in declaration order.
            positional_buffer.append(intent if len(positional_buffer) == 0 else context)

    call.positional = tuple(positional_buffer)
    return call

invoke_handler async

invoke_handler(
    handler: Callable[..., Awaitable[Any] | Any],
    resolved: ResolvedHandlerCall,
) -> Any

Invoke a handler with a resolved call and await its result.

Supports both sync and async handlers. Always closes the exit_stack after the handler returns, even on failure, so generator-style teardown runs reliably.

Source code in src/agenticapi/dependencies/solver.py

async def invoke_handler(
    handler: Callable[..., Awaitable[Any] | Any],
    resolved: ResolvedHandlerCall,
) -> Any:
    """Invoke a handler with a resolved call and await its result.

    Supports both sync and async handlers. Always closes the
    ``exit_stack`` after the handler returns, even on failure, so
    generator-style teardown runs reliably.
    """
    try:
        result = handler(*resolved.positional, **resolved.kwargs)
        if inspect.isawaitable(result):
            result = await result
        return result
    finally:
        await resolved.exit_stack.aclose()

ResolvedHandlerCall dataclass

The product of resolving a handler's :class:InjectionPlan.

Attributes:

Name	Type	Description
kwargs	dict[str, Any]	Keyword arguments to pass to the handler.
positional	tuple[Any, ...]	Positional arguments (Intent, AgentContext) for handlers using the legacy (intent, context) shape.
tasks	AgentTasks | None	The injected :class:AgentTasks, or None if the handler did not request one.
exit_stack	AsyncExitStack	The async exit stack that owns generator-based dependency teardown. Must be closed after the handler returns (success or failure).

Source code in src/agenticapi/dependencies/solver.py

@dataclass(slots=True)
class ResolvedHandlerCall:
    """The product of resolving a handler's :class:`InjectionPlan`.

    Attributes:
        kwargs: Keyword arguments to pass to the handler.
        positional: Positional arguments (Intent, AgentContext) for
            handlers using the legacy ``(intent, context)`` shape.
        tasks: The injected :class:`AgentTasks`, or ``None`` if the
            handler did not request one.
        exit_stack: The async exit stack that owns generator-based
            dependency teardown. Must be closed after the handler
            returns (success or failure).
    """

    kwargs: dict[str, Any] = field(default_factory=dict)
    positional: tuple[Any, ...] = ()
    tasks: _AgentTasks | None = None
    exit_stack: AsyncExitStack = field(default_factory=AsyncExitStack)

DependencyResolutionError

Bases: AgentRuntimeError

Raised when a dependency cannot be resolved.

Carries the dependency call chain so debugging is straightforward.

Source code in src/agenticapi/dependencies/solver.py

class DependencyResolutionError(AgentRuntimeError):
    """Raised when a dependency cannot be resolved.

    Carries the dependency call chain so debugging is straightforward.
    """

    def __init__(self, message: str, *, chain: list[str] | None = None) -> None:
        super().__init__(message)
        self.chain = chain or []