Skip to content

Editor Agents Reference

AWS Lambda

encord_agents.aws.editor_agent 

editor_agent(*, label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None) -> Callable[[AgentFunction], Callable[[Dict[str, Any], Any], Dict[str, Any]]]

Wrapper to make resources available for AWS Lambda editor agents.

Source code in encord_agents/aws/wrappers.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None,
) -> Callable[[AgentFunction], Callable[[Dict[str, Any], Any], Dict[str, Any]]]:
    """
    Wrapper to make resources available for AWS Lambda editor agents.
    """

    def context_wrapper_inner(func: AgentFunction) -> Callable[[Dict[str, Any], Any], Dict[str, Any]]:
        dependant = get_dependant(func=func)

        @wraps(func)
        def wrapper(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
            headers = event.get("headers", {})
            if headers.get(EDITOR_TEST_REQUEST_HEADER) or headers.get(EDITOR_TEST_REQUEST_HEADER.lower()):
                logging.info("Editor test request")
                return _generate_response()

            try:
                body = event.get("body")
                frame_data: FrameData | None = None
                if not body:
                    return {"statusCode": 400, "body": {"errors": ["No request body"], "message": "No request body"}}
                if isinstance(body, str):
                    logging.info("Parsing body as string json")
                    frame_data = FrameData.model_validate_json(body)
                elif isinstance(body, dict):
                    logging.info("Parsing body as json object")
                    frame_data = FrameData.model_validate(body)
                logging.info(f"Request: {frame_data}")
            except ValidationError as err:
                logging.error(f"Error parsing request: {err}")
                return {
                    "statusCode": 400,
                    "body": {
                        "errors": err.errors(),
                        "message": ", ".join([e["msg"] for e in err.errors()]),
                    },
                }
            frame_data = cast(FrameData, frame_data)

            client = get_user_client()
            try:
                project = client.get_project(frame_data.project_hash)
            except AuthorisationError:
                return {"statusCode": 403, "body": "Forbidden"}

            label_row: LabelRowV2 | None = None
            if dependant.needs_label_row:
                include_args = label_row_metadata_include_args or LabelRowMetadataIncludeArgs()
                init_args = label_row_initialise_labels_args or LabelRowInitialiseLabelsArgs()
                label_row = project.list_label_rows_v2(
                    data_hashes=[str(frame_data.data_hash)], **include_args.model_dump()
                )[0]
                label_row.initialise_labels(**init_args.model_dump())

            storage_item: StorageItem | None = None
            if dependant.needs_storage_item:
                if label_row is None:
                    label_row = project.list_label_rows_v2(data_hashes=[frame_data.data_hash])[0]
                assert label_row.backing_item_uuid, "This is a server response so guaranteed to have this"
                try:
                    storage_item = client.get_storage_item(label_row.backing_item_uuid)
                except AuthorisationError:
                    return {"statusCode": 403, "body": "Forbidden"}

            context_obj = Context(
                project=project, label_row=label_row, frame_data=frame_data, storage_item=storage_item
            )
            result: None | Any | EditorAgentResponse = None
            with ExitStack() as stack:
                try:
                    dependencies = solve_dependencies(context=context_obj, dependant=dependant, stack=stack)
                except AuthorisationError:
                    return {"statusCode": 403, "body": "Forbidden"}
                try:
                    result = func(**dependencies.values)
                    if isinstance(result, EditorAgentResponse):
                        return _generate_response(result.model_dump())
                    return _generate_response()
                except EncordEditorAgentException as exc:
                    return _generate_response(
                        exc.json_response_body,
                        status_code=HTTPStatus.BAD_REQUEST,
                    )

        return wrapper

    return context_wrapper_inner

encord_agents.aws.dependencies

DAssetPath module-attribute 

DAssetPath = Annotated[Path, Depends(dep_asset)]

Get a local file path to data asset temporarily stored till end of agent execution.

DEncordClient module-attribute 

DEncordClient = Annotated[EncordUserClient, Depends(dep_client)]

Get an authenticated user client.

DObjectCrops module-attribute 

DObjectCrops = Annotated[list[InstanceCrop], Depends(dep_object_crops)]

Get all object crops that the agent was triggered on. The instance crop contains the object instance, the frame content (pixel values), and the frame.

DObjectsInstances module-attribute 

DObjectsInstances = Annotated[list[ObjectInstance], Depends(dep_objects)]

Get all object instances that the agent was triggered on. No pixels, just the annotation.

DSingleFrame module-attribute 

DSingleFrame = Annotated[NDArray[uint8], Depends(dep_single_frame)]

Get the single frame that the agent was triggered on.

DStorageItem module-attribute 

DStorageItem = Annotated[StorageItem, Depends(dep_storage_item)]

Get the storage item associated with the underlying agent task to, for example, read/write client metadata or read data properties.

DVideoIterator module-attribute 

DVideoIterator = Annotated[Iterator[Frame], Depends(dep_video_iterator)]

Get a video frame iterator for doing things over many frames.

dep_asset 

dep_asset(storage_item: StorageItem) -> Generator[Path, None, None]

Returns a local file path to the data asset, temporarily stored for the duration of the agent's execution.

This dependency fetches the underlying data asset using a signed URL.

The asset is temporarily stored on disk for the duration of the task and is automatically removed once the task completes.

Example:

from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_asset
...
runner = Runner(project_hash="<project_hash_a>")

@editor_agent()
def my_agent(
    asset: Annotated[Path, Depends(dep_asset)]
) -> None:
    asset.stat()  # read file stats
    ...

Returns:

  • None –

    The path to the asset.

Raises:

  • ValueError –

    if the underlying assets are not videos, images, or audio.

  • EncordException –

    if data type not supported by SDK yet.

Source code in encord_agents/core/dependencies/serverless.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def dep_asset(storage_item: StorageItem) -> Generator[Path, None, None]:
    """
    Returns a local file path to the data asset, temporarily stored for the duration of the agent's execution.


    This dependency fetches the underlying data asset using a signed URL.

    The asset is temporarily stored on disk for the duration of the task and is automatically removed once the task
    completes.

    **Example:**

    ```python
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_asset
    ...
    runner = Runner(project_hash="<project_hash_a>")

    @editor_agent()
    def my_agent(
        asset: Annotated[Path, Depends(dep_asset)]
    ) -> None:
        asset.stat()  # read file stats
        ...
    ```

    Returns:
        The path to the asset.

    Raises:
        ValueError: if the underlying assets are not videos, images, or audio.
        EncordException: if data type not supported by SDK yet.
    """
    with download_asset(storage_item) as asset:
        yield asset

dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_client
...
@editor_agent()
def (
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
    client.get_dataset("")
Source code in encord_agents/core/dependencies/serverless.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example:**

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_client
    ...
    @editor_agent()
    def (
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
        client.get_dataset("")
    ```

    """
    return get_user_client()

dep_object_crops 

dep_object_crops(filter_ontology_objects: list[Object | str] | None = None) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]

Returns a list of object instances and frame crops associated with each object.

One example use-case is to run each crop against a model.

Example:

@editor_agent
def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
    for crop in crops:
        crop.content  # <- this is raw numpy rgb values
        crop.frame    # <- this is the frame number in video
        crop.instance # <- this is the object instance from the label row
        crop.b64_encoding()  # <- a base64 encoding of the image content
    ...

Parameters:

  • filter_ontology_objects (list[Object | str] | None, default: None ) –

    Specify a list of ontology objects to include. If provided, only instances of these object types are included. Strings are matched against feature_node_hashes.

Returns: The dependency to be injected into the cloud function.

Source code in encord_agents/core/dependencies/serverless.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def dep_object_crops(
    filter_ontology_objects: list[Object | str] | None = None,
) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]:
    """
    Returns a list of object instances and frame crops associated with each object.

    One example use-case is to run each crop against a model.

    **Example:**

    ```python
    @editor_agent
    def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
        for crop in crops:
            crop.content  # <- this is raw numpy rgb values
            crop.frame    # <- this is the frame number in video
            crop.instance # <- this is the object instance from the label row
            crop.b64_encoding()  # <- a base64 encoding of the image content
        ...
    ```

    Args:
        filter_ontology_objects: Specify a list of ontology objects to include.
            If provided, only instances of these object types are included.
            Strings are matched against `feature_node_hashes`.


    Returns: The dependency to be injected into the cloud function.

    """
    from encord_agents.core.vision import crop_to_object

    legal_feature_hashes = {
        o.feature_node_hash if isinstance(o, Object) else o for o in (filter_ontology_objects or [])
    }

    def _dep_object_crops(
        frame_data: FrameData, lr: LabelRowV2, frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ) -> list[InstanceCrop]:
        legal_shapes = {Shape.POLYGON, Shape.BOUNDING_BOX, Shape.ROTATABLE_BOUNDING_BOX, Shape.BITMASK}
        return [
            InstanceCrop(
                frame=frame_data.frame,
                content=crop_to_object(frame, o.get_annotation(frame=frame_data.frame).coordinates),  # type: ignore
                instance=o,
            )
            for o in lr.get_object_instances(filter_frames=frame_data.frame)
            if o.ontology_item.shape in legal_shapes
            and (not legal_feature_hashes or o.feature_hash in legal_feature_hashes)
            and (not frame_data.object_hashes or o.object_hash in frame_data.object_hashes)
        ]

    return _dep_object_crops

dep_single_frame 

dep_single_frame(storage_item: StorageItem, frame_data: FrameData) -> NDArray[np.uint8]

Dependency to inject the first frame of the underlying asset.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished running, the downloaded file is removed from the file system.

Example:

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_single_frame
...

@editor_agent()
def my_agent(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert frame.ndim == 3, "Will work"

Parameters:

  • storage_item (StorageItem) –

    The Storage item. Automatically injected (see example above).

Returns:

  • NDArray[uint8] –

    Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/core/dependencies/serverless.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def dep_single_frame(storage_item: StorageItem, frame_data: FrameData) -> NDArray[np.uint8]:
    """
    Dependency to inject the first frame of the underlying asset.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished running, the downloaded file is removed from the file system.

    **Example:**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_single_frame
    ...

    @editor_agent()
    def my_agent(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert frame.ndim == 3, "Will work"
    ```

    Args:
        storage_item: The Storage item. Automatically injected (see example above).

    Returns:
        Numpy array of shape [h, w, 3] RGB colors.

    """

    try:
        import cv2
    except ImportError:
        raise ImportError(
            "Your data agent is depending on computer vision capabilities and `opencv` is not installed. Please install either `opencv-python` or `opencv-python-headless`."
        )

    with download_asset(storage_item, frame=frame_data.frame) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)

    return np.asarray(img, dtype=np.uint8)

dep_storage_item 

dep_storage_item(storage_item: StorageItem) -> StorageItem

Get the storage item associated with the underlying agent task.

The StorageItem is useful for multiple things like

  • Updating client metadata
  • Reading file properties like storage location, fps, duration, DICOM tags, etc.

Example

from typing_extensions import Annotated
from encord.storage import StorageItem
from encord_agents.gcp import editor_agent, Depends
from encord_agents.gcp.dependencies import dep_storage_item


@editor_agent()
def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
    print("uuid", storage_item.uuid)
    print("client_metadata", storage_item.client_metadata)
    ...
Source code in encord_agents/core/dependencies/serverless.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def dep_storage_item(storage_item: StorageItem) -> StorageItem:
    r"""
    Get the storage item associated with the underlying agent task.

    The [`StorageItem`](https://docs.encord.com/sdk-documentation/sdk-references/StorageItem){ target="\_blank", rel="noopener noreferrer" }
    is useful for multiple things like

    * Updating client metadata
    * Reading file properties like storage location, fps, duration, DICOM tags, etc.

    **Example**

    ```python
    from typing_extensions import Annotated
    from encord.storage import StorageItem
    from encord_agents.gcp import editor_agent, Depends
    from encord_agents.gcp.dependencies import dep_storage_item


    @editor_agent()
    def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
        print("uuid", storage_item.uuid)
        print("client_metadata", storage_item.client_metadata)
        ...
    ```

    """
    return storage_item

dep_video_iterator 

dep_video_iterator(storage_item: StorageItem) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for performing operations over many frames.

Example:

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_video_iterator
...

@editor_agent()
def my_agent(
    video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
):
    for frame in video_frames:
        print(frame.frame, frame.content.shape)

Parameters:

  • storage_item (StorageItem) –

    Automatically injected storage item dependency.

Raises:

  • NotImplementedError –

    Fails for data types other than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/core/dependencies/serverless.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def dep_video_iterator(storage_item: StorageItem) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for performing operations over many frames.

    **Example:**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_video_iterator
    ...

    @editor_agent()
    def my_agent(
        video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
    ):
        for frame in video_frames:
            print(frame.frame, frame.content.shape)
    ```

    Args:
        storage_item: Automatically injected storage item dependency.

    Raises:
        NotImplementedError: Fails for data types other than video.

    Yields:
        An iterator.

    """
    from encord_agents.core.video import iter_video

    if not storage_item.item_type == StorageItemType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")

    with download_asset(storage_item, None) as asset:
        yield iter_video(asset)

encord_agents.aws.wrappers

editor_agent 

editor_agent(*, label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None) -> Callable[[AgentFunction], Callable[[Dict[str, Any], Any], Dict[str, Any]]]

Wrapper to make resources available for AWS Lambda editor agents.

Source code in encord_agents/aws/wrappers.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None,
) -> Callable[[AgentFunction], Callable[[Dict[str, Any], Any], Dict[str, Any]]]:
    """
    Wrapper to make resources available for AWS Lambda editor agents.
    """

    def context_wrapper_inner(func: AgentFunction) -> Callable[[Dict[str, Any], Any], Dict[str, Any]]:
        dependant = get_dependant(func=func)

        @wraps(func)
        def wrapper(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
            headers = event.get("headers", {})
            if headers.get(EDITOR_TEST_REQUEST_HEADER) or headers.get(EDITOR_TEST_REQUEST_HEADER.lower()):
                logging.info("Editor test request")
                return _generate_response()

            try:
                body = event.get("body")
                frame_data: FrameData | None = None
                if not body:
                    return {"statusCode": 400, "body": {"errors": ["No request body"], "message": "No request body"}}
                if isinstance(body, str):
                    logging.info("Parsing body as string json")
                    frame_data = FrameData.model_validate_json(body)
                elif isinstance(body, dict):
                    logging.info("Parsing body as json object")
                    frame_data = FrameData.model_validate(body)
                logging.info(f"Request: {frame_data}")
            except ValidationError as err:
                logging.error(f"Error parsing request: {err}")
                return {
                    "statusCode": 400,
                    "body": {
                        "errors": err.errors(),
                        "message": ", ".join([e["msg"] for e in err.errors()]),
                    },
                }
            frame_data = cast(FrameData, frame_data)

            client = get_user_client()
            try:
                project = client.get_project(frame_data.project_hash)
            except AuthorisationError:
                return {"statusCode": 403, "body": "Forbidden"}

            label_row: LabelRowV2 | None = None
            if dependant.needs_label_row:
                include_args = label_row_metadata_include_args or LabelRowMetadataIncludeArgs()
                init_args = label_row_initialise_labels_args or LabelRowInitialiseLabelsArgs()
                label_row = project.list_label_rows_v2(
                    data_hashes=[str(frame_data.data_hash)], **include_args.model_dump()
                )[0]
                label_row.initialise_labels(**init_args.model_dump())

            storage_item: StorageItem | None = None
            if dependant.needs_storage_item:
                if label_row is None:
                    label_row = project.list_label_rows_v2(data_hashes=[frame_data.data_hash])[0]
                assert label_row.backing_item_uuid, "This is a server response so guaranteed to have this"
                try:
                    storage_item = client.get_storage_item(label_row.backing_item_uuid)
                except AuthorisationError:
                    return {"statusCode": 403, "body": "Forbidden"}

            context_obj = Context(
                project=project, label_row=label_row, frame_data=frame_data, storage_item=storage_item
            )
            result: None | Any | EditorAgentResponse = None
            with ExitStack() as stack:
                try:
                    dependencies = solve_dependencies(context=context_obj, dependant=dependant, stack=stack)
                except AuthorisationError:
                    return {"statusCode": 403, "body": "Forbidden"}
                try:
                    result = func(**dependencies.values)
                    if isinstance(result, EditorAgentResponse):
                        return _generate_response(result.model_dump())
                    return _generate_response()
                except EncordEditorAgentException as exc:
                    return _generate_response(
                        exc.json_response_body,
                        status_code=HTTPStatus.BAD_REQUEST,
                    )

        return wrapper

    return context_wrapper_inner

FastAPI

encord_agents.fastapi.dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.fastapi.dependencies import dep_client
...
@app.post("/my-route")
def my_route(
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
Source code in encord_agents/fastapi/dependencies.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example**:

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.fastapi.dependencies import dep_client
    ...
    @app.post("/my-route")
    def my_route(
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
    ```

    """
    return get_user_client()

encord_agents.fastapi.dep_label_row 

dep_label_row(frame_data: FrameData) -> LabelRowV2

Dependency to provide an initialized label row.

Example:

from encord_agents.fastapi.dependencies import dep_label_row
...


@app.post("/my-route")
def my_route(
    lr: Annotated[LabelRowV2, Depends(dep_label_row)]
):
    assert lr.is_labelling_initialised  # will work

Parameters:

  • frame_data (FrameData) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above)

Returns:

  • LabelRowV2 –

    The initialized label row.

Source code in encord_agents/fastapi/dependencies.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def dep_label_row(frame_data: FrameData) -> LabelRowV2:
    """
    Dependency to provide an initialized label row.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_label_row
    ...


    @app.post("/my-route")
    def my_route(
        lr: Annotated[LabelRowV2, Depends(dep_label_row)]
    ):
        assert lr.is_labelling_initialised  # will work
    ```

    Args:
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above)

    Returns:
        The initialized label row.

    """
    return get_initialised_label_row(frame_data)

encord_agents.fastapi.dep_single_frame 

dep_single_frame(storage_item: Annotated[StorageItem, Depends(dep_storage_item)], frame_data: FrameData) -> NDArray[np.uint8]

Dependency to inject the underlying asset of the frame data.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished, the downloaded file will be removed from the file system.

Example:

from encord_agents.fastapi.dependencies import dep_single_frame
...

@app.post("/my-route")
def my_route(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert arr.ndim == 3, "Will work"

Parameters:

  • storage_item (Annotated[StorageItem, Depends(dep_storage_item)]) –

    The label row. Automatically injected (see example above).

  • frame_data (FrameData) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above).

Returns: Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/fastapi/dependencies.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def dep_single_frame(
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)], frame_data: FrameData
) -> NDArray[np.uint8]:
    """
    Dependency to inject the underlying asset of the frame data.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished, the downloaded file will be removed from the file system.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_single_frame
    ...

    @app.post("/my-route")
    def my_route(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert arr.ndim == 3, "Will work"
    ```

    Args:
        storage_item: The label row. Automatically injected (see example above).
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above).

    Returns: Numpy array of shape [h, w, 3] RGB colors.

    """

    try:
        import cv2
    except ImportError:
        raise ImportError(
            "Your data agent is depending on computer vision capabilities and `opencv` is not installed. Please install either `opencv-python` or `opencv-python-headless`."
        )

    with download_asset(storage_item, frame_data.frame) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)
    return np.asarray(img, dtype=np.uint8)

encord_agents.fastapi.get_encord_app 

get_encord_app(*, custom_cors_regex: str | None = None) -> FastAPI

Get a FastAPI app with the Encord middleware.

Parameters:

  • custom_cors_regex (str | None, default: None ) –

    A regex to use for the CORS middleware. Only necessary if you are not using the default Encord domain.

Returns:

  • FastAPI ( FastAPI ) –

    A FastAPI app with the Encord middleware.

Source code in encord_agents/fastapi/cors.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def get_encord_app(*, custom_cors_regex: str | None = None) -> FastAPI:
    """
    Get a FastAPI app with the Encord middleware.

    Args:
        custom_cors_regex (str | None, optional): A regex to use for the CORS middleware.
            Only necessary if you are not using the default Encord domain.

    Returns:
        FastAPI: A FastAPI app with the Encord middleware.
    """
    app = FastAPI()
    app.add_middleware(
        EncordCORSMiddleware,
        allow_origin_regex=custom_cors_regex or ENCORD_DOMAIN_REGEX,
    )
    app.add_middleware(EncordTestHeaderMiddleware)
    app.exception_handlers[AuthorisationError] = _authorization_error_exception_handler
    app.exception_handlers[EncordEditorAgentException] = _encord_editor_agent_exception_handler
    return app

encord_agents.fastapi.verify_auth 

verify_auth() -> None

FastAPI lifecycle start hook to fail early if ssh key is missing.

Example:

from fastapi import FastAPI

app = FastAPI(
    on_startup=[verify_auth]

This will make the server fail early if auth is not set up.

Source code in encord_agents/fastapi/utils.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def verify_auth() -> None:
    """
    FastAPI lifecycle start hook to fail early if ssh key is missing.

    **Example:**

    ```python
    from fastapi import FastAPI

    app = FastAPI(
        on_startup=[verify_auth]
    ```

    This will make the server fail early if auth is not set up.
    """
    from datetime import datetime, timedelta

    Settings()

    try:
        client = get_user_client()
        client.get_projects(created_after=datetime.now() - timedelta(days=1))
    except Exception:
        import traceback

        stack = traceback.format_exc()
        raise PrintableError(
            f"[red]Was able to read the SSH key, but couldn't list projects with Encord.[/red]. Original error was:{os.linesep}{stack}"
        )

encord_agents.fastapi.cors

Convenience method to easily extend FastAPI servers with the appropriate CORS Middleware to allow interactions from the Encord platform.

EncordCORSMiddleware

Bases: CORSMiddleware

Like a regular fastapi.middleware.cors.CORSMiddleware but matches against the Encord origin by default and handles X-Encord-Editor-Agent test header

Example: 

from fastapi import FastAPI
from encord_agents.fastapi.cors import EncordCORSMiddleware

app = FastAPI()
app.add_middleware(EncordCORSMiddleware)

The CORS middleware will allow POST requests from the Encord domain.

Source code in encord_agents/fastapi/cors.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class EncordCORSMiddleware(CORSMiddleware):  # type: ignore [misc, unused-ignore]
    """
    Like a regular `fastapi.middleware.cors.CORSMiddleware` but matches against
    the Encord origin by default and handles X-Encord-Editor-Agent test header

    **Example:**
    ```python
    from fastapi import FastAPI
    from encord_agents.fastapi.cors import EncordCORSMiddleware

    app = FastAPI()
    app.add_middleware(EncordCORSMiddleware)
    ```

    The CORS middleware will allow POST requests from the Encord domain.
    """

    def __init__(
        self,
        app: ASGIApp,
        allow_origins: typing.Sequence[str] = (),
        allow_methods: typing.Sequence[str] = ("POST",),
        allow_headers: typing.Sequence[str] = (EDITOR_TEST_REQUEST_HEADER,),
        allow_credentials: bool = False,
        allow_origin_regex: str = ENCORD_DOMAIN_REGEX,
        expose_headers: typing.Sequence[str] = (),
        max_age: int = 3600,
    ) -> None:
        super().__init__(
            app,
            allow_origins,
            allow_methods,
            allow_headers,
            allow_credentials,
            allow_origin_regex,
            expose_headers,
            max_age,
        )

EncordTestHeaderMiddleware

Bases: BaseHTTPMiddleware

Source code in encord_agents/fastapi/cors.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class EncordTestHeaderMiddleware(BaseHTTPMiddleware):  # type: ignore [misc, unused-ignore]
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        """
        Middleware to handle the X-Encord-Editor-Agent test header.

        Args:
            request (Request):
            call_next (RequestResponseEndpoint):

        Returns:
            Response
        """
        if request.method == "POST":
            if request.headers.get(EDITOR_TEST_REQUEST_HEADER):
                return JSONResponse(content=None, status_code=200)

        return await call_next(request)
dispatch async 
dispatch(request: Request, call_next: RequestResponseEndpoint) -> Response

Middleware to handle the X-Encord-Editor-Agent test header.

Parameters:

  • request (Request) –
  • call_next (RequestResponseEndpoint) –

Returns:

  • Response –

    Response

Source code in encord_agents/fastapi/cors.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
    """
    Middleware to handle the X-Encord-Editor-Agent test header.

    Args:
        request (Request):
        call_next (RequestResponseEndpoint):

    Returns:
        Response
    """
    if request.method == "POST":
        if request.headers.get(EDITOR_TEST_REQUEST_HEADER):
            return JSONResponse(content=None, status_code=200)

    return await call_next(request)

get_encord_app 

get_encord_app(*, custom_cors_regex: str | None = None) -> FastAPI

Get a FastAPI app with the Encord middleware.

Parameters:

  • custom_cors_regex (str | None, default: None ) –

    A regex to use for the CORS middleware. Only necessary if you are not using the default Encord domain.

Returns:

  • FastAPI ( FastAPI ) –

    A FastAPI app with the Encord middleware.

Source code in encord_agents/fastapi/cors.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def get_encord_app(*, custom_cors_regex: str | None = None) -> FastAPI:
    """
    Get a FastAPI app with the Encord middleware.

    Args:
        custom_cors_regex (str | None, optional): A regex to use for the CORS middleware.
            Only necessary if you are not using the default Encord domain.

    Returns:
        FastAPI: A FastAPI app with the Encord middleware.
    """
    app = FastAPI()
    app.add_middleware(
        EncordCORSMiddleware,
        allow_origin_regex=custom_cors_regex or ENCORD_DOMAIN_REGEX,
    )
    app.add_middleware(EncordTestHeaderMiddleware)
    app.exception_handlers[AuthorisationError] = _authorization_error_exception_handler
    app.exception_handlers[EncordEditorAgentException] = _encord_editor_agent_exception_handler
    return app

encord_agents.fastapi.dependencies

Dependencies for injection in FastAPI servers.

This module contains dependencies that you can inject within your api routes. Dependencies that depend on others don't need to be used together. They'll work just fine alone.

Note that you can also use the function parameter: 

from typing_extensions import Annotated
from fastapi import Form
from encord_agents import FrameData
...
@app.post("/my-agent-route")
def my_agent(
    frame_data: FrameData,
):
    ...
FrameData is automatically injected via the api request body.

dep_asset 

dep_asset(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]) -> Generator[Path, None, None]

Get a local file path to data asset temporarily stored till end of agent execution.

This dependency will fetch the underlying data asset based on a signed url. It will temporarily store the data on disk. Once the task is completed, the asset will be removed from disk again.

Example:

from encord_agents.fastapi.dependencies import dep_asset
...
runner = Runner(project_hash="<project_hash_a>")

@app.post("/my-route")
def my_agent(
    asset: Annotated[Path, Depends(dep_asset)],
) -> str | None:
    asset.stat()  # read file stats
    ...

Returns:

  • None –

    The path to the asset.

Raises:

  • ValueError –

    if the underlying assets are not videos, images, or audio.

  • EncordException –

    if data type not supported by SDK yet.

Source code in encord_agents/fastapi/dependencies.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def dep_asset(
    storage_item: Annotated[
        StorageItem,
        Depends(dep_storage_item),
    ],
) -> Generator[Path, None, None]:
    """
    Get a local file path to data asset temporarily stored till end of agent execution.

    This dependency will fetch the underlying data asset based on a signed url.
    It will temporarily store the data on disk. Once the task is completed, the
    asset will be removed from disk again.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_asset
    ...
    runner = Runner(project_hash="<project_hash_a>")

    @app.post("/my-route")
    def my_agent(
        asset: Annotated[Path, Depends(dep_asset)],
    ) -> str | None:
        asset.stat()  # read file stats
        ...
    ```

    Returns:
        The path to the asset.

    Raises:
        ValueError: if the underlying assets are not videos, images, or audio.
        EncordException: if data type not supported by SDK yet.
    """
    with download_asset(storage_item) as asset:
        yield asset

dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.fastapi.dependencies import dep_client
...
@app.post("/my-route")
def my_route(
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
Source code in encord_agents/fastapi/dependencies.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example**:

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.fastapi.dependencies import dep_client
    ...
    @app.post("/my-route")
    def my_route(
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
    ```

    """
    return get_user_client()

dep_data_lookup 

dep_data_lookup(lookup: Annotated[DataLookup, Depends(_lookup_adapter)]) -> DataLookup

Get a lookup to easily retrieve data rows and storage items associated with the given task.

Deprecated

dep_data_lookup is deprecated and will be removed in version 0.2.10. Use dep_storage_item instead for accessing storage items.

Migration Guide:

# Old way (deprecated)
from encord_agents.fastapi.dependencies import dep_data_lookup, DataLookup

@app.post("/my-agent")
def my_agent(
    frame_data: FrameData,
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
):
    storage_item = lookup.get_storage_item(frame_data.data_hash)
    print(storage_item.client_metadata)

# New way (recommended)
from encord_agents.fastapi.dependencies import dep_storage_item

@app.post("/my-agent")
def my_agent(
    frame_data: FrameData,
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
):
    # storage_item is directly available
    print(storage_item.client_metadata)

Parameters:

  • lookup (Annotated[DataLookup, Depends(_lookup_adapter)]) –

    The object that you can use to lookup data rows and storage items. Automatically injected.

Returns:

  • DataLookup –

    The (shared) lookup object.

Source code in encord_agents/fastapi/dependencies.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def dep_data_lookup(lookup: Annotated[DataLookup, Depends(_lookup_adapter)]) -> DataLookup:
    """
    Get a lookup to easily retrieve data rows and storage items associated with the given task.

    !!! warning "Deprecated"
        `dep_data_lookup` is deprecated and will be removed in version 0.2.10.
        Use `dep_storage_item` instead for accessing storage items.

    **Migration Guide:**

    ```python
    # Old way (deprecated)
    from encord_agents.fastapi.dependencies import dep_data_lookup, DataLookup

    @app.post("/my-agent")
    def my_agent(
        frame_data: FrameData,
        lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
    ):
        storage_item = lookup.get_storage_item(frame_data.data_hash)
        print(storage_item.client_metadata)

    # New way (recommended)
    from encord_agents.fastapi.dependencies import dep_storage_item

    @app.post("/my-agent")
    def my_agent(
        frame_data: FrameData,
        storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
    ):
        # storage_item is directly available
        print(storage_item.client_metadata)
    ```

    Args:
        lookup: The object that you can use to lookup data rows and storage items. Automatically injected.

    Returns:
        The (shared) lookup object.

    """
    import warnings

    warnings.warn(
        "dep_data_lookup is deprecated and will be removed in version 0.2.10. "
        "Use 'dep_storage_item' instead for accessing storage items. "
        "See the function docstring for migration examples.",
        DeprecationWarning,
        stacklevel=2,
    )
    return lookup

dep_label_row 

dep_label_row(frame_data: FrameData) -> LabelRowV2

Dependency to provide an initialized label row.

Example:

from encord_agents.fastapi.dependencies import dep_label_row
...


@app.post("/my-route")
def my_route(
    lr: Annotated[LabelRowV2, Depends(dep_label_row)]
):
    assert lr.is_labelling_initialised  # will work

Parameters:

  • frame_data (FrameData) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above)

Returns:

  • LabelRowV2 –

    The initialized label row.

Source code in encord_agents/fastapi/dependencies.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def dep_label_row(frame_data: FrameData) -> LabelRowV2:
    """
    Dependency to provide an initialized label row.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_label_row
    ...


    @app.post("/my-route")
    def my_route(
        lr: Annotated[LabelRowV2, Depends(dep_label_row)]
    ):
        assert lr.is_labelling_initialised  # will work
    ```

    Args:
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above)

    Returns:
        The initialized label row.

    """
    return get_initialised_label_row(frame_data)

dep_label_row_with_args 

dep_label_row_with_args(label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None) -> Callable[[FrameData], LabelRowV2]

Dependency to provide an initialized label row.

Example:

from encord_agents.core.data_model import LabelRowMetadataIncludeArgs, LabelRowInitialiseLabelsArgs
from encord_agents.fastapi.dependencies import dep_label_row_with_args
...

include_args = LabelRowMetadataIncludeArgs(
    include_client_metadata=True,
    include_workflow_graph_node=True,
)
init_args = LabelRowInitialiseLabelsArgs(
    include_signed_url=True,
)

@app.post("/my-route")
def my_route(
    lr: Annotated[LabelRowV2, Depends(dep_label_row_with_args(include_args, init_args))]
):
    assert lr.is_labelling_initialised  # will work
    assert lr.client_metadata           # will be available if set already

Parameters:

  • label_row_metadata_include_args (LabelRowMetadataIncludeArgs | None, default: None ) –

    What arguments to include on the metadata front

  • label_row_initialise_labels_args (LabelRowInitialiseLabelsArgs | None, default: None ) –

    How and whether to initialise the label rows

Returns:

  • Callable[[FrameData], LabelRowV2] –

    The initialized label row.

Source code in encord_agents/fastapi/dependencies.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def dep_label_row_with_args(
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None,
) -> Callable[[FrameData], LabelRowV2]:
    """
    Dependency to provide an initialized label row.

    **Example:**

    ```python
    from encord_agents.core.data_model import LabelRowMetadataIncludeArgs, LabelRowInitialiseLabelsArgs
    from encord_agents.fastapi.dependencies import dep_label_row_with_args
    ...

    include_args = LabelRowMetadataIncludeArgs(
        include_client_metadata=True,
        include_workflow_graph_node=True,
    )
    init_args = LabelRowInitialiseLabelsArgs(
        include_signed_url=True,
    )

    @app.post("/my-route")
    def my_route(
        lr: Annotated[LabelRowV2, Depends(dep_label_row_with_args(include_args, init_args))]
    ):
        assert lr.is_labelling_initialised  # will work
        assert lr.client_metadata           # will be available if set already
    ```

    Args:
        label_row_metadata_include_args: What arguments to include on the metadata front
        label_row_initialise_labels_args: How and whether to initialise the label rows


    Returns:
        The initialized label row.

    """

    def wrapper(frame_data: FrameData) -> LabelRowV2:
        return get_initialised_label_row(
            frame_data, include_args=label_row_metadata_include_args, init_args=label_row_initialise_labels_args
        )

    return wrapper

dep_object_crops 

dep_object_crops(filter_ontology_objects: list[Object | str] | None = None) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]

Create a dependency that provides crops of object instances.

Useful, e.g., to be able to run each crop against a model.

Example:

@app.post("/object_classification")
async def classify_objects(
    crops: Annotated[
        list[InstanceCrop],
        Depends(dep_object_crops(filter_ontology_objects=[generic_ont_obj])),
    ],
):
    for crop in crops:
        crop.content  # <- this is raw numpy rgb values
        crop.frame    # <- this is the frame number in video
        crop.instance # <- this is the object instance from the label row
        crop.b64_encoding()  # <- a base64 encoding of the image content
    ...

Parameters:

  • filter_ontology_objects (list[Object | str] | None, default: None ) –

    Optional list of ontology objects to filter by. If provided, only instances of these object types will be included. Strings are matched against feature_node_hashes.

Returns:

  • Callable[[FrameData, LabelRowV2, NDArray[uint8]], list[InstanceCrop]] –

    A FastAPI dependency function that yields a list of InstanceCrop.

Source code in encord_agents/fastapi/dependencies.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
def dep_object_crops(
    filter_ontology_objects: list[Object | str] | None = None,
) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]:
    """
    Create a dependency that provides crops of object instances.

    Useful, e.g., to be able to run each crop against a model.

    **Example:**

    ```python
    @app.post("/object_classification")
    async def classify_objects(
        crops: Annotated[
            list[InstanceCrop],
            Depends(dep_object_crops(filter_ontology_objects=[generic_ont_obj])),
        ],
    ):
        for crop in crops:
            crop.content  # <- this is raw numpy rgb values
            crop.frame    # <- this is the frame number in video
            crop.instance # <- this is the object instance from the label row
            crop.b64_encoding()  # <- a base64 encoding of the image content
        ...
    ```

    Args:
        filter_ontology_objects: Optional list of ontology objects to filter by.
            If provided, only instances of these object types will be included.
            Strings are matched against `feature_node_hashes`.

    Returns:
        A FastAPI dependency function that yields a list of InstanceCrop.
    """
    legal_feature_hashes = {
        o.feature_node_hash if isinstance(o, Object) else o for o in (filter_ontology_objects or [])
    }

    def _dep_object_crops(
        frame_data: FrameData,
        lr: Annotated[LabelRowV2, Depends(dep_label_row)],
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)],
    ) -> list[InstanceCrop]:
        legal_shapes = {Shape.POLYGON, Shape.BOUNDING_BOX, Shape.ROTATABLE_BOUNDING_BOX, Shape.BITMASK}
        return [
            InstanceCrop(
                frame=frame_data.frame,
                content=crop_to_object(frame, o.get_annotation(frame=frame_data.frame).coordinates),  # type: ignore
                instance=o,
            )
            for o in lr.get_object_instances(filter_frames=frame_data.frame)
            if o.ontology_item.shape in legal_shapes
            and (not legal_feature_hashes or o.feature_hash in legal_feature_hashes)
            and (not frame_data.object_hashes or o.object_hash in frame_data.object_hashes)
        ]

    return _dep_object_crops

dep_project 

dep_project(frame_data: FrameData, client: Annotated[EncordUserClient, Depends(dep_client)]) -> Project

Dependency to provide an instantiated Project.

Example:

from encord.project import Project
from encord_agents.fastapi.dependencies import dep_project
...
@app.post("/my-route")
def my_route(
    project: Annotated[Project, Depends(dep_project)]
):
    # Project will authenticated and ready to use.
    print(project.title)

Parameters:

  • frame_data (FrameData) –
  • client (Annotated[EncordUserClient, Depends(dep_client)]) –

Returns:

Source code in encord_agents/fastapi/dependencies.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def dep_project(frame_data: FrameData, client: Annotated[EncordUserClient, Depends(dep_client)]) -> Project:
    r"""
    Dependency to provide an instantiated
    [Project](https://docs.encord.com/sdk-documentation/sdk-references/LabelRowV2){ target="\_blank", rel="noopener noreferrer" }.

    **Example:**

    ```python
    from encord.project import Project
    from encord_agents.fastapi.dependencies import dep_project
    ...
    @app.post("/my-route")
    def my_route(
        project: Annotated[Project, Depends(dep_project)]
    ):
        # Project will authenticated and ready to use.
        print(project.title)
    ```


    Args:
        frame_data:
        client:

    Returns:

    """
    return client.get_project(project_hash=frame_data.project_hash)

dep_single_frame 

dep_single_frame(storage_item: Annotated[StorageItem, Depends(dep_storage_item)], frame_data: FrameData) -> NDArray[np.uint8]

Dependency to inject the underlying asset of the frame data.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished, the downloaded file will be removed from the file system.

Example:

from encord_agents.fastapi.dependencies import dep_single_frame
...

@app.post("/my-route")
def my_route(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert arr.ndim == 3, "Will work"

Parameters:

  • storage_item (Annotated[StorageItem, Depends(dep_storage_item)]) –

    The label row. Automatically injected (see example above).

  • frame_data (FrameData) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above).

Returns: Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/fastapi/dependencies.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def dep_single_frame(
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)], frame_data: FrameData
) -> NDArray[np.uint8]:
    """
    Dependency to inject the underlying asset of the frame data.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished, the downloaded file will be removed from the file system.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_single_frame
    ...

    @app.post("/my-route")
    def my_route(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert arr.ndim == 3, "Will work"
    ```

    Args:
        storage_item: The label row. Automatically injected (see example above).
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above).

    Returns: Numpy array of shape [h, w, 3] RGB colors.

    """

    try:
        import cv2
    except ImportError:
        raise ImportError(
            "Your data agent is depending on computer vision capabilities and `opencv` is not installed. Please install either `opencv-python` or `opencv-python-headless`."
        )

    with download_asset(storage_item, frame_data.frame) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)
    return np.asarray(img, dtype=np.uint8)

