GCP

wrappers

editor_agent

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 using dependency injections. You can learn more via out documentation. Arguments:

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 is used. Only required if the agent is not deployed on Encord’s platform.

Returns: A wrapped function suitable for gcp functions.

dependencies

FastAPI

cors

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

EncordCORSMiddleware Objects

class EncordCORSMiddleware(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: The CORS middleware allows POST requests from the Encord domain.

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

app = FastAPI()
app.add_middleware(EncordCORSMiddleware)

EncordTestHeaderMiddleware Objects

class EncordTestHeaderMiddleware(BaseHTTPMiddleware)

dispatch

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

Middleware to handle the X-Encord-Editor-Agent test header. Arguments: request (Request): call_next (RequestResponseEndpoint): Returns: Response

get_encord_app

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

Get a FastAPI app with the Encord middleware. Arguments:

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.

utils

verify_auth

def verify_auth() -> None

FastAPI lifecycle start hook to fail early if ssh key is missing. Example: This makes the server fail early if authentication is not set up.

from fastapi import FastAPI

app = FastAPI(
    on_startup=[verify_auth]

notifications

Receiving the task-ready notifications an agent stage sends. The verified notification is injected like any other dependency:

from typing import Annotated
from uuid import UUID

from fastapi import BackgroundTasks, Depends

from encord_agents.core.webhooks import TaskNotification
from encord_agents.fastapi import dep_task_notification, get_encord_app

app = get_encord_app()
SERVED_PROJECTS = {UUID("<project_hash>")}

@app.post("/encord/task-ready")
def task_ready(
    notification: Annotated[TaskNotification, Depends(dep_task_notification)],
    background: BackgroundTasks,
) -> None:
    if notification.project_hash not in SERVED_PROJECTS:
        return
    background.add_task(runner.run_stage, notification.stage_uuid, project_hash=notification.project_hash)

Note the check. A notification says where work is waiting; it does not say what this deployment is allowed to do, so project_hash is an input to check rather than an instruction to follow. Decide which projects a deployment serves, and act only on those. Returning None answers 200, which is what Encord expects; it records any other status as a failed delivery. Answer before doing the work, as above: Encord gives a response 180 seconds and retries on timeout, so a handler that drains the stage inline is re-notified while the first drain is still running. Requests that cannot be verified never reach the route. They are turned into responses by the handlers get_encord_app installs; on an app built by hand, call add_notification_handlers instead, or they surface as unhandled errors.

dep_task_notification

async def dep_task_notification(request: Request) -> TaskNotification

Verify that the request came from Encord and read it as an agent-stage notification. Reads the body as bytes, which is what the signature covers; a parsed and re-encoded body is an equivalent object with a different signature. Verification is authentication, not authorization: it shows the request came from Encord, not that project_hash names a project this deployment serves. Check that before doing any work — see the example above. The signing secret comes from ENCORD_WEBHOOK_SECRET. To pass one explicitly, or to widen the timestamp tolerance, call verify_and_parse_notification from a dependency of your own. Raises:

WebhookVerificationError - The request is not provably from Encord.
UnexpectedEventType - It is, but carries an event this receiver does not read.
UnsupportedNotificationVersion - It is on an envelope version this package cannot read.

dep_task_notification_with_args

def dep_task_notification_with_args(
    secret: str | None = None,
    tolerance_seconds: int = DEFAULT_TIMESTAMP_TOLERANCE_SECONDS
) -> Callable[[Request], Awaitable[TaskNotification]]

Build a notification dependency that does not read its secret from the environment. ENCORD_WEBHOOK_SECRET holds one secret. Use this where a single secret does not cover everything a deployment verifies: it gives one route the secret to verify against, rather than taking it from the environment. Example:

@app.post("/encord/task-ready")
def task_ready(
    notification: Annotated[
        TaskNotification,
        Depends(dep_task_notification_with_args(secret=os.environ["STAGE_WEBHOOK_SECRET"])),
    ],
) -> None:
    ...

Arguments:

secret - The signing secret this route verifies against. Read from ENCORD_WEBHOOK_SECRET when not given.
tolerance_seconds - How far the signed timestamp may be from now. Widen it only for a deployment whose clock cannot be kept closer than the default.

Returns: A dependency to pass to Depends.

add_notification_handlers

def add_notification_handlers(app: FastAPI) -> None

Install the exception handlers that turn notification outcomes into responses. get_encord_app calls this already. Use it on an app constructed by hand, so that a request which fails verification is refused rather than raised as an unhandled error. Arguments:

app - The FastAPI app serving the notification route.

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_client

def dep_client(
    trace_id: Annotated[str | None,
                        Depends(dep_trace_id)]) -> 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 is authenticated and ready to use.

dep_label_row_with_args

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:

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

Arguments:

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.

dep_label_row

def 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

Arguments:

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.

dep_storage_item

def 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 is authenticated and ready to use.
    print(storage_item.dicom_study_uid)
    print(storage_item.client_metadata)

dep_single_frame

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’s name had the following format: lr.data_hash.{suffix}. When the function has finished, the downloaded file is 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"

Arguments:

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.

dep_asset

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

Get a local file path to data asset temporarily stored till end of agent execution. This dependency fetches the underlying data asset based on a signed url. It temporarily stores the data on disk. Once the task is completed, the asset is 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: 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.

dep_video_iterator

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:

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)

Arguments:

storage_item - Automatically injected storage item dependency.

Raises:

NotImplementedError - Fails for other data types than video.

Yields: An iterator.

dep_project

def 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 is authenticated and ready to use.
    print(project.title)

Arguments: frame_data: client:

dep_data_lookup

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:

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

Arguments:

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

Returns: The (shared) lookup object.

dep_object_crops

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:

@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
    ...

Arguments:

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

Returns: A FastAPI dependency function that yields a list of InstanceCrop.

Get Started

Data

Fundamentals

Annotate & Review

Label and Model Evaluation

Users and Workspaces

Agents Reference Editor Agents

GCP

wrappers

editor_agent

dependencies

FastAPI

cors

EncordCORSMiddleware Objects

EncordTestHeaderMiddleware Objects

dispatch

get_encord_app

utils

verify_auth

notifications

dep_task_notification

dep_task_notification_with_args

add_notification_handlers

dependencies

dep_client

dep_label_row_with_args

dep_label_row

dep_storage_item

dep_single_frame

dep_asset

dep_video_iterator

dep_project

dep_data_lookup

dep_object_crops

​GCP

​wrappers

editor_agent

​dependencies

​FastAPI

​cors

​EncordCORSMiddleware Objects

​EncordTestHeaderMiddleware Objects

dispatch

get_encord_app

​utils

verify_auth

​notifications

dep_task_notification

dep_task_notification_with_args

add_notification_handlers

​dependencies

dep_client

dep_label_row_with_args

dep_label_row

dep_storage_item

dep_single_frame

dep_asset

dep_video_iterator

dep_project

dep_data_lookup

dep_object_crops

GCP

wrappers

dependencies

FastAPI

cors

EncordCORSMiddleware Objects

EncordTestHeaderMiddleware Objects

utils

notifications

dependencies