Resource

Declare a resource to inject into step functions.

Parameters:

Name	Type	Description	Default
factory	Callable[..., T] | None	Function returning the resource instance. May be async.	required
cache	bool	If True, reuse the produced resource across steps. Defaults to True.	True

Returns:

Type	Description
_Resource	_Resource[T]: A resource descriptor to be used in typing.Annotated.

Examples:

from typing import Annotated
from workflows.resource import Resource

def get_memory(**kwargs) -> Memory:
    return Memory.from_defaults("user123", token_limit=60000)

class MyWorkflow(Workflow):
    @step
    async def first(
        self,
        ev: StartEvent,
        memory: Annotated[Memory, Resource(get_memory)],
    ) -> StopEvent:
        await memory.aput(...)
        return StopEvent(result="ok")

Source code in workflows/resource.py

def Resource(
    factory: Callable[..., T],
    cache: bool = True,
) -> _Resource:
    """Declare a resource to inject into step functions.

    Args:
        factory (Callable[..., T] | None): Function returning the resource instance. May be async.
        cache (bool): If True, reuse the produced resource across steps. Defaults to True.

    Returns:
        _Resource[T]: A resource descriptor to be used in `typing.Annotated`.

    Examples:
        ```python
        from typing import Annotated
        from workflows.resource import Resource

        def get_memory(**kwargs) -> Memory:
            return Memory.from_defaults("user123", token_limit=60000)

        class MyWorkflow(Workflow):
            @step
            async def first(
                self,
                ev: StartEvent,
                memory: Annotated[Memory, Resource(get_memory)],
            ) -> StopEvent:
                await memory.aput(...)
                return StopEvent(result="ok")
        ```
    """
    return _Resource(factory, cache)