dep_storage_item 

dep_storage_item(label_row: Annotated[LabelRowV2, Depends(dep_label_row)], user_client: Annotated[EncordUserClient, Depends(dep_client)]) -> StorageItem

Get the storage item associated with the underlying agent task.

The StorageItem is useful for multiple things like

  • Updating client metadata
  • Reading file properties like storage location, fps, duration, DICOM tags, etc.

Example

from encord.storage import StorageItem
from encord_agents.fastapi.dependencies import dep_storage_item

@app.post("/my-agent")
def my_agent(
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
):
    # Client will authenticated and ready to use.
    print(storage_item.dicom_study_uid)
    print(storage_item.client_metadata)
Source code in encord_agents/fastapi/dependencies.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def dep_storage_item(
    label_row: Annotated[LabelRowV2, Depends(dep_label_row)],
    user_client: Annotated[EncordUserClient, Depends(dep_client)],
) -> StorageItem:
    r"""
    Get the storage item associated with the underlying agent task.

    The [`StorageItem`](https://docs.encord.com/sdk-documentation/sdk-references/StorageItem){ target="\_blank", rel="noopener noreferrer" }
    is useful for multiple things like

    * Updating client metadata
    * Reading file properties like storage location, fps, duration, DICOM tags, etc.

    **Example**

    ```python
    from encord.storage import StorageItem
    from encord_agents.fastapi.dependencies import dep_storage_item

    @app.post("/my-agent")
    def my_agent(
        storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
    ):
        # Client will authenticated and ready to use.
        print(storage_item.dicom_study_uid)
        print(storage_item.client_metadata)
    ```

    """
    assert label_row.backing_item_uuid, "All responses from BE include this field"
    return user_client.get_storage_item(label_row.backing_item_uuid)

