Stardog Cloud

The stardog.cloud subpackage provides clients for accessing Stardog Cloud services, such as Stardog Voicebox.

Installation

To use Stardog Cloud functionality, install pystardog with the cloud optional dependencies:

pip install pystardog[cloud]

Clients
Class	Description
`stardog.cloud.client.Client`	Ideal for simple blocking operations in traditional Python applications.
`stardog.cloud.client.AsyncClient`	Suitable for async programming models, such as `asyncio`.

Voicebox Integration

Access Stardog’s Voicebox capabilities through an instance of stardog.cloud.voicebox.VoiceboxApp class.

Note

The stardog.cloud.voicebox.VoiceboxApp provides both synchronous and asynchronous methods, offering the same functionality. Asynchronous methods are prefixed with async_ and return coroutines that can be awaited, while synchronous methods return objects directly. For example, stardog.cloud.voicebox.VoiceboxApp.async_ask returns a coroutine, while stardog.cloud.voicebox.VoiceboxApp.ask returns a stardog.cloud.voicebox.VoiceboxAnswer.

Synchronous usage

from stardog.cloud.client import Client

with Client() as client:
    app = client.voicebox_app(app_api_token="your-token", client_id="your-client-id")
    response = app.ask(question="Your question here")

Asynchronous usage

from stardog.cloud.client import AsyncClient

async with AsyncClient() as client:
    app = client.voicebox_app(app_api_token="your-token", client_id="your-client-id")
    response = await app.async_ask(question="Your question here")

Streaming

For long-running questions, use the streaming API to receive incremental updates. Each streamed answer is a VoiceboxAnswer — the same model returned by ask. The pending field indicates whether more answers are coming (True) or the stream is complete (False).

Synchronous streaming:

from stardog.cloud.client import Client

with Client() as client:
    app = client.voicebox_app(app_api_token="your-token", client_id="your-client-id")

    with app.stream_ask(question="Your question here") as stream:
        for answer in stream:
            if not answer.pending:
                print(answer.content)

Asynchronous streaming:

from stardog.cloud.client import AsyncClient

async with AsyncClient() as client:
    app = client.voicebox_app(app_api_token="your-token", client_id="your-client-id")

    async with app.async_stream_ask(question="Your question here") as stream:
        async for answer in stream:
            if not answer.pending:
                print(answer.content)

Error Handling

The package raises custom exceptions defined in the stardog.cloud.exceptions module. These exceptions can be caught when interacting with the Stardog Cloud clients.

All clients raise exceptions if needed, with stardog.cloud.exceptions.StardogCloudException serving as the base exception for any error raised.

from stardog.cloud.exceptions import (
    StardogCloudException,
    RateLimitExceededException
)

try:
    with Client() as client:
        app = client.voicebox_app(app_api_token="your-token", client_id="your-client-id")
        response = app.ask(question="Your question")
except RateLimitExceededException as e:
    print(f"Rate limit exceeded: {e}")
except StardogCloudException as e:
    print(f"Error occurred: {e}")

API Reference

stardog.cloud.client

class stardog.cloud.client.AsyncClient(base_url='https://cloud.stardog.com/api', timeout=None)[source]

Bases: BaseClient

An asynchronous client for interacting with the Stardog Cloud API.

Use async with stardog.cloud.AsyncClient() if you want a context-managed client

>>> async with stardog.cloud.AsyncClient() as client:
>>>     voicebox = client.voicebox_app(app_api_token="my-secret-token", client_id="some-id")
>>>     answer = await voicebox.async_ask(question="Who produced the most cars in 2023?")

Alternatively, use await client.aclose() if you want to close a client explicitly.
This is needed to avoid resource leaks. There is no need to close aclose() if you are using a context-managed client.

>>> client = stardog.cloud.AsyncClient()
...do stuff
>>> await client.aclose()

__init__(base_url='https://cloud.stardog.com/api', timeout=None)[source]

Parameters:

base_url (str, default: 'https://cloud.stardog.com/api') – The base URL of the Stardog Cloud API.
timeout (Optional[float], default: None) – Request timeout in seconds. Defaults to 30s if not provided.

async aclose()[source]: Close the transport and proxies. Should be called if this client instance is not being used as a context manager.

