Streamlit dashboard with Marvin agent

[mattijsdp]

The following is a somewhat minimal example of creating a Streamlit dashboard where you can chat to a Marvin agent. This way you can easily have non-technical folks play with your agent. Hope it's useful for somebody!

import streamlit as st
import marvin
from marvin.handlers import Handler
import uuid
from marvin.engine.events import Event
from typing import Callable


EVENT_TYPES_KNOWN_BUT_NOT_DISPLAYED = [
    "orchestrator-start",
    "orchestrator-end",
    "actor-start-turn",
    "actor-end-turn",
    "tool-call-delta",
    "end-turn-tool-call",
]


class CallbackHandler(Handler):
    def __init__(self, callback: Callable):
        super().__init__()
        self.callback = callback

    def on_event(self, event: Event):
        self.callback(event)


def render_event(event):
    event_type = event.type

    if event_type == "user-message":
        with st.chat_message("user"):
            for single_message in event.message:
                st.markdown(single_message)
            st.caption(f"Time: {event.timestamp.isoformat()}")

    elif event_type == "actor-message":
        with st.chat_message("assistant"):
            st.markdown(f"{event.message.content}")
            st.caption(f"Time: {event.timestamp.isoformat()}")

    elif event_type == "tool-call":
        # Use assistant message for tool calls as they originate from the agent
        if event.message.tool_name == "cli":
            with st.chat_message("assistant"):
                st.markdown(f"{event.message.args_as_dict()['message']}")
                st.caption(f"Time: {event.timestamp.isoformat()}")
        else:
            with st.chat_message("tool", avatar="🛠️"):
                st.markdown(f"Tool Call: `{event.message.tool_name}`")
                st.json(event.message.args)  # Input is already a JSON string
                st.caption(f"Time: {event.timestamp.isoformat()}")

    elif event_type == "tool-retry":
        with st.chat_message("tool", avatar="🛠️"):
            st.markdown(f"Tool retry: `{event.message.content}`")

    elif event_type == "tool-result":
        with st.chat_message("tool", avatar="🛠️"):
            with st.expander(f"Tool Result: `{event.message.tool_name}`"):
                st.markdown("**Output:**")
                if isinstance(event.message.content, str):
                    st.markdown(f"Is a string: {event.message.content}")
                else:
                    st.json(event.message.content, expanded=1)
            st.caption(f"Time: {event.timestamp.isoformat()}")

    elif event_type == "end-turn-tool-result":
        with st.chat_message("assistant"):
            st.markdown(f"{event.result.data.result}")

    elif event_type in EVENT_TYPES_KNOWN_BUT_NOT_DISPLAYED:
        pass

    else:
        raise ValueError(f"Unknown event type: {event_type}. Event: {event}")


def render_events():
    """Renders the events stored in st.session_state.events."""
    for event in st.session_state.events:
        render_event(event)


def add_event_to_session_state(event):
    st.session_state.events.append(event)
    render_event(event)


def disable_chat_input():
    st.session_state.chat_input_disabled = True


@st.cache_resource
def get_agent():
    agent = marvin.Agent(name="Marvin")
    return agent


# --- 4. Main Streamlit App ---
def main():
    st.set_page_config(layout="wide")
    st.title("Marvin Agent Demo")

    # Initialize session state variables
    if "events" not in st.session_state:
        st.session_state.events = []
    if "thread_id" not in st.session_state:
        st.session_state.thread_id = str(uuid.uuid4())
    if "chat_input_disabled" not in st.session_state:
        st.session_state.chat_input_disabled = False

    agent = get_agent()

    # Sidebar for controls
    with st.sidebar:
        st.header("Controls")
        if st.button("Reset Conversation"):
            st.session_state.events = []
            st.rerun()

    # Main area for conversation display
    st.header("Conversation")
    render_events()  # Render events on every run

    callback_handler = CallbackHandler(callback=add_event_to_session_state)
    handlers = [callback_handler]

    thread = marvin.Thread(id=st.session_state.thread_id)
    with thread:
        prompt = st.chat_input(
            "Enter your prompt:",
            disabled=st.session_state.chat_input_disabled,
            on_submit=disable_chat_input,
        )
        if prompt:
            agent.run(
                prompt, handlers=handlers, prompt_template="{{ task.instructions }}"
            )
            st.session_state.chat_input_disabled = False
            st.rerun()


if __name__ == "__main__":
    main()

[zzstoatzz]

cool - thanks for sharing @mattijsdp !