dep_video_iterator 

dep_video_iterator(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for doing things over many frames.

Example:

from encord_agents.fastapi.dependencies import dep_video_iterator, Frame
...

@app.post("/my-route")
def my_route(
    video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
):
    for frame in video_frames:
        print(frame.frame, frame.content.shape)

Parameters:

  • storage_item (Annotated[StorageItem, Depends(dep_storage_item)]) –

    Automatically injected storage item dependency.

Raises:

  • NotImplementedError –

    Will fail for other data types than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/fastapi/dependencies.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def dep_video_iterator(
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)],
) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for doing things over many frames.

    **Example:**

    ```python
    from encord_agents.fastapi.dependencies import dep_video_iterator, Frame
    ...

    @app.post("/my-route")
    def my_route(
        video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
    ):
        for frame in video_frames:
            print(frame.frame, frame.content.shape)
    ```

    Args:
        storage_item: Automatically injected storage item dependency.

    Raises:
        NotImplementedError: Will fail for other data types than video.

    Yields:
        An iterator.

    """
    if not storage_item.item_type == StorageItemType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")
    with download_asset(storage_item, None) as asset:
        yield iter_video(asset)

encord_agents.fastapi.utils

verify_auth 

verify_auth() -> None

FastAPI lifecycle start hook to fail early if ssh key is missing.

Example:

from fastapi import FastAPI

app = FastAPI(
    on_startup=[verify_auth]

This will make the server fail early if auth is not set up.

Source code in encord_agents/fastapi/utils.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def verify_auth() -> None:
    """
    FastAPI lifecycle start hook to fail early if ssh key is missing.

    **Example:**

    ```python
    from fastapi import FastAPI

    app = FastAPI(
        on_startup=[verify_auth]
    ```

    This will make the server fail early if auth is not set up.
    """
    from datetime import datetime, timedelta

    Settings()

    try:
        client = get_user_client()
        client.get_projects(created_after=datetime.now() - timedelta(days=1))
    except Exception:
        import traceback

        stack = traceback.format_exc()
        raise PrintableError(
            f"[red]Was able to read the SSH key, but couldn't list projects with Encord.[/red]. Original error was:{os.linesep}{stack}"
        )

GCP

encord_agents.gcp.editor_agent 

editor_agent(*, label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None, custom_cors_regex: str | None = None) -> Callable[[AgentFunction], Callable[[Request], Response]]

Wrapper to make resources available for gcp editor agents.

The editor agents are intended to be used via dependency injections. You can learn more via out documentation.

Parameters:

  • label_row_metadata_include_args (LabelRowMetadataIncludeArgs | None, default: None ) –

    arguments to overwrite default arguments on project.list_label_rows_v2().

  • label_row_initialise_labels_args (LabelRowInitialiseLabelsArgs | None, default: None ) –

    Arguments to overwrite default arguments on label_row.initialise_labels(...)

  • custom_cors_regex (str | None, default: None ) –

    A regex to use for the CORS settings. If not provided, the default regex will be used. Only required if the agent is not deployed on Encord's platform.

Returns:

  • Callable[[AgentFunction], Callable[[Request], Response]] –

    A wrapped function suitable for gcp functions.

Source code in encord_agents/gcp/wrappers.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None,
    custom_cors_regex: str | None = None,
) -> Callable[[AgentFunction], Callable[[Request], Response]]:
    """
    Wrapper to make resources available for gcp editor agents.

    The editor agents are intended to be used via dependency injections.
    You can learn more via out [documentation](https://agents-docs.encord.com).

    Args:
        label_row_metadata_include_args: arguments to overwrite default arguments
            on `project.list_label_rows_v2()`.
        label_row_initialise_labels_args: Arguments to overwrite default arguments
            on `label_row.initialise_labels(...)`
        custom_cors_regex: A regex to use for the CORS settings. If not provided, the default regex will be used.
            Only required if the agent is not deployed on Encord's platform.

    Returns:
        A wrapped function suitable for gcp functions.
    """

    def context_wrapper_inner(func: AgentFunction) -> Callable[[Request], Response]:
        dependant = get_dependant(func=func)
        cors_regex = re.compile(custom_cors_regex or ENCORD_DOMAIN_REGEX)

        @wraps(func)
        def wrapper(request: Request) -> Response:
            if request.method == "OPTIONS":
                response = make_response("")
                response.headers["Vary"] = "Origin"

                if not cors_regex.fullmatch(request.origin):
                    response.status_code = 403
                    return response

                headers = {
                    "Access-Control-Allow-Origin": request.origin,
                    "Access-Control-Allow-Methods": "POST",
                    "Access-Control-Allow-Headers": f"Content-Type, {EDITOR_TEST_REQUEST_HEADER}",
                    "Access-Control-Max-Age": "3600",
                }
                response.headers.update(headers)
                response.status_code = 204
                return response

            if request.headers.get(EDITOR_TEST_REQUEST_HEADER):
                logging.info("Editor test request")
                return _generate_response()
            if not request.is_json:
                raise Exception("Request should be JSON. Migrated over to new format")
            frame_data = FrameData.model_validate(request.get_json())
            logging.info(f"Request: {frame_data}")

            client = get_user_client()
            try:
                project = client.get_project(frame_data.project_hash)
            except AuthorisationError as err:
                response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                return response

            label_row: LabelRowV2 | None = None
            if dependant.needs_label_row:
                include_args = label_row_metadata_include_args or LabelRowMetadataIncludeArgs()
                init_args = label_row_initialise_labels_args or LabelRowInitialiseLabelsArgs()
                label_row = project.list_label_rows_v2(
                    data_hashes=[str(frame_data.data_hash)], **include_args.model_dump()
                )[0]
                label_row.initialise_labels(**init_args.model_dump())

            storage_item: StorageItem | None = None
            if dependant.needs_storage_item:
                if label_row is None:
                    label_row = project.list_label_rows_v2(data_hashes=[frame_data.data_hash])[0]
                assert label_row.backing_item_uuid, "This is a server response so guaranteed to have this"
                try:
                    storage_item = client.get_storage_item(label_row.backing_item_uuid)
                except AuthorisationError as err:
                    response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                    return response

            context = Context(project=project, label_row=label_row, frame_data=frame_data, storage_item=storage_item)
            result: Any | None | EditorAgentResponse = None
            with ExitStack() as stack:
                try:
                    # Solving dependencies can execute arbitrary code including fetch from Encord platform
                    # e.g: Get storage item which can throw error
                    dependencies = solve_dependencies(context=context, dependant=dependant, stack=stack)
                except AuthorisationError as err:
                    response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                    return response
                try:
                    result = func(**dependencies.values)
                    if isinstance(result, EditorAgentResponse):
                        response = _generate_response(result.model_dump_json(), HTTPStatus.OK)
                        return response
                except EncordEditorAgentException as exc:
                    response = _generate_response(to_jsonable_python(exc.json_response_body), HTTPStatus.BAD_REQUEST)
                    return response
            return _generate_response()

        return wrapper

    return context_wrapper_inner

encord_agents.gcp.dependencies

DAssetPath module-attribute 

DAssetPath = Annotated[Path, Depends(dep_asset)]

Get a local file path to data asset temporarily stored till end of agent execution.

DEncordClient module-attribute 

DEncordClient = Annotated[EncordUserClient, Depends(dep_client)]

Get an authenticated user client.

DObjectCrops module-attribute 

DObjectCrops = Annotated[list[InstanceCrop], Depends(dep_object_crops)]

Get all object crops that the agent was triggered on. The instance crop contains the object instance, the frame content (pixel values), and the frame.

DObjectsInstances module-attribute 

DObjectsInstances = Annotated[list[ObjectInstance], Depends(dep_objects)]

Get all object instances that the agent was triggered on. No pixels, just the annotation.

DSingleFrame module-attribute 

DSingleFrame = Annotated[NDArray[uint8], Depends(dep_single_frame)]

Get the single frame that the agent was triggered on.

DStorageItem module-attribute 

DStorageItem = Annotated[StorageItem, Depends(dep_storage_item)]

Get the storage item associated with the underlying agent task to, for example, read/write client metadata or read data properties.

DVideoIterator module-attribute 

DVideoIterator = Annotated[Iterator[Frame], Depends(dep_video_iterator)]

Get a video frame iterator for doing things over many frames.

dep_asset 

dep_asset(storage_item: StorageItem) -> Generator[Path, None, None]

Returns a local file path to the data asset, temporarily stored for the duration of the agent's execution.

This dependency fetches the underlying data asset using a signed URL.

The asset is temporarily stored on disk for the duration of the task and is automatically removed once the task completes.

Example:

from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_asset
...
runner = Runner(project_hash="<project_hash_a>")

@editor_agent()
def my_agent(
    asset: Annotated[Path, Depends(dep_asset)]
) -> None:
    asset.stat()  # read file stats
    ...

Returns:

  • None –

    The path to the asset.

Raises:

  • ValueError –

    if the underlying assets are not videos, images, or audio.

  • EncordException –

    if data type not supported by SDK yet.

Source code in encord_agents/core/dependencies/serverless.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def dep_asset(storage_item: StorageItem) -> Generator[Path, None, None]:
    """
    Returns a local file path to the data asset, temporarily stored for the duration of the agent's execution.


    This dependency fetches the underlying data asset using a signed URL.

    The asset is temporarily stored on disk for the duration of the task and is automatically removed once the task
    completes.

    **Example:**

    ```python
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_asset
    ...
    runner = Runner(project_hash="<project_hash_a>")

    @editor_agent()
    def my_agent(
        asset: Annotated[Path, Depends(dep_asset)]
    ) -> None:
        asset.stat()  # read file stats
        ...
    ```

    Returns:
        The path to the asset.

    Raises:
        ValueError: if the underlying assets are not videos, images, or audio.
        EncordException: if data type not supported by SDK yet.
    """
    with download_asset(storage_item) as asset:
        yield asset

dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_client
...
@editor_agent()
def (
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
    client.get_dataset("")
Source code in encord_agents/core/dependencies/serverless.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example:**

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_client
    ...
    @editor_agent()
    def (
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
        client.get_dataset("")
    ```

    """
    return get_user_client()

dep_object_crops 

dep_object_crops(filter_ontology_objects: list[Object | str] | None = None) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]

