Core

vision

b64_encode_image

def b64_encode_image(img: NDArray[np.uint8],
                     format: Base64Formats = ".jpg") -> str

Encode an image to a base64 string. Arguments:

img - The image to encode. Expects [RGB] channels
format - The format of the image.

Returns: The base64 encoded image.

rich_columns

TaskSpeedColumn Objects

class TaskSpeedColumn(ProgressColumn)

Renders human readable transfer speed.

render

def render(task: Task) -> Text

Show data transfer speed.

data_model

LabelRowMetadataIncludeArgs Objects

class LabelRowMetadataIncludeArgs(BaseModel)

Warning, including metadata via label rows is good for reading metadata not for writing to the metadata. If you need to write to metadata, use the dep_storage_item dependencies instead.

LabelRowInitialiseLabelsArgs Objects

class LabelRowInitialiseLabelsArgs(BaseModel)

Arguments used to specify how to initialise labels via the SDK. The arguments are passed to LabelRowV2.initialise_labels.

FrameData Objects

class FrameData(BaseModel)

Holds the data sent from the Encord Label Editor at the time of triggering the agent.

project_hash

The identifier of the given project.

data_hash

The identifier of the given data asset.

frame

The frame number. If single image, it’s default 0.

object_hashes

Object hashes if the request was made on particular objects from the App

Frame Objects

@dataclass(frozen=True)
class Frame()

A dataclass to hold the content of one frame in a video.

frame

The frame number within the video

content

An [h,w,c] np.array with color channels RGB.

b64_encoding

def b64_encoding(
    image_format: Literal[".jpeg", ".jpg", ".png"] = ".jpeg",
    output_format: Literal["url", "openai", "anthropic", "raw"] = "url"
) -> str | dict[str, str | dict[str, str]]

Get a base64 representation of the image content. This method allows you to convert the content into a base64 representation based on various different image encodings. This is useful, e.g., for prompting LLMs with image content. Please see details for formats below. Arguments:

image_format - Which type of image encoding to use.
output_format - Different common formats.
- raw: the image content as a raw b64 string
- url: url encoded image content. Compatible with, e.g., <img src="<the_encoding>" />
- openai: a dict with type and image_url keys _ anthropic: a dict with media_type, type, and data keys.
Returns - a dict or string depending on output_format.

InstanceCrop Objects

@dataclass(frozen=True)
class InstanceCrop(Frame)

A dataclass to hold the frame content of one object instance in a video or image.

instance

The ObjectInstance associated to the crop.

EditorAgentResponse Objects

class EditorAgentResponse(BaseModel)

A base class for all return types of editor agent functions.

message

A message to be displayed to the user. Purely informational: it never affects routing or labels. Encord truncates messages longer than 256 characters.

decision

The name of the workflow pathway that the task should follow. Only honored for workflow-triggered agents. An agent invoked from the Label Editor has no task to route, so Encord ignores the field there. The value must match one of the agent stage’s configured pathway names. An unmatched name fails the execution rather than falling back, and the name is matched verbatim — a pathway UUID is not accepted. When omitted, the stage’s default pathway is used, or its only pathway if it has exactly one.

utils

get_user_client

def get_user_client(settings: Settings | None = None,
                    *,
                    trace_id: str | None = None) -> EncordUserClient

Generate an user client to access Encord. Returns: An EncordUserClient authenticated with the credentials from the encord_agents.core.settings.Settings.

get_initialised_label_row

def get_initialised_label_row(
        frame_data: FrameData,
        include_args: LabelRowMetadataIncludeArgs | None = None,
        init_args: LabelRowInitialiseLabelsArgs | None = None) -> LabelRowV2

Get an initialized label row from the frame_data information. Arguments:

frame_data - The data pointing to the data asset.

Raises:

Exception - If the frame_data cannot be matched to a label row

Returns: The initialized label row.

download_asset

@contextmanager
def download_asset(storage_item: StorageItem,
                   frame: int | None = None) -> Generator[Path, None, None]

