Clarifai Protocol

This is a proprietary protocol used by our runners to communicate with our API. This should be installed as part of our python SDK.

Installation

Install the base package:

pip install clarifai-protocol

To use the auto-annotation feature (requires av for video decoding and shapely for polygon geometry), install the optional extra:

pip install clarifai-protocol[auto-annotation]

The dependencies for this extra are listed in requirements-auto-annotation.txt.

Request Cancellation Support

The protocol now supports request cancellation, allowing models to abort in-flight requests to external inference servers when a user cancels their request.

Features

• Request ID Access: Models can access the current request ID via get_request_id()
• Abort Callbacks: Models can register a callback that will be invoked when requests are cancelled or when connections are aborted
• Thread Safety: All operations are thread-safe and work correctly with concurrent requests
• Background Execution: Abort callbacks run in background threads to avoid blocking the protocol

Usage

from clarifai_protocol import get_request_id, register_abort_callback
import requests

class MyModel:
    def __init__(self):
        super().__init__()
        self.sglang_url = "http://localhost:30000"
        # Register abort callback during initialization
        register_abort_callback(self._abort_sglang)
    
    def _abort_sglang(self, req_id: str) -> None:
        """Abort handler called when requests are cancelled."""
        try:
            requests.post(
                f"{self.sglang_url}/abort_request",
                json={"rid": req_id},
                timeout=2.0
            )
        except Exception:
            pass  # Handle exceptions gracefully
    
    def generate(self, prompt: str):
        """Generate text using external inference server."""
        # Get the request ID to pass to the external server
        req_id = get_request_id()
        
        # Use req_id when calling external services
        for token in self._call_sglang(prompt, req_id):
            yield token

API Reference

get_request_id() -> Optional[str]

Returns the current request ID, or None if called outside of a request context.

Use this as an identifier when calling external inference servers so that cancellation can properly abort the request.

register_abort_callback(callback: Callable[[str], None]) -> None

Register a function to be called when a request is cancelled or the connection is aborted.

• Should be called once during model initialization
• The callback receives the cancelled request's ID as a parameter
• The callback runs in a background thread
• The callback should be idempotent (safe to call multiple times with the same req_id)
• Exceptions in the callback are logged but don't crash the protocol
• Triggered when: explicit cancellation (RUNNER_ITEM_CANCELLED) or connection abort (stream done/cancelled)

Release Instructions

Standard Release

1. Bump the version in the VERSION file
2. Merge the PR to main
3. Pull the latest main: git pull origin main
4. Create and push a tag matching the version: git tag <version> && git push origin <version>

Pre-release / RC Versions

To publish a release candidate (RC) or pre-release version (e.g., for testing):

1. Update the VERSION file to the RC version (e.g., 0.0.60rc1, 0.0.60rc2, 1.2.3alpha1)
2. Commit and push to your branch
3. Create and push a tag matching the version:

   git tag 0.0.60rc1
   git push origin 0.0.60rc1
   

The GitHub Actions workflow will automatically build and publish the package to PyPI when a tag matching the pattern [0-9]+.[0-9]+.[0-9a-zA-Z]+ is pushed. This includes standard versions (e.g., 1.2.3) and pre-release versions (e.g., 0.0.60rc1, 1.2.3alpha1).