Skip to content

Editor Agents Reference

GCP

encord_agents.gcp.editor_agent 

editor_agent(*, label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None, label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | 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(...)

Returns:

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

    A wrapped function suitable for gcp functions.

Source code in encord_agents/gcp/wrappers.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
 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
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | 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(...)`

    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(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": "Content-Type",
                    "Access-Control-Max-Age": "3600",
                }
                response.headers.update(headers)
                response.status_code = 204
                return response

            # TODO: We'll remove FF from FE on Jan. 31 2025.
            #   At that point, only the if statement applies and the else should be removed.
            if request.is_json:
                frame_data = FrameData.model_validate(request.get_json())
            else:
                frame_data = FrameData.model_validate_json(request.get_data())
            logging.info(f"Request: {frame_data}")

            client = get_user_client()
            project = client.get_project(str(frame_data.project_hash))

            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())

            context = Context(project=project, label_row=label_row, frame_data=frame_data)
            with ExitStack() as stack:
                dependencies = solve_dependencies(context=context, dependant=dependant, stack=stack)
                func(**dependencies.values)
            return generate_response()

        return wrapper

    return context_wrapper_inner

encord_agents.gcp.dependencies

Dependencies for injection in GCP Cloud run functions.

This module contains dependencies that you can inject within your cloud functions. 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 following typed parameters. If the type annotations are not present, the injection mechanism cannot resolve the them:

from encord.project import Project
from encord.objects.ontology_labels_impl import LabelRowV2
from encord_agents import FrameData
...
@app.post("/my-agent-route")
def my_agent(
    frame_data: FrameData,
    project: Project,
    label_row: LabelRowV2,
):
    ...
  • FrameData is automatically injected via the api request body.
  • Project is automatically loaded based on the frame data.
  • label_row_v2 is automatically loaded based on the frame data.

dep_asset 

dep_asset(lr: LabelRowV2) -> 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.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.

Source code in encord_agents/gcp/dependencies.py 
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
def dep_asset(lr: LabelRowV2) -> 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.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(lr) 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/gcp/dependencies.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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_data_lookup 

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

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

Info

If you're just looking to get the associated storage item to a task, consider using dep_storage_item instead.

The lookup can, e.g., be useful for

  • Updating client metadata
  • Downloading data from signed urls
  • Matching data to other projects

Example:

from typing_extensions import Annotated
from encord.storage import StorageItem
from encord_agents import FrameData
from encord_agents.gcp import editor_agent, Depends
from encord_agents.gcp.dependencies import DataLookup, dep_data_lookup

@editor_agent()
def my_agent(
    frame_data: FrameData,
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
):
    print("data hash", lookup.get_data_row(frame_data.data_hash))
    print("storage item", lookup.get_storage_item(frame_data.data_hash))
    ...

Parameters:

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

    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/gcp/dependencies.py 
182
183
184
185
186
187
188
189
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
def dep_data_lookup(lookup: Annotated[DataLookup, Depends(DataLookup.sharable)]) -> DataLookup:
    """
    Get a lookup to easily retrieve data rows and storage items associated with the given task.

    !!! info
        If you're just looking to get the associated storage item to a task, consider using `dep_storage_item` instead.


    The lookup can, e.g., be useful for

    * Updating client metadata
    * Downloading data from signed urls
    * Matching data to other projects

    **Example:**

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

    @editor_agent()
    def my_agent(
        frame_data: FrameData,
        lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
    ):
        print("data hash", lookup.get_data_row(frame_data.data_hash))
        print("storage item", lookup.get_storage_item(frame_data.data_hash))
        ...


    ```


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

    Returns:
        The (shared) lookup object.

    """
    return lookup

dep_object_crops 

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

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

Useful, e.g., to be able 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 will be included. Strings are matched against feature_node_hashes.

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

Source code in encord_agents/gcp/dependencies.py 
261
262
263
264
265
266
267
268
269
270
271
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
306
307
308
309
310
def dep_object_crops(
    filter_ontology_objects: list[Object | str] | None = None,
) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]:
    """
    Get a list of object instances and frame crops associated with each object.

    Useful, e.g., to be able 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 will be included.
            Strings are matched against `feature_node_hashes`.


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

    """
    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)
        ]

    return _dep_object_crops

dep_single_frame 

dep_single_frame(lr: LabelRowV2) -> 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, the downloaded file will be 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:

  • lr (LabelRowV2) –

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

Returns:

  • NDArray[uint8] –

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

Source code in encord_agents/gcp/dependencies.py 
 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
def dep_single_frame(lr: LabelRowV2) -> 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, the downloaded file will be 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:
        lr: The label row. Automatically injected (see example above).

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

    """
    with download_asset(lr, frame=0) 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(lookup: Annotated[DataLookup, Depends(dep_data_lookup)], frame_data: FrameData) -> 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/gcp/dependencies.py 
228
229
230
231
232
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
def dep_storage_item(
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)],
    frame_data: FrameData,
) -> 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 lookup.get_storage_item(frame_data.data_hash)

