​

literalai.client

​

BaseLiteralClient Objects

class BaseLiteralClient()

Attributes:

• api Union[LiteralAPI, AsyncLiteralAPI] - Instance of the Literal API.
• disabled bool - Flag indicating if the client is disabled.
• event_processor EventProcessor - Processor for handling events.

Arguments:

• batch_size int, optional - Batch size for event processing. Defaults to 5.
• is_async bool, optional - Flag indicating if the API should be asynchronous. Defaults to False.
• api_key Optional[str], optional - API key for authentication. If not provided, it will be fetched from the environment variable ‘LITERAL_API_KEY’.
• url Optional[str], optional - URL of the Literal API. Defaults to ’https://cloud.getliteral.ai‘.
• disabled bool, optional - Flag to disable the client. Defaults to False.

Raises:

• Exception - If ‘LITERAL_API_KEY’ is not provided and not found in environment variables.

​

to_sync

def to_sync() -> "LiteralClient"

Convert the client to synchronous mode.

Returns:

• LiteralClient - Synchronous LiteralClient instance.

​

instrument_openai

def instrument_openai()

Instruments the OpenAI API calls with the current client.

​

langchain_callback

def langchain_callback(to_ignore: Optional[List[str]] = None,
                       to_keep: Optional[List[str]] = None,
                       **kwargs: Any)

Get a Langchain callback tracer configured with the client.

Arguments:

• to_ignore Optional[List[str]], optional - List of event types to ignore. Defaults to None.
• to_keep Optional[List[str]], optional - List of event types to keep. Defaults to None.
• **kwargs - Additional arguments for the tracer.

Returns:

• LangchainTracer - Configured Langchain tracer.

​

llama_index_callback

def llama_index_callback(**kwargs: Any)

Get a Llama Index callback tracer configured with the client.

Arguments:

• **kwargs - Additional arguments for the tracer.

Returns:

• LlamaIndexTracer - Configured Llama Index tracer.

​

thread

def thread(original_function=None,
           *,
           thread_id: Optional[str] = None,
           name: Optional[str] = None,
           **kwargs)

Decorator or context manager for thread-level event grouping.

Arguments:

• original_function callable, optional - Function to decorate. Defaults to None.
• thread_id Optional[str], optional - ID of the thread. Defaults to None.
• name Optional[str], optional - Name of the thread. Defaults to None.
• **kwargs - Additional arguments for the thread context manager or decorator.

Returns:

Union[ThreadContextManager, callable]: Thread context manager or decorated function.

​

step

def step(original_function=None,
         *,
         name: str = "",
         type: TrueStepType = "undefined",
         id: Optional[str] = None,
         parent_id: Optional[str] = None,
         thread_id: Optional[str] = None,
         **kwargs)

Decorator or context manager for step-level event grouping.

Arguments:

• original_function callable, optional - Function to decorate. Defaults to None.
• name str, optional - Name of the step. Defaults to "".
• type TrueStepType, optional - Type of the step. Defaults to “undefined”.
• id Optional[str], optional - ID of the step. Defaults to None.
• parent_id Optional[str], optional - Parent ID of the step. Defaults to None.
• thread_id Optional[str], optional - Thread ID of the step. Defaults to None.
• **kwargs - Additional arguments for the step context manager or decorator.

Returns:

Union[StepContextManager, callable]: Step context manager or decorated function.

​

run

def run(original_function=None,
        *,
        name: str = "",
        id: Optional[str] = None,
        parent_id: Optional[str] = None,
        thread_id: Optional[str] = None)

Decorator or context manager for run-level event grouping.

Arguments:

• original_function callable, optional - Function to decorate. Defaults to None.
• name str, optional - Name of the run. Defaults to "".
• id Optional[str], optional - ID of the run. Defaults to None.
• parent_id Optional[str], optional - Parent ID of the run. Defaults to None.
• thread_id Optional[str], optional - Thread ID of the run. Defaults to None.

Returns:

Union[StepContextManager, callable]: Run context manager or decorated function.

​

message

def message(content: str = "",
            id: Optional[str] = None,
            parent_id: Optional[str] = None,
            type: Optional[MessageStepType] = None,
            name: Optional[str] = None,
            thread_id: Optional[str] = None,
            attachments: List[Attachment] = [],
            tags: Optional[List[str]] = None,
            metadata: Dict = {})

Creates and ends a message step.

Arguments:

• content str, optional - Content of the message. Defaults to "".
• id Optional[str], optional - ID of the message. Defaults to None.
• parent_id Optional[str], optional - Parent ID of the message. Defaults to None.
• type Optional[MessageStepType], optional - Type of the message. Defaults to None.
• name Optional[str], optional - Name of the message. Defaults to None.
• thread_id Optional[str], optional - Thread ID of the message. Defaults to None.
• attachments List[Attachment], optional - List of attachments. Defaults to [].
• tags Optional[List[str]], optional - List of tags. Defaults to None.
• metadata Dict, optional - Metadata for the message. Defaults to .

Returns:

• Message - Created message step.

​

start_step

def start_step(name: str = "",
               type: Optional[TrueStepType] = None,
               id: Optional[str] = None,
               parent_id: Optional[str] = None,
               thread_id: Optional[str] = None,
               **kwargs)

Starts a new step.

Arguments:

• name str, optional - Name of the step. Defaults to "".
• type Optional[TrueStepType], optional - Type of the step. Defaults to None.
• id Optional[str], optional - ID of the step. Defaults to None.
• parent_id Optional[str], optional - Parent ID of the step. Defaults to None.
• thread_id Optional[str], optional - Thread ID of the step. Defaults to None.
• **kwargs - Additional arguments for the step.

Returns:

• Step - Started step.

​

get_current_step

def get_current_step()

Retrieves the current active step.

Returns:

• Optional[Step] - The current active step or None if no steps are active.

​

get_current_thread

def get_current_thread()

Retrieves the current active thread.

Returns:

• Optional[ThreadContextManager] - The current active thread or None if no threads are active.

​

reset_context

def reset_context()

Resets the active steps and thread context.

​

flush_and_stop

def flush_and_stop()

Flushes the event processor and stops it.

​

LiteralClient Objects

class LiteralClient(BaseLiteralClient)

Synchronous LiteralClient for interacting with the Literal API.

Arguments:

• batch_size int, optional - Batch size for event processing. Defaults to 5.
• api_key Optional[str], optional - API key for authentication. If not provided, it will be fetched from the environment variable ‘LITERAL_API_KEY’.
• url Optional[str], optional - URL of the Literal API. Defaults to ’https://cloud.getliteral.ai‘.
• disabled bool, optional - Flag to disable the client. Defaults to False.

​

flush

def flush()

Flushes the event processor.

​

AsyncLiteralClient Objects

class AsyncLiteralClient(BaseLiteralClient)

Asynchronous LiteralClient for interacting with the Literal API.

Attributes:

• api AsyncLiteralAPI - Instance of the asynchronous Literal API.

Arguments:

• batch_size int, optional - Batch size for event processing. Defaults to 5.
• api_key Optional[str], optional - API key for authentication. If not provided, it will be fetched from the environment variable ‘LITERAL_API_KEY’.
• url Optional[str], optional - URL of the Literal API. Defaults to ’https://cloud.getliteral.ai‘.
• disabled bool, optional - Flag to disable the client. Defaults to False.

​

flush

async def flush()

Asynchronously flushes the event processor.

Returns:

None