Download the asset associated to a label row to disk. This function is a context manager. Data is cleaned up when the context is left. Example usage: with download_asset(storage_item, 10) as asset_path:

In here the file exists

pixel_values = np.asarray(Image.open(asset_path)) Arguments:

storage_item - The Storage item for which you want to download the associated asset.
frame - The frame that you need. If frame is none for a video, you the video path is returned.

Raises:

NotImplementedError - If you try to get all frames of an image group.
ValueError - If you try to download an unsupported data type (e.g., DICOM).

Yields: The file path for the requested asset.

get_frame_count

def get_frame_count(storage_item: StorageItem) -> int

Get the number of frames in a video.

batch_iterator

def batch_iterator(iterator: Iterable[T],
                   batch_size: int) -> Iterable[List[T]]

Yield batches of items from an iterator. Arguments:

iterator - The source iterator
batch_size - Size of each batch > 0

Returns: Iterable of lists, each containing up to batch_size items

ontology

GenericFieldModel Objects

class GenericFieldModel(BaseModel)

set_answer

def set_answer(instance: ClassificationInstance) -> None

This function is called from the parsing loop to allow the model to set it self as answer on the classification instance.

FieldType

Field from pydantic can be anything so hard to type. This is supposed to indicate that you should use the pydantic.Field function to construct this var.

OntologyDataModel Objects

class OntologyDataModel(Generic[OntologyType])

Class to create a pydantic model equivalent to an arbitrary classification ontology. The model can be used to form a json schema based on the ontology. This is useful if you are, e.g., trying to get a structured response from an LLM. Example:

from pydantic import ValidationError

classifications = project.ontology_structure.classifications
objects = project.ontology_structure.classifications

data_model = OntologyDataModel([objects])
## or
data_model = OntologyDataModel([classifications])

## Get a json schema for the ontology
print(data_model.model_json_schema_str)

## Parse json following the schema into label instances
json_str = my_favourite_llm(
    f"what is this? pls follow {schema}", img
)
try:
    instances = data_model(json_str)
except ValidationError:
    # invalid json
    ...

for ins in instances:
    label_row.add_classification_instance(ins)

label_row.save()

Attributes: ontology: DataModel:

call

def __call__(answer: str) -> list[ClassificationInstance] | ObjectInstance

Validate a json response in accordance to the pydantic model. This function allows you to convert from a json object (e.g., coming from an llm) back to the encord “instance format”. Arguments:

answer - The json object as a raw string.
Returns - a list of classification / object instances that must be added to a label row.

validate_json

def validate_json(
        answer_str: str) -> list[ClassificationInstance] | ObjectInstance

answer_str - The json object as a raw string.
Returns - a list of classification / object instances that must be added to a label row.

settings

Settings used throughout the module. Note that central settings are read using environment variables.

Settings Objects

class Settings(BaseSettings)

ssh_key_file

The path to the private ssh key file to authenticate with Encord. Either this or the ENCORD_SSH_KEY needs to be set for most use-cases. To setup a key with Encord, see the platform docs.

ssh_key_content

The content of the private ssh key file to authenticate with Encord. Either this or the ENCORD_SSH_KEY needs to be set for most use-cases. To setup a key with Encord, see the platform docs.

WebhookSettings Objects

class WebhookSettings(BaseSettings)

Settings for verifying the requests Encord signs. Deliberately separate from Settings: checking that a request came from Encord is something an endpoint should be able to do before — or without ever — holding Encord credentials.

webhook_secret

The signing secret this deployment verifies incoming requests with. Encord shows it in the app alongside the endpoint’s configuration; check it again if you change that configuration. Read only when the secret is not passed explicitly. Use this when a single secret covers everything this deployment verifies. Otherwise, pass the secret explicitly. Verifying and reading the requests Encord signs. Encord signs every request it makes to a configured URL: the task-ready notifications an agent stage sends, and the calls it makes to a custom agent endpoint. Those URLs have to be publicly reachable, so the signature is what separates a genuine request from anything else that finds them. Verification needs the signing secret, which Encord shows in the app alongside the endpoint’s configuration. Set it as ENCORD_WEBHOOK_SECRET, or pass it explicitly, and check it again if you change that configuration.