property base_url: str: The base URL of the Stardog Cloud API.

voicebox_app(app_api_token, client_id=None)[source]

Initialize the Voicebox app.

Return type:: VoiceboxApp

class stardog.cloud.client.BaseClient[source]

Bases: ABC

abstract property base_url: str: The base URL of the Stardog Cloud API.

abstract voicebox_app(app_api_token, client_id=None)[source]: Initialize the Voicebox app.

class stardog.cloud.client.Client(base_url='https://cloud.stardog.com/api', timeout=None)[source]

Bases: BaseClient

A synchronous client for interacting with the Stardog Cloud API.

Use with stardog.cloud.Client() if you want a context-managed client

>>> with stardog.cloud.Client() as client:
>>>     voicebox = client.voicebox_app(app_api_token="my-secret-token", client_id="some-id")
>>>     answer = voicebox.ask(question="Who produced the most cars in 2023?")

Alternatively, use client.close() if you want to close a client explicitly: This is needed to avoid resource leaks. There is no need to close close() if you are using a context-managed client.

>>> client = stardog.cloud.Client()
...do stuff
>>> client.close()

__init__(base_url='https://cloud.stardog.com/api', timeout=None)[source]

Parameters:

base_url (str, default: 'https://cloud.stardog.com/api') – The base URL of the Stardog Cloud API.
timeout (Optional[float], default: None) – Request timeout in seconds.

property base_url: str: The base URL of the Stardog Cloud API.

close()[source]: Close transport and proxies. Should be called if this client instance is not being used as a context manager.

voicebox_app(app_api_token, client_id=None)[source]

Initialize the Voicebox app.

Return type:: VoiceboxApp

