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)

Create a config-backed resource that loads a Pydantic model from a JSON file.

Parameters:

Name	Type	Description	Default
config_file	str	JSON file where the configuration is stored.	required
path_selector	str | None	Path selector to retrieve a specific value from the JSON map.	None
label	str | None	Human-friendly short name for display in visualizations.	None
description	str | None	Longer description explaining the purpose and contents of this config.	None

Returns:

Name	Type	Description
_ResourceConfig	_ResourceConfig	A configured resource representation.

Example

from typing import Annotated
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.resource import ResourceConfig


class ClassifierConfig(BaseModel):
    categories: list[str]
    threshold: float


class MyWorkflow(Workflow):
    @step
    async def classify(
        self,
        ev: StartEvent,
        config: Annotated[
            ClassifierConfig,
            ResourceConfig(
                config_file="classifier.json",
                label="Classifier",
                description="Classification categories and threshold",
            ),
        ],
    ) -> StopEvent:
        return StopEvent(result=config.categories)

Source code in workflows/resource.py

def ResourceConfig(
    config_file: str,
    path_selector: str | None = None,
    label: str | None = None,
    description: str | None = None,
) -> _ResourceConfig:
    """
    Create a config-backed resource that loads a Pydantic model from a JSON file.

    Args:
        config_file: JSON file where the configuration is stored.
        path_selector: Path selector to retrieve a specific value from the JSON map.
        label: Human-friendly short name for display in visualizations.
        description: Longer description explaining the purpose and contents of this config.

    Returns:
        _ResourceConfig: A configured resource representation.

    Example:
        ```python
        from typing import Annotated
        from pydantic import BaseModel
        from workflows import Workflow, step
        from workflows.events import StartEvent, StopEvent
        from workflows.resource import ResourceConfig


        class ClassifierConfig(BaseModel):
            categories: list[str]
            threshold: float


        class MyWorkflow(Workflow):
            @step
            async def classify(
                self,
                ev: StartEvent,
                config: Annotated[
                    ClassifierConfig,
                    ResourceConfig(
                        config_file="classifier.json",
                        label="Classifier",
                        description="Classification categories and threshold",
                    ),
                ],
            ) -> StopEvent:
                return StopEvent(result=config.categories)
        ```
    """
    return _ResourceConfig(
        config_file=config_file,
        path_selector=path_selector,
        label=label,
        description=description,
    )