Returns a list of object instances and frame crops associated with each object.

One example use-case is to run each crop against a model.

Example:

@editor_agent
def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
    for crop in crops:
        crop.content  # <- this is raw numpy rgb values
        crop.frame    # <- this is the frame number in video
        crop.instance # <- this is the object instance from the label row
        crop.b64_encoding()  # <- a base64 encoding of the image content
    ...

Parameters:

  • filter_ontology_objects (list[Object | str] | None, default: None ) –

    Specify a list of ontology objects to include. If provided, only instances of these object types are included. Strings are matched against feature_node_hashes.

Returns: The dependency to be injected into the cloud function.

Source code in encord_agents/core/dependencies/serverless.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def dep_object_crops(
    filter_ontology_objects: list[Object | str] | None = None,
) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]:
    """
    Returns a list of object instances and frame crops associated with each object.

    One example use-case is to run each crop against a model.

    **Example:**

    ```python
    @editor_agent
    def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
        for crop in crops:
            crop.content  # <- this is raw numpy rgb values
            crop.frame    # <- this is the frame number in video
            crop.instance # <- this is the object instance from the label row
            crop.b64_encoding()  # <- a base64 encoding of the image content
        ...
    ```

    Args:
        filter_ontology_objects: Specify a list of ontology objects to include.
            If provided, only instances of these object types are included.
            Strings are matched against `feature_node_hashes`.


    Returns: The dependency to be injected into the cloud function.

    """
    from encord_agents.core.vision import crop_to_object

    legal_feature_hashes = {
        o.feature_node_hash if isinstance(o, Object) else o for o in (filter_ontology_objects or [])
    }

    def _dep_object_crops(
        frame_data: FrameData, lr: LabelRowV2, frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ) -> list[InstanceCrop]:
        legal_shapes = {Shape.POLYGON, Shape.BOUNDING_BOX, Shape.ROTATABLE_BOUNDING_BOX, Shape.BITMASK}
        return [
            InstanceCrop(
                frame=frame_data.frame,
                content=crop_to_object(frame, o.get_annotation(frame=frame_data.frame).coordinates),  # type: ignore
                instance=o,
            )
            for o in lr.get_object_instances(filter_frames=frame_data.frame)
            if o.ontology_item.shape in legal_shapes
            and (not legal_feature_hashes or o.feature_hash in legal_feature_hashes)
            and (not frame_data.object_hashes or o.object_hash in frame_data.object_hashes)
        ]

    return _dep_object_crops

