Astra Assistants

2.5.5 · active · verified Wed Apr 15

The `astra-assistants` library provides a Pythonic, drop-in replacement for OpenAI's Assistants API, powered by AstraDB. It allows developers to build AI assistants with capabilities like persistent memory, tool use (e.g., web search, code interpreter), and RAG, leveraging AstraDB as the vector database backend. The current version is 2.5.5, with regular releases aligning with OpenAI's API updates.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up an `AstraAssistants` client, create an assistant, manage a conversation thread by adding messages, running the assistant, and retrieving its responses. Ensure your AstraDB authentication environment variables are correctly configured before running.

import os
from astra_assistants import AstraAssistants
from astra_assistants.models import ThreadMessage

# --- Authentication Configuration ---
# Ensure these environment variables are set:
# os.environ["ASTRA_DB_APPLICATION_TOKEN"] = "AstraCS:your-token"
# os.environ["ASTRA_DB_API_ENDPOINT"] = "https://<REGION>.aws.a.astra.datastax.com/api/rest"
# Optional: os.environ["OPENAI_API_KEY"] = "sk-..." # Only if using OpenAI models via AstraDB

# Initialize the client. It automatically picks up credentials from environment variables.
client = AstraAssistants()

# 1. Create an assistant
assistant = client.beta.assistants.create(
    name="Math Tutor",
    instructions="You are a personal math tutor. Answer questions briefly.",
    tools=[{"type": "code_interpreter"}], # Example tool. For web_search, use {"type": "web_search"}
    model="gpt-4o", # Or a model accessible via your AstraDB setup, e.g., 'gpt-3.5-turbo'
)
print(f"Created Assistant: {assistant.id}")

# 2. Create a thread
thread = client.beta.threads.create()
print(f"Created Thread: {thread.id}")

# 3. Add a message to the thread
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="What is the result of 15 * 3 + 2?",
)
print("Added message to thread.")

# 4. Run the assistant on the thread and poll for completion
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id,
    instructions="Please address the user as 'Student'.",
)
print(f"Run completed with status: {run.status}")

# 5. Retrieve and print the messages
if run.status == "completed":
    messages_page = client.beta.threads.messages.list(thread_id=thread.id, order="asc")
    for msg in messages_page.data:
        # The content can be a list of different block types. For text, access .text.value
        if msg.content and hasattr(msg.content[0], 'text'):
             print(f"{msg.role}: {msg.content[0].text.value}")
        else:
             print(f"{msg.role}: {msg.content}") # Fallback for other content types
else:
    print(f"Run finished with status {run.status}. Could not retrieve messages.")

# In a real application, you might delete the assistant and thread here.
# client.beta.assistants.delete(assistant.id)
# client.beta.threads.delete(thread.id)

view raw JSON →