Skip to content

Asst

AssistantAPI

Bases: OpenAIConfig

Methods:

Name Description
create_or_retrieve_thread
retrieve_assistant_by_settings
delete_thread
create_or_retrieve_assistant
delete_assistant
create_message
list_messages
retrieve_message
delete_message
retrieve_run
delete_run
create_run

thread_id 

thread_id: Optional[str] = Field(default=None)

assistant_id 

assistant_id: Optional[str] = Field(default=None)

message_id 

message_id: Optional[str] = Field(default=None)

metadata 

metadata: dict[str, str] = Field(default={'backend_id': 'default'})

client 

client: OpenAI

last_message 

last_message: SyncCursorPage[Message]

api_type 

api_type: str = Field(
    default="openai",
    description="The api type from openai for calling models.",
    examples=["openai", "azure"],
    validation_alias=AliasChoices("OPENAI_API_TYPE"),
    frozen=False,
    deprecated=False,
)

base_url 

base_url: str = Field(
    ...,
    description="The base url from openai for calling models.",
    examples=["https://api.openai.com/v1", "https://xxxx.openai.azure.com"],
    validation_alias=AliasChoices("OPENAI_BASE_URL", "AZURE_OPENAI_ENDPOINT"),
    frozen=False,
    deprecated=False,
)

api_key 

api_key: str = Field(
    ...,
    description="The api key from openai for calling models.",
    examples=["sk-proj-...", "141698ac..."],
    validation_alias=AliasChoices("OPENAI_API_KEY", "AZURE_OPENAI_API_KEY"),
    frozen=False,
    deprecated=False,
)

api_version 

api_version: str = Field(
    default="2025-04-01-preview",
    description="The api version from openai for calling models.",
    examples=["2025-04-01-preview"],
    validation_alias=AliasChoices("OPENAI_API_VERSION"),
    frozen=False,
    deprecated=False,
)

create_or_retrieve_thread 

create_or_retrieve_thread() -> Thread
Source code in src/sdk/asst.py 
27
28
29
30
31
32
33
def create_or_retrieve_thread(self) -> Thread:
    try:
        thread = self.client.beta.threads.retrieve(self.thread_id)
    except Exception:
        thread = self.client.beta.threads.create(metadata=self.metadata)
    self.thread_id = thread.id
    return thread

retrieve_assistant_by_settings 

retrieve_assistant_by_settings(
    name: str, model: str, description: str, instructions: str
) -> Assistant
Source code in src/sdk/asst.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def retrieve_assistant_by_settings(
    self, name: str, model: str, description: str, instructions: str
) -> Assistant:
    existing_assistants = self.client.beta.assistants.list()
    current_asst_config = {
        "name": name,
        "model": model,
        "description": description,
        "instructions": instructions,
    }
    matched_assistants: list[Assistant] = []
    for _existing_assistant in existing_assistants:
        existing_asst_config = _existing_assistant.model_dump(
            include={"name", "model", "description", "instructions"}
        )
        if existing_asst_config == current_asst_config:
            matched_assistants.append(_existing_assistant)

    if not matched_assistants:
        assistant = self.client.beta.assistants.create(
            name=name,
            model=model,
            description=description,
            instructions=instructions,
            metadata=self.metadata,
        )
        matched_assistants.append(assistant)

    self.assistant_id = matched_assistants[0].id
    return matched_assistants[0]

delete_thread 

delete_thread(thread_id: str) -> ThreadDeleted
Source code in src/sdk/asst.py 
66
67
68
def delete_thread(self, thread_id: str) -> ThreadDeleted:
    deleted_thread = self.client.beta.threads.delete(thread_id=thread_id)
    return deleted_thread

create_or_retrieve_assistant 

create_or_retrieve_assistant(
    name: str, model: str, description: str, instructions: str
) -> Assistant
Source code in src/sdk/asst.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def create_or_retrieve_assistant(
    self, name: str, model: str, description: str, instructions: str
) -> Assistant:
    try:
        assistant = self.client.beta.assistants.retrieve(self.assistant_id)
    except Exception:
        assistant = self.client.beta.assistants.create(
            name=name,
            model=model,
            description=description,
            instructions=instructions,
            metadata=self.metadata,
        )
    self.assistant_id = assistant.id
    return assistant

delete_assistant 

delete_assistant(asst_id: str) -> AssistantDeleted
Source code in src/sdk/asst.py 
86
87
88
def delete_assistant(self, asst_id: str) -> AssistantDeleted:
    deleted_assistant = self.client.beta.assistants.delete(assistant_id=asst_id)
    return deleted_assistant

create_message 

create_message(content: str, image_urls: list[str]) -> Message
Source code in src/sdk/asst.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def create_message(self, content: str, image_urls: list[str]) -> Message:
    base_content: list[dict[str, Any]] = [{"type": "text", "text": content}]
    for image_url in image_urls:
        image = get_pil_image(image_file=image_url)
        image_base64 = pil_to_data_uri(image=image)
        base_content.append({"type": "image_url", "image_url": {"url": image_base64}})

    message = self.client.beta.threads.messages.create(
        thread_id=self.thread_id, content=base_content, role="user"
    )
    self.message_id = message.id
    return message

list_messages 

list_messages(thread_id: str) -> list[Message]
Source code in src/sdk/asst.py 
103
104
105
def list_messages(self, thread_id: str) -> list[Message]:
    messages = self.client.beta.threads.messages.list(thread_id=thread_id)
    return messages

retrieve_message 

retrieve_message(thread_id: str, message_id: str) -> Message
Source code in src/sdk/asst.py 
107
108
109
110
111
def retrieve_message(self, thread_id: str, message_id: str) -> Message:
    message = self.client.beta.threads.messages.retrieve(
        thread_id=thread_id, message_id=message_id
    )
    return message

delete_message 

delete_message(thread_id: str, message_id: str) -> MessageDeleted
Source code in src/sdk/asst.py 
113
114
115
116
117
def delete_message(self, thread_id: str, message_id: str) -> MessageDeleted:
    deleted_message = self.client.beta.threads.messages.delete(
        thread_id=thread_id, message_id=message_id
    )
    return deleted_message

retrieve_run 

retrieve_run(thread_id: str, run_id: str) -> Run
Source code in src/sdk/asst.py 
119
120
121
def retrieve_run(self, thread_id: str, run_id: str) -> Run:
    message = self.client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
    return message

delete_run 

delete_run(thread_id: str, run_id: str) -> Run
Source code in src/sdk/asst.py 
123
124
125
def delete_run(self, thread_id: str, run_id: str) -> Run:
    deleted_message = self.client.beta.threads.runs.cancel(thread_id=thread_id, run_id=run_id)
    return deleted_message

create_run 

create_run() -> Run
Source code in src/sdk/asst.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create_run(self) -> Run:
    run = self.client.beta.threads.runs.create(
        thread_id=self.thread_id, assistant_id=self.assistant_id
    )

    def wait_on_run(run: Run) -> Run:
        while run.status == "queued" or run.status == "in_progress":
            run = self.client.beta.threads.runs.retrieve(
                thread_id=self.thread_id, run_id=run.id
            )
            time.sleep(0.5)
        return run

    wait_on_run(run=run)
    return run