dep_video_iterator 

dep_video_iterator(lr: LabelRowV2) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for doing things 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:

  • lr (LabelRowV2) –

    Automatically injected label row dependency.

Raises:

  • NotImplementedError –

    Will fail for other data types than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/gcp/dependencies.py 
145
146
147
148
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
def dep_video_iterator(lr: LabelRowV2) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for doing things 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:
        lr: Automatically injected label row dependency.

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

    Yields:
        An iterator.

    """
    if not lr.data_type == DataType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")

    with download_asset(lr, 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) -> 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(...)

Returns:

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

    A wrapped function suitable for gcp functions.

Source code in encord_agents/gcp/wrappers.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
 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
def editor_agent(
    *,
    label_row_metadata_include_args: LabelRowMetadataIncludeArgs | None = None,
    label_row_initialise_labels_args: LabelRowInitialiseLabelsArgs | 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(...)`

    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(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": "Content-Type",
                    "Access-Control-Max-Age": "3600",
                }
                response.headers.update(headers)
                response.status_code = 204
                return response

            # TODO: We'll remove FF from FE on Jan. 31 2025.
            #   At that point, only the if statement applies and the else should be removed.
            if request.is_json:
                frame_data = FrameData.model_validate(request.get_json())
            else:
                frame_data = FrameData.model_validate_json(request.get_data())
            logging.info(f"Request: {frame_data}")

            client = get_user_client()
            project = client.get_project(str(frame_data.project_hash))

            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())

            context = Context(project=project, label_row=label_row, frame_data=frame_data)
            with ExitStack() as stack:
                dependencies = solve_dependencies(context=context, dependant=dependant, stack=stack)
                func(**dependencies.values)
            return generate_response()

        return wrapper

    return context_wrapper_inner

generate_response 

generate_response() -> Response

Generate a Response object with status 200 in order to tell the FE that the function has finished successfully. :return: Response object with the right CORS settings.

Source code in encord_agents/gcp/wrappers.py 
21
22
23
24
25
26
27
28
def generate_response() -> Response:
    """
    Generate a Response object with status 200 in order to tell the FE that the function has finished successfully.
    :return: Response object with the right CORS settings.
    """
    response = make_response("")
    response.headers["Access-Control-Allow-Origin"] = "*"
    return response

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.depencencies 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.depencencies 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.depencencies 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.depencencies 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(lr: Annotated[LabelRowV2, Depends(dep_label_row)], 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.depencencies 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:

  • lr (Annotated[LabelRowV2, Depends(dep_label_row)]) –

    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 
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_single_frame(lr: Annotated[LabelRowV2, Depends(dep_label_row)], 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.depencencies 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:
        lr: 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.

    """
    with download_asset(lr, 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.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.midleware.cors.CORSMiddleware but matches against the Encord origin by default.

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 
23
24
25
26
27
28
29
30
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
class EncordCORSMiddleware(CORSMiddleware):  # type: ignore [misc]
    """
    Like a regular `fastapi.midleware.cors.CORSMiddleware` but matches against
    the Encord origin by default.

    **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] = (),
        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,
        )

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(lr: Annotated[LabelRowV2, Depends(dep_label_row_with_args(label_row_initialise_labels_args=LabelRowInitialiseLabelsArgs(include_signed_url=True)))]) -> 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.depencencies 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.

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_asset(
    lr: Annotated[
        LabelRowV2,
        Depends(
            dep_label_row_with_args(
                label_row_initialise_labels_args=LabelRowInitialiseLabelsArgs(include_signed_url=True)
            )
        ),
    ],
) -> 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.depencencies 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(lr) 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.depencencies 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.depencencies 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.

Info

If you're just looking to get the associated storage item to a task, consider using dep_storage_item instead.

The lookup can, e.g., be useful for

  • Updating client metadata
  • Downloading data from signed urls
  • Matching data to other projects

Example:

from fastapi import Form
from typing_extensions import Annotated
from encord_agents import FrameData
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)]
):
    # Client will authenticated and ready to use.
    print(lookup.get_data_row(frame_data.data_hash).title)
    print(lookup.get_storage_item(frame_data.data_hash).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 
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
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
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.

    !!! info
        If you're just looking to get the associated storage item to a task, consider using `dep_storage_item` instead.


    The lookup can, e.g., be useful for

    * Updating client metadata
    * Downloading data from signed urls
    * Matching data to other projects

    **Example:**

    ```python
    from fastapi import Form
    from typing_extensions import Annotated
    from encord_agents import FrameData
    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)]
    ):
        # Client will authenticated and ready to use.
        print(lookup.get_data_row(frame_data.data_hash).title)
        print(lookup.get_storage_item(frame_data.data_hash).client_metadata)
    ```

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

    Returns:
        The (shared) lookup object.

    """
    return lookup

dep_label_row 

dep_label_row(frame_data: FrameData) -> LabelRowV2

Dependency to provide an initialized label row.

Example:

from encord_agents.fastapi.depencencies 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.depencencies 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.depencencies 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:

  • 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:

  • 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.depencencies 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:
        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.

    """

    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 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
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
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)
        ]

    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.depencencies 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 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
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.depencencies 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(lr: Annotated[LabelRowV2, Depends(dep_label_row)], 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.depencencies 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:

  • lr (Annotated[LabelRowV2, Depends(dep_label_row)]) –

    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 
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_single_frame(lr: Annotated[LabelRowV2, Depends(dep_label_row)], 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.depencencies 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:
        lr: 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.

    """
    with download_asset(lr, 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(lookup: Annotated[DataLookup, Depends(dep_data_lookup)], frame_data: FrameData) -> 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 
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
def dep_storage_item(
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)],
    frame_data: FrameData,
) -> 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)
    ```

    """
    return lookup.get_storage_item(frame_data.data_hash)

dep_video_iterator 

dep_video_iterator(lr: Annotated[LabelRowV2, Depends(dep_label_row)]) -> Generator[Iterator[Frame], None, None]

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

Example:

from encord_agents.fastapi.depencencies 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:

  • lr (Annotated[LabelRowV2, Depends(dep_label_row)]) –

    Automatically injected label row dependency.

Raises:

  • NotImplementedError –

    Will fail for other data types than video.

Yields:

  • Iterator[Frame] –

    An iterator.

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
def dep_video_iterator(lr: Annotated[LabelRowV2, Depends(dep_label_row)]) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for doing things over many frames.

    **Example:**

    ```python
    from encord_agents.fastapi.depencencies 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:
        lr: Automatically injected label row dependency.

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

    Yields:
        An iterator.

    """
    if not lr.data_type == DataType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")
    with download_asset(lr, 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}"
        )