For step-by-step examples, see the Annotation Queue Using SDK guide.

AnnotationQueue Class

The AnnotationQueue class is the SDK client for managing annotation queues, items, scores, and analytics. Annotation queues let you organize traces, sessions, datasets, and simulation outputs for structured human review. You can define custom labels, set how many annotations are needed per item, and add guidelines to keep feedback consistent.

Installation

pip install futureagi

Initialization

from fi.queues import AnnotationQueue

client = AnnotationQueue(
    fi_api_key="your_api_key",
    fi_secret_key="your_secret_key",
    fi_base_url="https://api.futureagi.com",  # optional
)

Arguments:

• fi_api_key (Optional[str]): API key for authentication.
• fi_secret_key (Optional[str]): Secret key for authentication.
• fi_base_url (Optional[str]): Base URL for the API.

Labels

create_label

Creates an annotation label. Labels define what annotators evaluate (e.g. sentiment, quality, relevance).

def create_label(
    self,
    name: str,
    type: str,
    *,
    settings: Optional[Dict[str, Any]] = None,
    description: Optional[str] = None,
    project: Optional[str] = None,
    timeout: Optional[int] = None,
) -> AnnotationLabel

• Arguments:
  • name (str): Label name. Must be unique per organization, type, and project.
  • type (str): Label type — "categorical", "text", "numeric", "star", or "thumbs_up_down".
  • settings (Optional[Dict[str, Any]]): Type-specific configuration. See Label Settings by Type below.
  • description (Optional[str]): Description of the label.
  • project (Optional[str]): Project ID to scope the label to. If omitted, the label is organization-wide.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • AnnotationLabel instance

Label Settings by Type

{
    "rule_prompt": "Classify the sentiment",       # str, required
    "multi_choice": False,                          # bool, required
    "options": [                                    # list, required (min 2)
        {"label": "Positive"},
        {"label": "Negative"},
        {"label": "Neutral"},
    ],
    "auto_annotate": False,                         # bool, required
    "strategy": None,                               # "Rag" or None, required
}

{
    "placeholder": "Enter your feedback...",        # str, required
    "max_length": 500,                              # int, required
    "min_length": 1,                                # int, required
}

{
    "min": 0,                                       # number, required
    "max": 10,                                      # number, required
    "step_size": 1,                                 # number, required
    "display_type": "slider",                       # "slider" or "button", required
}

{
    "no_of_stars": 5,                               # int, required (>= 1)
}

{}  # No settings required

list_labels

Lists annotation labels available to the organization.

def list_labels(
    self,
    *,
    project_id: Optional[str] = None,
    timeout: Optional[int] = None,
) -> List[AnnotationLabel]

• Arguments:
  • project_id (Optional[str]): Filter labels by project ID.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • List[AnnotationLabel]

get_label

Gets a single annotation label by ID or name.