DEFAULT_TIMESTAMP_TOLERANCE_SECONDS

How far the signed timestamp may be from now before a request is refused as a replay.

AGENT_STAGE_WORK_AVAILABLE_EVENT

The event Encord sends when tasks are waiting at an agent stage.

SUPPORTED_NOTIFICATION_VERSION

The envelope version this module reads. Encord bumps it only for a change that breaks the previous shape; fields and enum values are added without one. So an envelope on this version is safe to read even if it carries things this module has never heard of, and one on another version is not.

WebhookVerificationError Objects

class WebhookVerificationError(Exception)

A request could not be shown to have come from Encord. Treat it as hostile: do not read the body, and answer with a 4xx.

UnexpectedEventType Objects

class UnexpectedEventType(Exception)

The request is genuine, but carries an event this module does not read. A single URL can be configured on more than one workflow stage, so a receiver written for agent-stage notifications can legitimately be sent, say, a task_submitted_event. Acknowledge those rather than failing on them.

UnsupportedNotificationVersion Objects

class UnsupportedNotificationVersion(Exception)

The request is genuine, but its envelope is a version this module cannot read. Encord reserves a version bump for a breaking change, so the payload cannot be read as the version below. Upgrade encord-agents.

AgentStageWorkReason Objects

class AgentStageWorkReason(str, Enum)

Which condition made Encord send the notification.

BATCH_SIZE_REACHED

The number of queued tasks reached the stage’s minimum batch size.

MAX_WAIT_ELAPSED

A task had been queued longer than the stage’s maximum wait.

StageWorkAvailable Objects

class StageWorkAvailable(BaseModel)

What the notification says about the stage.

pending_count

How many tasks were queued when the notification was raised. A hint, not a promise: fetch the queue to find what is actually there now.

reason

Which condition fired. A condition added platform-side arrives as a plain string rather than failing validation, since Encord adds those without bumping the envelope version. Compare against AgentStageWorkReason for the ones known here — and to tell a known one from a new one, use isinstance(reason, AgentStageWorkReason), since the enum subclasses str and so is a str either way.

TaskNotification Objects

class TaskNotification(BaseModel)

Tasks are waiting at an agent stage. Carries no task data. Act on it by fetching that stage’s queue, which may by then hold more, fewer or none of the tasks the notification counted.

uid

Identifies this delivery. Not a deduplication key: Encord recomputes the state every cycle and re-notifies with a fresh uid while work remains.

verify_signature

def verify_signature(
        body: bytes,
        *,
        signature: str | None,
        timestamp: str | None,
        secret: str | None = None,
        tolerance_seconds: int = DEFAULT_TIMESTAMP_TOLERANCE_SECONDS) -> None

Check that body carries a signature only Encord could have produced. Arguments:

body - The request body exactly as received. Re-serializing it first will not
verify - the signature covers the bytes that were sent.
signature - The X-Encord-Signature header, if present.
timestamp - The X-Encord-Timestamp header, if present.
secret - The signing secret to verify against. Read from ENCORD_WEBHOOK_SECRET when not given.
tolerance_seconds - How far from now the signed timestamp may be.

Raises:

WebhookVerificationError - If the headers are missing or malformed, the timestamp is outside the tolerance, or the signature does not match.
PrintableError - If no secret was given or configured.

parse_notification

def parse_notification(body: bytes | str) -> TaskNotification

Read a verified body as an agent-stage notification. Verify first: this trusts what it is given. Raises:

UnexpectedEventType - If the body is a different Encord event.
UnsupportedNotificationVersion - If it is that event on a later envelope version.
pydantic.ValidationError - If it is that event but does not match the model.

verify_and_parse_notification

def verify_and_parse_notification(
    body: bytes,
    headers: Mapping[str, str],
    *,
    secret: str | None = None,
    tolerance_seconds: int = DEFAULT_TIMESTAMP_TOLERANCE_SECONDS
) -> TaskNotification