dep_single_frame 

dep_single_frame(storage_item: StorageItem, frame_data: FrameData) -> NDArray[np.uint8]

Dependency to inject the first frame of the underlying asset.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished running, the downloaded file is removed from the file system.

Example:

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_single_frame
...

@editor_agent()
def my_agent(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert frame.ndim == 3, "Will work"

Parameters:

  • storage_item (StorageItem) –

    The Storage item. Automatically injected (see example above).

Returns:

  • NDArray[uint8] –

    Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/core/dependencies/serverless.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def dep_single_frame(storage_item: StorageItem, frame_data: FrameData) -> NDArray[np.uint8]:
    """
    Dependency to inject the first frame of the underlying asset.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished running, the downloaded file is removed from the file system.

    **Example:**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_single_frame
    ...

    @editor_agent()
    def my_agent(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert frame.ndim == 3, "Will work"
    ```

    Args:
        storage_item: The Storage item. Automatically injected (see example above).

    Returns:
        Numpy array of shape [h, w, 3] RGB colors.

    """

    try:
        import cv2
    except ImportError:
        raise ImportError(
            "Your data agent is depending on computer vision capabilities and `opencv` is not installed. Please install either `opencv-python` or `opencv-python-headless`."
        )

    with download_asset(storage_item, frame=frame_data.frame) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)

    return np.asarray(img, dtype=np.uint8)