class stardog.cloud.client.StardogCloudAPIEndpoints(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Stardog Cloud API endpoints

EU = 'https://eu-cloud.stardog.com/api'

US = 'https://cloud.stardog.com/api'

stardog.cloud.voicebox

pydantic model stardog.cloud.voicebox.VoiceboxAction[source]

Bases: BaseModel

Config:

extra: str = allow

Fields:

label (str | None)
type (str)
value (str)

field label: Optional[str] = None

field type: str [Required]

field value: str [Required]

pydantic model stardog.cloud.voicebox.VoiceboxAnswer[source]

Bases: BaseModel

Config:

extra: str = allow

Fields:

actions (List[stardog.cloud.voicebox.VoiceboxAction])
content (str)
conversation_id (str)
message_id (str)
pending (bool | None)

field actions: List[VoiceboxAction] [Optional]: The raw “actions” returned by Voicebox. Generally, you can just use the properties like stardog.cloud.voicebox.VoiceboxAnswer.interpreted_question or stardog.cloud.voicebox.VoiceboxAnswer.sparql_query instead of filtering the actions for these specific ones.

field content: str [Required]: The main response to your question. This is intended to display to an end user.

field conversation_id: str [Required]: The id (UUID) of the Voicebox conversation. Can be provided to different methods on the stardog.cloud.voicebox.VoiceboxApp class to continue an existing conversation.

field message_id: str [Required]: The id (UUID) of the Voicebox message.

field pending: Optional[bool] = None: Streaming indicator. None for non-streaming responses (from ask), True for intermediate streaming events (more data coming), False for the final streaming event (stream complete).

property interpreted_question: str | None: How Voicebox interpreted your question.

property sparql_query: str | None: The SPARQL query used to generate the response.

class stardog.cloud.voicebox.VoiceboxApp(client, app_api_token, client_id=None)[source]

Bases: object

The Voicebox App created in the Stardog Cloud Portal.

__init__(client, app_api_token, client_id=None)[source]

Parameters:

client (BaseClient) – the Stardog Cloud API Client
app_api_token (str) – the Voicebox app API token from Stardog Cloud
client_id (Optional[str], default: None) – a unique identifier to specify the identity of the Voicebox user

ask(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask a question to Voicebox.

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required only if client_id was not provided when creating a Voicebox instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

Return type:

VoiceboxAnswer

async async_ask(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask Voicebox to generate a SPARQL query based on a natural language question.

Note

Async version of stardog.cloud.voicebox.VoiceboxApp.ask

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required if client_id was not provided when creating a stardog.cloud.voicebox.VoiceboxApp instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

async async_generate_query(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask Voicebox to generate a SPARQL query based on a natural language question.

Note

Async version of stardog.cloud.voicebox.VoiceboxApp.generate_query

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required if client_id was not provided when creating a stardog.cloud.voicebox.VoiceboxApp instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

async async_settings()[source]

Note

Async version of stardog.cloud.voicebox.VoiceboxApp.settings

Returns the Voicebox application information registered with the app_api_token

Return type:: VoiceboxAppSettings

async_stream_ask(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask a question to Voicebox with a streaming response.

Note

Async version of stardog.cloud.voicebox.VoiceboxApp.stream_ask

Returns an async context manager that yields an async iterator of VoiceboxAnswer. Each yielded answer has a pending field: True for intermediate events and False for the final event containing the complete answer.

Must be used with an async with statement to ensure proper resource cleanup.

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required if client_id was not provided when creating a stardog.cloud.voicebox.VoiceboxApp instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

Return type:

AsyncIterator[AsyncIterator[VoiceboxAnswer]]

generate_query(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask Voicebox to generate a SPARQL query based on a natural language question.

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required if client_id was not provided when creating a stardog.cloud.voicebox.VoiceboxApp instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

settings()[source]

Returns the Voicebox application information registered with the app_api_token

Return type:: VoiceboxAppSettings

stream_ask(question, conversation_id=None, client_id=None, stardog_auth_token_override=None)[source]

Ask a question to Voicebox with a streaming response.

Returns a context manager that yields an iterator of VoiceboxAnswer. Each yielded answer has a pending field: True for intermediate events and False for the final event containing the complete answer.

Must be used with a with statement to ensure proper resource cleanup.

Parameters:

question (str) – the question to ask Voicebox, e.g. "How many products were sold in 2024?"
conversation_id (Optional[str], default: None) – the id of the Voicebox conversation on Stardog Cloud. If not provided, a new conversation will be created and the conversation id will be returned in the response.
client_id (Optional[str], default: None) – only required if client_id was not provided when creating a stardog.cloud.voicebox.VoiceboxApp instance
stardog_auth_token_override (Optional[str], default: None) – optional bearer token to override the default Stardog token associated with your Voicebox app token. This is especially useful when your Voicebox App connects to Stardog via an SSO provider (e.g., Microsoft Entra) and you need to supply your own SSO-issued token to authenticate requests to your Stardog server

Return type:

Iterator[Iterator[VoiceboxAnswer]]

pydantic model stardog.cloud.voicebox.VoiceboxAppSettings[source]

Bases: BaseModel

Voicebox App settings. Some settings can be edited in the Stardog Cloud Portal.

Fields:

database (str)
model (str)
name (str)
named_graphs (List[str])
reasoning (bool)

field database: str [Required]: Stardog database the Voicebox App was configured with

field model: str [Required]: The model within the Stardog database

field name: str [Required]: name of the Voicebox App

field named_graphs: List[str] [Required]: The named graphs within the Stardog database Voicebox will execute queries against

field reasoning: bool [Required]: Whether Voicebox should use reasoning or not in its queries to the Stardog database.

stardog.cloud.exceptions

exception stardog.cloud.exceptions.BadRequestException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 400 Bad Request

exception stardog.cloud.exceptions.ForbiddenException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 403 Forbidden

exception stardog.cloud.exceptions.GatewayTimeoutException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 504 Gateway Timeout Error

exception stardog.cloud.exceptions.InternalServerException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 500 Internal Server Error

exception stardog.cloud.exceptions.NotFoundException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 404 Not Found

exception stardog.cloud.exceptions.RateLimitExceededException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 429 Too Many Requests

exception stardog.cloud.exceptions.StardogCloudException(message, status_code=None)[source]

Bases: Exception

Base exception of all exceptions raised by the stardog.cloud subpackage.

__init__(message, status_code=None)[source]

exception stardog.cloud.exceptions.UnauthorizedException(message, status_code=None)[source]

Bases: StardogCloudException

Exception when Stardog Cloud API replies with a 401 Unauthorized