Verify a request and read it as an agent-stage notification. Raises:

WebhookVerificationError - If the request cannot be shown to be from Encord.
UnexpectedEventType - If it is genuine but carries a different event.
UnsupportedNotificationVersion - If it is on a later envelope version.
PrintableError - If no secret was given or configured.

video

get_frame

def get_frame(video_path: Path, desired_frame: int) -> NDArray[np.uint8]

Extract an exact frame from a video. Arguments:

video_path - The file path to where the video is stored.
desired_frame - The frame to extract

Raises:

Exception - If the video cannot be opened properly or the requested frame could not be retrieved from the video.

Returns: Numpy array of shape [h, w, c] where channels are BGR.

write_frame

def write_frame(frame_path: Path, frame: NDArray[np.uint8]) -> None

Write a frame to a file. Arguments:

frame_path - The file path to write the frame to.
frame - The frame to write.

iter_video

def iter_video(video_path: Path) -> Iterator[Frame]

Iterate video frame by frame. Arguments:

video_path - The file path to the video you wish to iterate.

Raises:

Exception - If the video file could not be opened properly.

Yields: Frames from the video.

iter_video_with_indices

def iter_video_with_indices(video_path: Path,
                            frame_indices: Iterable[int]) -> Iterator[Frame]

Iterate video frame by frame with specified frame indices. Arguments:

video_path - The file path to the video you wish to iterate.
frame_indices - The frame indices to iterate over.

Yields: Frames from the video.

dependencies.serverless

This module defines dependencies available for injection within serverless Editor Agents. These dependencies can be used independently, even when reliant on other dependencies. Note: The injection mechanism necessitates the presence of type annotations for the following parameters to ensure proper resolution.

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.

from encord.project import Project
from encord.objects.ontology_labels_impl import LabelRowV2
from encord_agents import FrameData
...
@editor_agent()
def my_agent(
    frame_data: FrameData,
    project: Project,
    label_row: LabelRowV2,
):
    ...

dep_client

def 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 is authenticated and ready to use.
    client.get_dataset("")

dep_single_frame

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’s name has the following format: 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"

Arguments:

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

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

dep_asset

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:

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.

dep_video_iterator

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:

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)

Arguments:

storage_item - Automatically injected storage item dependency.

Raises:

NotImplementedError - Fails for data types other than video.

Yields: An iterator.

dep_data_lookup

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

Returns a lookup for easily retrieving 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.core.dependencies.serverless import dep_data_lookup, DataLookup

@editor_agent()
def my_agent(
    frame_data: FrameData,
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
):
    storage_item = lookup.get_storage_item(frame_data.data_hash)
    ...

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

@editor_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_storage_item

def 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)
    ...

dep_object_crops

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:

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

Arguments:

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.

DEncordClient

Get an authenticated user client.

DObjectsInstances

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

DObjectCrops

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.

DSingleFrame

Get the single frame that the agent was triggered on.

DAssetPath

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

DVideoIterator

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

DStorageItem

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

dependencies.shares

DataLookup Objects

class DataLookup()

!!! warning “Deprecated” DataLookup is deprecated and will be removed in version 0.2.10. Migration Guide:

For accessing storage items, use dep_storage_item instead:

# Old way (deprecated)
from encord_agents.core.dependencies.shares import DataLookup
lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
storage_item = lookup.get_storage_item(data_hash)

# New way (recommended)
from encord_agents.tasks.dependencies import dep_storage_item
# or from encord_agents.aws.dependencies import dep_storage_item
# or from encord_agents.gcp.dependencies import dep_storage_item
# or from encord_agents.fastapi.dependencies import dep_storage_item
storage_item: Annotated[StorageItem, Depends(dep_storage_item)]

backing_item_uuids

@property
def backing_item_uuids() -> list[UUID]