dep_storage_item 

dep_storage_item(storage_item: StorageItem) -> StorageItem

Get the storage item associated with the underlying agent task.

The StorageItem is useful for multiple things like

  • Updating client metadata
  • Reading file properties like storage location, fps, duration, DICOM tags, etc.

Example

from typing_extensions import Annotated
from encord.storage import StorageItem
from encord_agents.gcp import editor_agent, Depends
from encord_agents.gcp.dependencies import dep_storage_item


@editor_agent()
def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
    print("uuid", storage_item.uuid)
    print("client_metadata", storage_item.client_metadata)
    ...
Source code in encord_agents/core/dependencies/serverless.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def dep_storage_item(storage_item: StorageItem) -> StorageItem:
    r"""
    Get the storage item associated with the underlying agent task.

    The [`StorageItem`](https://docs.encord.com/sdk-documentation/sdk-references/StorageItem){ target="\_blank", rel="noopener noreferrer" }
    is useful for multiple things like

    * Updating client metadata
    * Reading file properties like storage location, fps, duration, DICOM tags, etc.

    **Example**

    ```python
    from typing_extensions import Annotated
    from encord.storage import StorageItem
    from encord_agents.gcp import editor_agent, Depends
    from encord_agents.gcp.dependencies import dep_storage_item


    @editor_agent()
    def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
        print("uuid", storage_item.uuid)
        print("client_metadata", storage_item.client_metadata)
        ...
    ```

    """
    return storage_item

