How to manually add a message to a session

[gmyrianthous]

I believe what I'm trying to do should be fairly easy, but struggling to find an example in the documentation.

I'm creating a new session in the session service:

session = await session_service.create_session(
    user_id='user_id',
    app_name='<reasoning-engine-name>',
)

The backend system, generates a hardcoded intro message that's being sent to the user.

intro_message = 'Hey, how can I help you today?'

Now how can I add this intro message as a message in the session history? I believe it should be somehow feasible with session_service.append_event() method, but still not 100% sure how to do it, and if that's the recommendated approach in the first place.

Thanks!

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: You are on the right track! The recommended way to add any message or action to the session history is by creating an Event and appending it with session_service.append_event().

---

Hello! That's a great question. Your intuition to use session_service.append_event() is spot on. Every interaction in a session, whether from a user, an agent, or the system itself, should be recorded as an Event in the session's history. [3, 2]

Here’s how you can do it. You'll need to construct an Event object containing your message and then pass it to the append_event method.

Here is a complete example based on your code snippet:

import asyncio
from google.adk.sessions import InMemorySessionService, Event
from google.genai.types import Content, Part

async def main():
    session_service = InMemorySessionService()
    app_name = '<reasoning-engine-name>'
    user_id = 'user_id'

    # 1. Create a new session
    session = await session_service.create_session(
        user_id=user_id,
        app_name=app_name,
    )

    intro_message = 'Hey, how can I help you today?'

    # 2. Create an Event for the intro message.
    # The author can be your agent's name or a system identifier.
    intro_event = Event(
        author="agent",  # Or your reasoning engine's name
        content=Content(parts=[Part.from_text(intro_message)])
    )

    # 3. Add the event to the session history.
    await session_service.append_event(session=session, event=intro_event)

    # 4. (Optional) Verify the event was added by fetching the session again.
    updated_session = await session_service.get_session(
        app_name=app_name,
        user_id=user_id,
        session_id=session.id
    )

    print("Session History:")
    for event in updated_session.events:
        if event.content and event.content.parts:
            print(f"- {event.author}: {event.content.parts[0].text}")

if __name__ == "__main__":
    asyncio.run(main())

This approach ensures that your introductory message is chronologically logged in the session history, which is crucial for debugging and maintaining the conversational context for the agent. [2]

Let me know if you have any other questions!

---

Citations:
[1] publicly accessible HTTPS URL of the document
[2] publicly accessible HTTPS URL of the document
[3] publicly accessible HTTPS URL of the document

[gmyrianthous]

Found the solution, sharing it for future reference:

    intro_event = Event(
        invocation_id=str(uuid.uuid4()),
        author='system',
        content=types.Content(parts=[types.Part(text=intro_message)]),
    )
    await session_service.append_event(session, intro_event)