How to use AssistantEventHandler in Python
Direct answer
Use
AssistantEventHandler in Python by subclassing it and overriding event methods to handle assistant events, then register your handler with the OpenAI assistant client.Setup
Install
pip install openai Env vars
OPENAI_API_KEY Imports
from openai import OpenAI
from openai.assistants import AssistantEventHandler
import os Examples
inInitialize AssistantEventHandler to log messages on assistant response.
outPrinted log of assistant response content.
inCreate a custom event handler to modify messages before sending.
outModified message content sent to the assistant and printed response.
inUse AssistantEventHandler to handle errors during assistant calls.
outError details printed when an exception occurs.
Integration steps
- Import
AssistantEventHandlerand subclass it to define event methods. - Override methods like
on_messageoron_errorto handle events. - Initialize the OpenAI client with your API key from
os.environ. - Register your custom event handler with the assistant client or event system.
- Send messages to the assistant and let the event handler process events.
- Extract and use the assistant's response as needed.
Full code
import os
from openai import OpenAI
from openai.assistants import AssistantEventHandler
class MyAssistantEventHandler(AssistantEventHandler):
def on_message(self, message):
print(f"Assistant sent a message: {message}")
def on_error(self, error):
print(f"Error occurred: {error}")
def main():
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
handler = MyAssistantEventHandler()
# Example usage: simulate sending a message and triggering event
try:
# Normally you would register handler with assistant client or event system
# Here we simulate calling the handler manually for demonstration
user_message = {"role": "user", "content": "Hello, assistant!"}
# Simulate assistant response
assistant_response = {"role": "assistant", "content": "Hello! How can I help you today?"}
# Trigger event handler method
handler.on_message(assistant_response)
# Call the chat completion endpoint
response = client.chat.completions.create(
model="gpt-4o",
messages=[user_message]
)
print("Assistant reply:", response.choices[0].message.content)
except Exception as e:
handler.on_error(e)
if __name__ == "__main__":
main() output
Assistant sent a message: {'role': 'assistant', 'content': 'Hello! How can I help you today?'}
Assistant reply: Hello! How can I help you today? API trace
Request
{"model": "gpt-4o", "messages": [{"role": "user", "content": "Hello, assistant!"}]} Response
{"choices": [{"message": {"role": "assistant", "content": "Hello! How can I help you today?"}}], "usage": {"total_tokens": 20}} Extract
response.choices[0].message.contentVariants
Streaming event handler ›
Use streaming event handler to process partial assistant responses in real time for better UX.
import os
from openai import OpenAI
from openai.assistants import AssistantEventHandler
class StreamingEventHandler(AssistantEventHandler):
def on_message(self, message):
print(f"Streamed message chunk: {message}")
def main():
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
handler = StreamingEventHandler()
# Example streaming call with event handler
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Stream this response."}],
stream=True
)
for chunk in response:
handler.on_message(chunk.choices[0].delta)
if __name__ == "__main__":
main() Async event handler ›
Use async event handler when integrating with async frameworks or concurrent workflows.
import os
import asyncio
from openai import OpenAI
from openai.assistants import AssistantEventHandler
class AsyncEventHandler(AssistantEventHandler):
async def on_message(self, message):
print(f"Async received message: {message}")
async def main():
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
handler = AsyncEventHandler()
user_message = {"role": "user", "content": "Hello async!"}
response = await client.chat.completions.acreate(
model="gpt-4o",
messages=[user_message]
)
await handler.on_message(response.choices[0].message)
print("Assistant reply:", response.choices[0].message.content)
if __name__ == "__main__":
asyncio.run(main()) Alternative model usage ›
Use a smaller model like gpt-4o-mini for faster, lower-cost responses with slightly reduced capability.
import os
from openai import OpenAI
from openai.assistants import AssistantEventHandler
class SimpleHandler(AssistantEventHandler):
def on_message(self, message):
print(f"Message from assistant: {message}")
def main():
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
handler = SimpleHandler()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Use a smaller model."}]
)
handler.on_message(response.choices[0].message)
print("Assistant reply:", response.choices[0].message.content)
if __name__ == "__main__":
main() Performance
Latency~800ms for <code>gpt-4o</code> non-streaming calls
Cost~$0.002 per 500 tokens for <code>gpt-4o</code>
Rate limitsTier 1: 500 requests per minute / 30,000 tokens per minute
- Keep messages concise to reduce token usage.
- Use smaller models like <code>gpt-4o-mini</code> for cheaper calls.
- Cache frequent prompts and responses to avoid repeated calls.
| Approach | Latency | Cost/call | Best for |
|---|---|---|---|
| Standard event handler | ~800ms | ~$0.002 | Simple event handling in synchronous code |
| Streaming event handler | Varies, near real-time | Similar to standard | Real-time partial response processing |
| Async event handler | ~800ms | ~$0.002 | Integration with async applications |
Quick tip
Subclass <code>AssistantEventHandler</code> and override only the events you need to keep your code clean and maintainable.
Common mistake
Beginners often forget to register or invoke their <code>AssistantEventHandler</code>, resulting in no event callbacks firing.