def get_label(
    self,
    label_id: Optional[str] = None,
    *,
    label_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> AnnotationLabel

• Arguments:
  • label_id (Optional[str]): UUID of the annotation label.
  • label_name (Optional[str]): Name of the annotation label (alternative to label_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • AnnotationLabel instance

delete_label

Deletes an annotation label.

def delete_label(
    self,
    label_id: Optional[str] = None,
    *,
    label_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • label_id (Optional[str]): UUID of the annotation label.
  • label_name (Optional[str]): Name of the annotation label (alternative to label_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

add_label

Attaches an existing annotation label to the queue.

def add_label(
    self,
    queue_id: Optional[str] = None,
    label_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    label_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • label_id (Optional[str]): UUID of the annotation label.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • label_name (Optional[str]): Name of the annotation label (alternative to label_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

remove_label

Removes an annotation label from the queue.

def remove_label(
    self,
    queue_id: Optional[str] = None,
    label_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    label_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • label_id (Optional[str]): UUID of the annotation label.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • label_name (Optional[str]): Name of the annotation label (alternative to label_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

Queue Management

create

Creates a new annotation queue.

def create(
    self,
    name: str,
    *,
    description: Optional[str] = None,
    instructions: Optional[str] = None,
    assignment_strategy: Optional[str] = None,
    annotations_required: Optional[int] = None,
    reservation_timeout_minutes: Optional[int] = None,
    requires_review: Optional[bool] = None,
    project: Optional[str] = None,
    dataset: Optional[str] = None,
    agent_definition: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueDetail

• Arguments:
  • name (str): Name of the annotation queue.
  • description (Optional[str]): Description of the queue’s purpose.
  • instructions (Optional[str]): Guidelines for annotators.
  • assignment_strategy (Optional[str]): How items are assigned — "manual", "round_robin", or "load_balanced".
  • annotations_required (Optional[int]): Number of annotations needed per item.
  • reservation_timeout_minutes (Optional[int]): Time limit (in minutes) for an annotator to complete an item.
  • requires_review (Optional[bool]): Whether completed annotations require reviewer approval.
  • project (Optional[str]): Project ID to scope the queue to.
  • dataset (Optional[str]): Dataset ID to associate with the queue.
  • agent_definition (Optional[str]): Agent definition ID to associate with the queue.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueDetail instance

list_queues

Lists annotation queues with optional filters.

def list_queues(
    self,
    *,
    status: Optional[str] = None,
    search: Optional[str] = None,
    include_counts: bool = True,
    page: int = 1,
    page_size: int = 20,
    timeout: Optional[int] = None,
) -> List[QueueDetail]

• Arguments:
  • status (Optional[str]): Filter by queue status — "draft", "active", "paused", or "completed".
  • search (Optional[str]): Search queues by name.
  • include_counts (bool): Whether to include item/completed counts. Defaults to True.
  • page (int): Page number for pagination. Defaults to 1.
  • page_size (int): Number of results per page. Defaults to 20.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • List[QueueDetail]

get

Gets a single annotation queue by ID or name.

def get(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueDetail

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueDetail instance

update

Updates an annotation queue.

def update(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    instructions: Optional[str] = None,
    assignment_strategy: Optional[str] = None,
    annotations_required: Optional[int] = None,
    reservation_timeout_minutes: Optional[int] = None,
    requires_review: Optional[bool] = None,
    timeout: Optional[int] = None,
) -> QueueDetail

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • name (Optional[str]): Updated queue name.
  • description (Optional[str]): Updated description.
  • instructions (Optional[str]): Updated annotator instructions.
  • assignment_strategy (Optional[str]): Updated assignment strategy.
  • annotations_required (Optional[int]): Updated annotations required per item.
  • reservation_timeout_minutes (Optional[int]): Updated reservation timeout.
  • requires_review (Optional[bool]): Updated review requirement.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueDetail instance

delete

Deletes (soft-deletes) an annotation queue.

def delete(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

Queue Lifecycle

activate

Activates a queue, transitioning it from draft to active status.

def activate(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueDetail

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueDetail instance

complete_queue

Marks a queue as completed.

def complete_queue(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueDetail

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueDetail instance

Queue Items

add_items

Adds items to the queue for annotation.

def add_items(
    self,
    queue_id: Optional[str] = None,
    items: Optional[List[Dict[str, str]]] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> AddItemsResponse

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • items (List[Dict[str, str]]): List of dicts, each with source_type and source_id.
    • Valid source_type values: "trace", "observation_span", "trace_session", "call_execution", "prototype_run", "dataset_row".
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • AddItemsResponse with added and duplicates counts.

list_items

Lists items in a queue with optional filters.

def list_items(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    status: Optional[str] = None,
    assigned_to: Optional[str] = None,
    page: int = 1,
    page_size: int = 50,
    timeout: Optional[int] = None,
) -> List[QueueItem]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • status (Optional[str]): Filter by item status — "pending", "in_progress", or "completed".
  • assigned_to (Optional[str]): Filter by assigned user ID.
  • page (int): Page number. Defaults to 1.
  • page_size (int): Results per page. Defaults to 50.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • List[QueueItem]

remove_items

Bulk-removes items from the queue.

def remove_items(
    self,
    queue_id: Optional[str] = None,
    item_ids: Optional[List[str]] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_ids (List[str]): List of item UUIDs to remove.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

assign_items

Assigns items to an annotator. Pass user_id=None to unassign.

def assign_items(
    self,
    queue_id: Optional[str] = None,
    item_ids: Optional[List[str]] = None,
    *,
    queue_name: Optional[str] = None,
    user_id: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_ids (List[str]): List of item UUIDs to assign.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • user_id (Optional[str]): User UUID to assign to. Pass None to unassign.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

complete_item

Marks a queue item as completed.

def complete_item(
    self,
    queue_id: Optional[str] = None,
    item_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_id (str): UUID of the queue item.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

skip_item

Skips a queue item.

def skip_item(
    self,
    queue_id: Optional[str] = None,
    item_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_id (str): UUID of the queue item.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

Annotations

submit_annotations

Submits annotations for a queue item as the authenticated user.

def submit_annotations(
    self,
    queue_id: Optional[str] = None,
    item_id: Optional[str] = None,
    annotations: Optional[List[Dict[str, Any]]] = None,
    *,
    queue_name: Optional[str] = None,
    notes: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_id (str): UUID of the queue item.
  • annotations (List[Dict[str, Any]]): List of dicts, each with label_id and value.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • notes (Optional[str]): Free-text notes.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

import_annotations

Imports annotations for a queue item programmatically. Use this when you want to bulk-import annotations from an external source or automated pipeline.

def import_annotations(
    self,
    queue_id: Optional[str] = None,
    item_id: Optional[str] = None,
    annotations: Optional[List[Dict[str, Any]]] = None,
    *,
    queue_name: Optional[str] = None,
    annotator_id: Optional[str] = None,
    timeout: Optional[int] = None,
) -> ImportAnnotationsResponse

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_id (str): UUID of the queue item.
  • annotations (List[Dict[str, Any]]): List of dicts, each with label_id and value. Optionally include score_source (default: "imported").
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • annotator_id (Optional[str]): User ID to attribute the annotations to.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • ImportAnnotationsResponse with imported count.

get_annotations

Gets all annotations for a queue item.

def get_annotations(
    self,
    queue_id: Optional[str] = None,
    item_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> List[Score]

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • item_id (str): UUID of the queue item.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • List[Score]

Scores

Scores provide a unified annotation model that can be used independently of queues to annotate any source entity.

create_score

Creates a single score with upsert semantics.

def create_score(
    self,
    source_type: str,
    source_id: str,
    label_id: Optional[str] = None,
    value: Any = None,
    *,
    label_name: Optional[str] = None,
    score_source: str = "api",
    notes: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Score

• Arguments:
  • source_type (str): Source entity type — "trace", "observation_span", "trace_session", "call_execution", "prototype_run", or "dataset_row".
  • source_id (str): UUID of the source entity.
  • label_id (Optional[str]): UUID of the annotation label.
  • value (Any): Annotation value (str, float, bool, or list depending on label type).
  • label_name (Optional[str]): Name of the annotation label (alternative to label_id).
  • score_source (str): Origin of the score — "human", "api", or "auto". Defaults to "api".
  • notes (Optional[str]): Free-text notes.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Score instance

create_scores

Creates multiple scores on a single source entity in one request.

def create_scores(
    self,
    source_type: str,
    source_id: str,
    scores: List[Dict[str, Any]],
    *,
    notes: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Dict[str, Any]

• Arguments:
  • source_type (str): Source entity type.
  • source_id (str): UUID of the source entity.
  • scores (List[Dict[str, Any]]): List of dicts, each with label_id, value, and optionally score_source.
  • notes (Optional[str]): Shared free-text notes.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • Dict[str, Any]

get_scores

Gets all scores for a given source entity.

def get_scores(
    self,
    source_type: str,
    source_id: str,
    *,
    timeout: Optional[int] = None,
) -> List[Score]

• Arguments:
  • source_type (str): Source entity type.
  • source_id (str): UUID of the source entity.
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • List[Score]

Progress & Analytics

get_progress

Gets queue progress metrics.

def get_progress(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueProgress

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueProgress instance with total, pending, in_progress, completed, skipped, progress_pct, and annotator_stats.

get_analytics

Gets queue analytics including throughput, annotator performance, and label distribution.

def get_analytics(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueAnalytics

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueAnalytics instance with throughput, annotator_performance, label_distribution, status_breakdown, and total.

get_agreement

Gets inter-annotator agreement metrics for a queue.

def get_agreement(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    timeout: Optional[int] = None,
) -> QueueAgreement

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • QueueAgreement instance with overall_agreement, per_label, and annotator_pairs.

Export

export

Exports queue annotations in JSON or CSV format.

def export(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    export_format: str = "json",
    status: Optional[str] = None,
    timeout: Optional[int] = None,
) -> Any

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • export_format (str): Export format — "json" or "csv". Defaults to "json".
  • status (Optional[str]): Filter by item status (e.g. "completed").
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • For JSON: List[Dict]. For CSV: raw text string.

export_to_dataset

Exports annotated queue items to a Future AGI dataset.

def export_to_dataset(
    self,
    queue_id: Optional[str] = None,
    *,
    queue_name: Optional[str] = None,
    dataset_name: Optional[str] = None,
    dataset_id: Optional[str] = None,
    status_filter: Optional[str] = None,
    timeout: Optional[int] = None,
) -> ExportToDatasetResponse

• Arguments:
  • queue_id (Optional[str]): UUID of the annotation queue.
  • queue_name (Optional[str]): Name of the annotation queue (alternative to queue_id).
  • dataset_name (Optional[str]): Name for a new dataset. Mutually exclusive with dataset_id.
  • dataset_id (Optional[str]): UUID of an existing dataset to append to. Mutually exclusive with dataset_name.
  • status_filter (Optional[str]): Item status to export. Defaults to "completed".
  • timeout (Optional[int]): Request timeout in seconds.
• Returns:
  • ExportToDatasetResponse with dataset_id, dataset_name, and rows_created.

Data Models

All data models are importable from the SDK and work with IDE autocomplete:

from fi.queues import (
    AnnotationLabel, QueueDetail, QueueItem, Score,
    QueueProgress, QueueAnalytics, QueueAgreement,
    AddItemsResponse, ExportToDatasetResponse, ImportAnnotationsResponse,
)

AnnotationLabel

QueueDetail

QueueItem

Score

QueueProgress

QueueAnalytics

QueueAgreement

AddItemsResponse

ExportToDatasetResponse

ImportAnnotationsResponse