dep_video_iterator 

dep_video_iterator(storage_item: StorageItem) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for performing operations over many frames.

Example:

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_video_iterator
...

@editor_agent()
def my_agent(
    video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
):
    for frame in video_frames:
        print(frame.frame, frame.content.shape)

Parameters:

  • storage_item (StorageItem) –

    Automatically injected storage item dependency.

Raises:

  • NotImplementedError –

    Fails for data types other than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/core/dependencies/serverless.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def dep_video_iterator(storage_item: StorageItem) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for performing operations over many frames.

    **Example:**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_video_iterator
    ...

    @editor_agent()
    def my_agent(
        video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
    ):
        for frame in video_frames:
            print(frame.frame, frame.content.shape)
    ```

    Args:
        storage_item: Automatically injected storage item dependency.

    Raises:
        NotImplementedError: Fails for data types other than video.

    Yields:
        An iterator.

    """
    from encord_agents.core.video import iter_video

    if not storage_item.item_type == StorageItemType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")

    with download_asset(storage_item, None) as asset:
        yield iter_video(asset)

encord_agents.gcp.wrappers

editor_agent 

editor_agent(*, label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None, custom_cors_regex: str | None = None) -> Callable[[AgentFunction], Callable[[Request], Response]]

Wrapper to make resources available for gcp editor agents.

The editor agents are intended to be used via dependency injections. You can learn more via out documentation.

Parameters:

  • label_row_metadata_include_args (LabelRowMetadataIncludeArgs | None, default: None ) –

    arguments to overwrite default arguments on project.list_label_rows_v2().

  • label_row_initialise_labels_args (LabelRowInitialiseLabelsArgs | None, default: None ) –

    Arguments to overwrite default arguments on label_row.initialise_labels(...)

  • custom_cors_regex (str | None, default: None ) –

    A regex to use for the CORS settings. If not provided, the default regex will be used. Only required if the agent is not deployed on Encord's platform.

Returns:

  • Callable[[AgentFunction], Callable[[Request], Response]] –

    A wrapped function suitable for gcp functions.

Source code in encord_agents/gcp/wrappers.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | None = None,
    custom_cors_regex: str | None = None,
) -> Callable[[AgentFunction], Callable[[Request], Response]]:
    """
    Wrapper to make resources available for gcp editor agents.

    The editor agents are intended to be used via dependency injections.
    You can learn more via out [documentation](https://agents-docs.encord.com).

    Args:
        label_row_metadata_include_args: arguments to overwrite default arguments
            on `project.list_label_rows_v2()`.
        label_row_initialise_labels_args: Arguments to overwrite default arguments
            on `label_row.initialise_labels(...)`
        custom_cors_regex: A regex to use for the CORS settings. If not provided, the default regex will be used.
            Only required if the agent is not deployed on Encord's platform.

    Returns:
        A wrapped function suitable for gcp functions.
    """

    def context_wrapper_inner(func: AgentFunction) -> Callable[[Request], Response]:
        dependant = get_dependant(func=func)
        cors_regex = re.compile(custom_cors_regex or ENCORD_DOMAIN_REGEX)

        @wraps(func)
        def wrapper(request: Request) -> Response:
            if request.method == "OPTIONS":
                response = make_response("")
                response.headers["Vary"] = "Origin"

                if not cors_regex.fullmatch(request.origin):
                    response.status_code = 403
                    return response

                headers = {
                    "Access-Control-Allow-Origin": request.origin,
                    "Access-Control-Allow-Methods": "POST",
                    "Access-Control-Allow-Headers": f"Content-Type, {EDITOR_TEST_REQUEST_HEADER}",
                    "Access-Control-Max-Age": "3600",
                }
                response.headers.update(headers)
                response.status_code = 204
                return response

            if request.headers.get(EDITOR_TEST_REQUEST_HEADER):
                logging.info("Editor test request")
                return _generate_response()
            if not request.is_json:
                raise Exception("Request should be JSON. Migrated over to new format")
            frame_data = FrameData.model_validate(request.get_json())
            logging.info(f"Request: {frame_data}")

            client = get_user_client()
            try:
                project = client.get_project(frame_data.project_hash)
            except AuthorisationError as err:
                response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                return response

            label_row: LabelRowV2 | None = None
            if dependant.needs_label_row:
                include_args = label_row_metadata_include_args or LabelRowMetadataIncludeArgs()
                init_args = label_row_initialise_labels_args or LabelRowInitialiseLabelsArgs()
                label_row = project.list_label_rows_v2(
                    data_hashes=[str(frame_data.data_hash)], **include_args.model_dump()
                )[0]
                label_row.initialise_labels(**init_args.model_dump())

            storage_item: StorageItem | None = None
            if dependant.needs_storage_item:
                if label_row is None:
                    label_row = project.list_label_rows_v2(data_hashes=[frame_data.data_hash])[0]
                assert label_row.backing_item_uuid, "This is a server response so guaranteed to have this"
                try:
                    storage_item = client.get_storage_item(label_row.backing_item_uuid)
                except AuthorisationError as err:
                    response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                    return response

            context = Context(project=project, label_row=label_row, frame_data=frame_data, storage_item=storage_item)
            result: Any | None | EditorAgentResponse = None
            with ExitStack() as stack:
                try:
                    # Solving dependencies can execute arbitrary code including fetch from Encord platform
                    # e.g: Get storage item which can throw error
                    dependencies = solve_dependencies(context=context, dependant=dependant, stack=stack)
                except AuthorisationError as err:
                    response = _generate_response(err.message, HTTPStatus.FORBIDDEN)
                    return response
                try:
                    result = func(**dependencies.values)
                    if isinstance(result, EditorAgentResponse):
                        response = _generate_response(result.model_dump_json(), HTTPStatus.OK)
                        return response
                except EncordEditorAgentException as exc:
                    response = _generate_response(to_jsonable_python(exc.json_response_body), HTTPStatus.BAD_REQUEST)
                    return response
            return _generate_response()

        return wrapper

    return context_wrapper_inner