Python SDK

import asyncio
import os

from interhumanai import InterhumanClient, UploadJobIncludeFlag, UploadJobStatus, UploadModel

async def main() -> None:
    client = InterhumanClient(access_token=os.environ["INTERHUMAN_API_KEY"])
    job = await client.upload.submit(
        "meeting.mp4", model=UploadModel.INTER_2, include=[UploadJobIncludeFlag.SIGNALS]
    )
    job = await client.upload.wait_for_job(job)
    if job.status is UploadJobStatus.COMPLETED:
        for signal in job.result.signals:
            print(signal.type.value, signal.start, signal.end)
    else:
        print(job.error.error_id, job.error.message)

asyncio.run(main())

AuthClient(*, environment: Optional[Literal['production', 'staging']] = None, base_url: str | None = None, http_client: httpx.AsyncClient | None = None) -> None

@property
def base_url(self) -> str

async def create_client_token(self, *, api_key: str, scopes: collections.abc.Sequence[Scope | str] | None = None, expires_in: int | None = None, max_duration_seconds: int | None = None, max_bytes: int | None = None, max_concurrent: int | None = None, max_video_seconds: int | None = None, allowed_origins: collections.abc.Sequence[str] | None = None) -> ClientTokenResponse

async def revoke_client_token(self, *, api_key: str, token: str) -> None

InterhumanClient(*, key_id: str | None = None, key_secret: str | None = None, access_token: str | None = None, scopes: collections.abc.Sequence[Scope | str] | None = None, environment: Optional[Literal['production', 'staging']] = None, base_url: str | None = None, http_client: httpx.AsyncClient | None = None, refresh_skew_seconds: float = 30.0) -> None

async def get_token(self) -> str

def stream(self, *, api_version: Literal['v1', 'v2'] = 'v1', model: StreamModel | str | None = None) -> StreamClient

SessionSocketClient(*, token_provider: TokenProvider, environment: Optional[Literal['production', 'staging']] = None, base_url: str | None = None) -> None

async def close(self, code: int = 1000, reason: str = '') -> None

@property
def close_info(self) -> CloseInfo | None

async def connect(self) -> 'SessionSocketClient[EventT]'

@property
def is_open(self) -> bool

async def request_close(self) -> None

async def send_video(self, chunk: bytes | bytearray | memoryview) -> None

@property
def url(self) -> str

async def wait_closed(self) -> CloseInfo

async def wait_for_session_ready(self) -> EventT

StaticTokenProvider(token: str) -> None

async def get_token(self) -> str

StreamClient(*, token_provider: TokenProvider, environment: Optional[Literal['production', 'staging']] = None, base_url: str | None = None, api_version: Literal['v1', 'v2'] = 'v1', model: StreamModel | str | None = None) -> None

@property
def api_version(self) -> Literal['v1', 'v2']

@property
def model(self) -> str | None

async def update_config(self, *, include: collections.abc.Sequence[IncludeFlag | str] | None = None, goal_dimensions: collections.abc.Sequence[GoalDimension | str] | None = None, model: StreamModel | str | None = None) -> None

TokenManager(*, auth_client: AuthClient, key_id: str, key_secret: str, scopes: collections.abc.Sequence[Scope | str] = (<Scope.UPLOAD: 'interhumanai.upload'>, <Scope.STREAM: 'interhumanai.stream'>), refresh_skew_seconds: float = 30.0, clock: Callable[[], float] = <built-in function monotonic>) -> None

async def get_token(self) -> str

def invalidate(self) -> None

TokenProvider(*args, **kwargs)

async def get_token(self) -> str

UploadClient(*, token_provider: TokenProvider, environment: Optional[Literal['production', 'staging']] = None, base_url: str | None = None, http_client: httpx.AsyncClient | None = None) -> None

async def analyze(self, file: Union[bytes, bytearray, memoryview, IO[bytes], str, os.PathLike[str]], *, filename: str | None = None, content_type: str | None = None, include: collections.abc.Sequence[IncludeFlag | str] | None = None, goal_dimensions: collections.abc.Sequence[GoalDimension | str] | None = None, conversation_context: str | None = None) -> AnalysisResult

async def get_job(self, job_id: str | UploadJob) -> UploadJob

async def submit(self, file: Union[bytes, bytearray, memoryview, IO[bytes], str, os.PathLike[str]], *, model: UploadModel | str, wait_seconds: int = 0, include: collections.abc.Sequence[UploadJobIncludeFlag | IncludeFlag | str] | None = None, filename: str | None = None, content_type: str | None = None) -> UploadJob

async def wait_for_job(self, job_id: str | UploadJob, *, timeout: float | None = None, poll_interval: float = 2.0) -> UploadJob

def http_to_ws_base_url(http_base_url: str) -> str

def parse_stream_event(payload: dict[str, Any]) -> Union[SessionReadyEvent, SessionUpdatedEvent, SessionClosingEvent, SessionEndedEvent, SignalDetectedEvent, SignalUpdatedEvent, SignalEndedEvent, EngagementUpdatedEvent, ConversationQualityUpdatedEvent, FeedbackGeneratedEvent, CoverageDroppedEvent, CoverageDegradedEvent, ErrorEvent, UnknownEvent]

def resolve_http_base_url(*, base_url: str | None = None, environment: Optional[Literal['production', 'staging']] = None) -> str

def sdk_header_value() -> str

DEFAULT_JOB_POLL_INTERVAL_SECONDS = 2.0

DEFAULT_REFRESH_SKEW_SECONDS = 30.0

DEFAULT_SCOPES = (<Scope.UPLOAD: 'interhumanai.upload'>, <Scope.STREAM: 'interhumanai.stream'>)

Environment = Literal['production', 'staging']

HTTP_BASE_URLS = {'production': 'https://api.interhuman.ai', 'staging': 'https://staging-api.interhuman.ai'}

InteractionFeedback = Feedback | NoFeedback

SDK_HEADER_NAME = 'X-Interhuman-SDK'

SDK_NAME = 'python'

STREAM_ENDPOINT_PATHS = {'v1': '/v1/stream/analyze', 'v2': '/v2/stream/analyze'}

StreamApiVersion = Literal['v1', 'v2']

StreamEvent = Union[SessionReadyEvent, SessionUpdatedEvent, SessionClosingEvent, SessionEndedEvent, SignalDetectedEvent, SignalUpdatedEvent, SignalEndedEvent, EngagementUpdatedEvent, ConversationQualityUpdatedEvent, FeedbackGeneratedEvent, CoverageDroppedEvent, CoverageDegradedEvent, ErrorEvent, UnknownEvent]

VideoInput = Union[bytes, bytearray, memoryview, IO[bytes], str, os.PathLike[str]]