astra-assistants

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)