fromagentsimportAgent,Runner,SQLiteSession# Create agentagent=Agent(name="Assistant",instructions="Reply very concisely.",)# Create a session instance with a session IDsession=SQLiteSession("conversation_123")# First turnresult=awaitRunner.run(agent,"What city is the Golden Gate Bridge in?",session=session)print(result.final_output)# "San Francisco"# Second turn - agent automatically remembers previous contextresult=awaitRunner.run(agent,"What state is it in?",session=session)print(result.final_output)# "California"# Also works with synchronous runnerresult=Runner.run_sync(agent,"What's the population?",session=session)print(result.final_output)# "Approximately 39 million"

fromagentsimportSQLiteSessionsession=SQLiteSession("user_123","conversations.db")# Get all items in a sessionitems=awaitsession.get_items()# Add new items to a sessionnew_items=[{"role":"user","content":"Hello"},{"role":"assistant","content":"Hi there!"}]awaitsession.add_items(new_items)# Remove and return the most recent itemlast_item=awaitsession.pop_item()print(last_item)# {"role": "assistant", "content": "Hi there!"}# Clear all items from a sessionawaitsession.clear_session()

fromagentsimportAgent,Runner,SQLiteSessionagent=Agent(name="Assistant")session=SQLiteSession("correction_example")# Initial conversationresult=awaitRunner.run(agent,"What's 2 + 2?",session=session)print(f"Agent:{result.final_output}")# User wants to correct their questionassistant_item=awaitsession.pop_item()# Remove agent's responseuser_item=awaitsession.pop_item()# Remove user's question# Ask a corrected questionresult=awaitRunner.run(agent,"What's 2 + 3?",session=session)print(f"Agent:{result.final_output}")

fromagentsimportAgent,Runner,OpenAIConversationsSession# Create agentagent=Agent(name="Assistant",instructions="Reply very concisely.",)# Create a new conversationsession=OpenAIConversationsSession()# Optionally resume a previous conversation by passing a conversation ID# session = OpenAIConversationsSession(conversation_id="conv_123")# Start conversationresult=awaitRunner.run(agent,"What city is the Golden Gate Bridge in?",session=session)print(result.final_output)# "San Francisco"# Continue the conversationresult=awaitRunner.run(agent,"What state is it in?",session=session)print(result.final_output)# "California"

fromagentsimportAgent,Runner,SQLiteSessionfromagents.memoryimportOpenAIResponsesCompactionSessionunderlying=SQLiteSession("conversation_123")session=OpenAIResponsesCompactionSession(session_id="conversation_123",underlying_session=underlying,)agent=Agent(name="Assistant")result=awaitRunner.run(agent,"Hello",session=session)print(result.final_output)

fromagentsimportAgent,Runner,SQLiteSessionfromagents.memoryimportOpenAIResponsesCompactionSessionunderlying=SQLiteSession("conversation_123")session=OpenAIResponsesCompactionSession(session_id="conversation_123",underlying_session=underlying,# Disable triggering the auto compactionshould_trigger_compaction=lambda_:False,)agent=Agent(name="Assistant")result=awaitRunner.run(agent,"Hello",session=session)# Decide when to compact (e.g., on idle, every N turns, or size thresholds).awaitsession.run_compaction({"force":True})

fromagentsimportSQLiteSession# In-memory database (lost when process ends)session=SQLiteSession("user_123")# Persistent file-based databasesession=SQLiteSession("user_123","conversations.db")# Use the sessionresult=awaitRunner.run(agent,"Hello",session=session)

fromagents.extensions.memoryimportSQLAlchemySession# Using database URLsession=SQLAlchemySession.from_url("user_123",url="postgresql+asyncpg://user:pass@localhost/db",create_tables=True)# Using existing enginefromsqlalchemy.ext.asyncioimportcreate_async_engineengine=create_async_engine("postgresql+asyncpg://user:pass@localhost/db")session=SQLAlchemySession("user_123",engine=engine,create_tables=True)

fromagents.extensions.memoryimportAdvancedSQLiteSession# Create with advanced featuressession=AdvancedSQLiteSession(session_id="user_123",db_path="conversations.db",create_tables=True)# Automatic usage trackingresult=awaitRunner.run(agent,"Hello",session=session)awaitsession.store_run_usage(result)# Track token usage# Conversation branchingawaitsession.create_branch_from_turn(2)# Branch from turn 2

fromagents.extensions.memoryimportEncryptedSession,SQLAlchemySession# Create underlying sessionunderlying_session=SQLAlchemySession.from_url("user_123",url="sqlite+aiosqlite:///conversations.db",create_tables=True)# Wrap with encryption and TTLsession=EncryptedSession(session_id="user_123",underlying_session=underlying_session,encryption_key="your-secret-key",ttl=600# 10 minutes)result=awaitRunner.run(agent,"Hello",session=session)

fromagentsimportAgent,Runner,SQLiteSessionagent=Agent(name="Assistant")# Different sessions maintain separate conversation historiessession_1=SQLiteSession("user_123","conversations.db")session_2=SQLiteSession("user_456","conversations.db")result1=awaitRunner.run(agent,"Help me with my account",session=session_1)result2=awaitRunner.run(agent,"What are my charges?",session=session_2)

# Different agents can share the same sessionsupport_agent=Agent(name="Support")billing_agent=Agent(name="Billing")session=SQLiteSession("user_123")# Both agents will see the same conversation historyresult1=awaitRunner.run(support_agent,"Help me with my account",session=session)result2=awaitRunner.run(billing_agent,"What are my charges?",session=session)

importasynciofromagentsimportAgent,Runner,SQLiteSessionasyncdefmain():# Create an agentagent=Agent(name="Assistant",instructions="Reply very concisely.",)# Create a session instance that will persist across runssession=SQLiteSession("conversation_123","conversation_history.db")print("=== Sessions Example ===")print("The agent will remember previous messages automatically.\n")# First turnprint("First turn:")print("User: What city is the Golden Gate Bridge in?")result=awaitRunner.run(agent,"What city is the Golden Gate Bridge in?",session=session)print(f"Assistant:{result.final_output}")print()# Second turn - the agent will remember the previous conversationprint("Second turn:")print("User: What state is it in?")result=awaitRunner.run(agent,"What state is it in?",session=session)print(f"Assistant:{result.final_output}")print()# Third turn - continuing the conversationprint("Third turn:")print("User: What's the population of that state?")result=awaitRunner.run(agent,"What's the population of that state?",session=session)print(f"Assistant:{result.final_output}")print()print("=== Conversation Complete ===")print("Notice how the agent remembered the context from previous turns!")print("Sessions automatically handles conversation history.")if__name__=="__main__":asyncio.run(main())

fromagents.memory.sessionimportSessionABCfromagents.itemsimportTResponseInputItemfromtypingimportListclassMyCustomSession(SessionABC):"""Custom session implementation following the Session protocol."""def__init__(self,session_id:str):self.session_id=session_id# Your initialization hereasyncdefget_items(self,limit:int|None=None)->List[TResponseInputItem]:"""Retrieve conversation history for this session."""# Your implementation herepassasyncdefadd_items(self,items:List[TResponseInputItem])->None:"""Store new items for this session."""# Your implementation herepassasyncdefpop_item(self)->TResponseInputItem|None:"""Remove and return the most recent item from this session."""# Your implementation herepassasyncdefclear_session(self)->None:"""Clear all items for this session."""# Your implementation herepass# Use your custom sessionagent=Agent(name="Assistant")result=awaitRunner.run(agent,"Hello",session=MyCustomSession("my_session"))