Get all backing item uuids for all data rows in the data lookup. !!! warning “Deprecated” This property is deprecated and will be removed in version 0.2.10. Use the EncordUserClient directly to access backing item UUIDs from label rows.

get_storage_item

def get_storage_item(data_hash: str | UUID,
                     dataset_hash: str | UUID | None = None,
                     sign_url: bool = False) -> StorageItem

!!! warning “Deprecated” This method is deprecated and will be removed in version 0.2.10. Use dep_storage_item dependency instead. Arguments:

data_hash - Data hash for the asset for which you need the underlying storage item.
dataset_hash - If you didn’t provide the associated dataset hash in the constructor, this is your last chance.
sign_url - If True, pre-fetch a signed URLs for the items (otherwise the URLs will be signed on demand).

Raises:

ValueError - Mainly if underlying data row cannot be found.

Returns: The underlying storage item from which, e.g., client metadata can be updated.

get_storage_items

def get_storage_items(data_hashes: list[str | UUID],
                      dataset_hash: str | UUID | None = None,
                      sign_urls: bool = False) -> list[StorageItem]

!!! warning “Deprecated” This method is deprecated and will be removed in version 0.2.10. Use the EncordUserClient directly for bulk storage item access. Arguments:

data_hashes - Data hashes for the assets for which you need the underlying storage items.
dataset_hash - If you didn’t provided the associated dataset hash in the constructor, this is your last chance.
sign_urls - If True, pre-fetch a signed URLs for the items (otherwise the URLs will be signed on demand).

Raises:

ValueError - Mainly if underlying data row cannot be found.

Returns: list of underlying storage items from which, e.g., client metadata can be updated.

​Core

​vision

b64_encode_image

​rich_columns

​TaskSpeedColumn Objects

render

​data_model

​LabelRowMetadataIncludeArgs Objects

​LabelRowInitialiseLabelsArgs Objects

​FrameData Objects

project_hash

data_hash

frame

object_hashes

​Frame Objects

frame

content

b64_encoding

​InstanceCrop Objects

instance

​EditorAgentResponse Objects

message

decision

​utils

get_user_client

get_initialised_label_row

download_asset

​In here the file exists

get_frame_count

batch_iterator

​ontology

​GenericFieldModel Objects

set_answer

FieldType

​OntologyDataModel Objects

__call__

validate_json

​settings

​Settings Objects

ssh_key_file

ssh_key_content

​WebhookSettings Objects

webhook_secret

DEFAULT_TIMESTAMP_TOLERANCE_SECONDS

AGENT_STAGE_WORK_AVAILABLE_EVENT

SUPPORTED_NOTIFICATION_VERSION

​WebhookVerificationError Objects

​UnexpectedEventType Objects

​UnsupportedNotificationVersion Objects

​AgentStageWorkReason Objects

BATCH_SIZE_REACHED

MAX_WAIT_ELAPSED

​StageWorkAvailable Objects

pending_count

reason

​TaskNotification Objects

uid

verify_signature

parse_notification

verify_and_parse_notification

​video

get_frame

write_frame

iter_video

iter_video_with_indices

​dependencies.serverless

dep_client

dep_single_frame

dep_asset

dep_video_iterator

dep_data_lookup

dep_storage_item

dep_object_crops

DEncordClient

DObjectsInstances

DObjectCrops

DSingleFrame

DAssetPath

DVideoIterator

DStorageItem

Core

vision

rich_columns

TaskSpeedColumn Objects

data_model

LabelRowMetadataIncludeArgs Objects

LabelRowInitialiseLabelsArgs Objects

FrameData Objects

Frame Objects

InstanceCrop Objects

EditorAgentResponse Objects

utils

In here the file exists

ontology

GenericFieldModel Objects

OntologyDataModel Objects

call

settings

Settings Objects

WebhookSettings Objects

WebhookVerificationError Objects

UnexpectedEventType Objects

UnsupportedNotificationVersion Objects

AgentStageWorkReason Objects

StageWorkAvailable Objects

TaskNotification Objects

video

dependencies.serverless

dependencies.shares

DataLookup Objects