Human handoff is a key component of customer service automation—it ensures that when AI reaches its limits, a skilled human can seamlessly take over. In this tutorial, we’ll implement a human handoff system for an AI-powered insurance agent using Parlant. You’ll learn how to create a Streamlit-based interface that allows a human operator (Tier 2) to view live customer messages and respond directly within the same session, bridging the gap between automation and human expertise. Check out the FULL CODES here.
Setting up the dependencies
Make sure you have a valid OpenAI API key before starting. Once you’ve generated it from your OpenAI dashboard, create a .env file in your project’s root directory and store the key securely there like this:
We’ll start by building the agent script, which defines the AI’s behavior, conversation journeys, glossary, and the human handoff mechanism. This will form the core logic that powers our insurance assistant in Parlant. Once the agent is ready and capable of escalating to manual mode, we’ll move on to developing the Streamlit-based human handoff interface, where human operators can view ongoing sessions, read customer messages, and respond in real time — creating a seamless collaboration between AI automation and human expertise. Check out the FULL CODES here.
The code block introduces three tools that simulate interactions an insurance assistant might need.
The get_open_claims tool represents an asynchronous function that retrieves a list of open insurance claims, allowing the agent to provide users with up-to-date information about pending or approved claims. The file_claim tool accepts claim details as input and simulates the process of filing a new insurance claim, returning a confirmation message to the user.
Finally, the get_policy_details tool provides essential policy information, such as the policy number and coverage limits, enabling the agent to respond accurately to questions about insurance coverage. Check out the FULL CODES here.
@p.toolasync def initiate_human_handoff(context: p.ToolContext, reason: str) -> p.ToolResult: """ Initiate handoff to a human agent when the AI cannot adequately help the customer. """ print(f" Initiating human handoff: {reason}") # Setting session to manual mode stops automatic AI responses return p.ToolResult( data=f"Human handoff initiated because: {reason}", control={ "mode": "manual" # Switch session to manual mode } )
The initiate_human_handoff tool enables the AI agent to gracefully transfer a conversation to a human operator when it detects that the issue requires human intervention. By switching the session to manual mode, it pauses all automated responses, ensuring the human agent can take full control. This tool helps maintain a smooth transition between AI and human assistance, ensuring complex or sensitive customer queries are handled with the appropriate level of expertise.
Defining the Glossary
A glossary defines key terms and phrases that the AI agent should recognize and respond to consistently. It helps maintain accuracy and brand alignment by giving the agent clear, predefined answers for common domain-specific queries. Check out the FULL CODES here.
async def add_domain_glossary(agent: p.Agent): await agent.create_term( name="Customer Service Number", description="You can reach us at +1-555-INSURE", ) await agent.create_term( name="Operating Hours", description="We are available Mon-Fri, 9AM-6PM", )
# ---------------------------# Claim Journey# ---------------------------async def create_claim_journey(agent: p.Agent) -> p.Journey: journey = await agent.create_journey( title="File an Insurance Claim", description="Helps customers report and submit a new claim.", conditions=["The customer wants to file a claim"], ) s0 = await journey.initial_state.transition_to(chat_state="Ask for accident details") s1 = await s0.target.transition_to(tool_state=file_claim, condition="Customer provides details") s2 = await s1.target.transition_to(chat_state="Confirm claim was submitted", condition="Claim successfully created") await s2.target.transition_to(state=p.END_JOURNEY, condition="Customer confirms submission") return journey# ---------------------------# Policy Journey# ---------------------------async def create_policy_journey(agent: p.Agent) -> p.Journey: journey = await agent.create_journey( title="Explain Policy Coverage", description="Retrieves and explains customer's insurance coverage.", conditions=["The customer asks about their policy"], ) s0 = await journey.initial_state.transition_to(tool_state=get_policy_details) await s0.target.transition_to( chat_state="Explain the policy coverage clearly", condition="Policy info is available", ) await agent.create_guideline( condition="Customer presses for legal interpretation of coverage", action="Politely explain that legal advice cannot be provided", ) return journey
The Claim Journey guides customers through the process of filing a new insurance claim. It collects accident details, triggers the claim filing tool, confirms successful submission, and then ends the journey—automating the entire claim initiation flow.
The Policy Journey helps customers understand their insurance coverage by retrieving policy details and explaining them clearly. It also includes a guideline to ensure the AI avoids giving legal interpretations, maintaining compliance and professionalism. Check out the FULL CODES here.
async def main(): async with p.Server() as server: agent = await server.create_agent( name="Insurance Support Agent", description=( "Friendly Tier-1 AI assistant that helps with claims and policy questions. " "Escalates complex or unresolved issues to human agents (Tier-2)." ), ) # Add shared terms & definitions await add_domain_glossary(agent) # Journeys claim_journey = await create_claim_journey(agent) policy_journey = await create_policy_journey(agent) # Disambiguation rule status_obs = await agent.create_observation( "Customer mentions an issue but doesn't specify if it's a claim or policy" ) await status_obs.disambiguate([claim_journey, policy_journey]) # Global Guidelines await agent.create_guideline( condition="Customer asks about unrelated topics", action="Kindly redirect them to insurance-related support only", ) # Human Handoff Guideline await agent.create_guideline( condition="Customer requests human assistance or AI is uncertain about the next step", action="Initiate human handoff and notify Tier-2 support.", tools=[initiate_human_handoff], ) print(" Insurance Support Agent with Human Handoff is ready! Open the Parlant UI to chat.")if __name__ == "__main__": asyncio.run(main())
This will start the Parlant agent locally on http://localhost:8800 , where it will handle all conversation logic and session management.
In the next step, we’ll connect this running agent to our Streamlit-based Human Handoff interface, allowing a human operator to seamlessly join and manage live conversations using the Parlant session ID. Check out the FULL CODES here.
When you run the agent and get a session ID, we’ll use that ID in this UI to connect and manage that specific conversation.
Session State Management
Streamlit’s session_state is used to persist data across user interactions — such as storing received messages and tracking the latest event offset to fetch new ones efficiently. Check out the FULL CODES here.
if "events" not in st.session_state: st.session_state.events = []if "last_offset" not in st.session_state: st.session_state.last_offset = 0
Message Rendering Function
This function controls how messages appear in the Streamlit interface — differentiating between customers, AI, and human agents for clarity. Check out the FULL CODES here.
Two helper functions are defined to send messages:
One as a human operator (source=”human_agent”)Another as if sent by the AI, but manually triggered by a human (source=”human_agent_on_behalf_of_ai_agent”)Check out the FULL CODES here.
st.title(" Human Handoff Assistant")session_id = st.text_input("Enter Parlant Session ID:")if session_id: st.subheader("Chat History") if st.button("Refresh Messages"): asyncio.run(fetch_events(session_id)) for msg, source, participant_name, timestamp, event_id in st.session_state.events: render_message(msg, source, participant_name, timestamp) st.subheader("Send a Message") operator_msg = st.text_input("Type your message:") if st.button("Send as Human"): if operator_msg.strip(): asyncio.run(send_human_message(session_id, operator_msg)) st.success("Message sent as human agent ") asyncio.run(fetch_events(session_id)) if st.button("Send as AI"): if operator_msg.strip(): asyncio.run(send_message_as_ai(session_id, operator_msg)) st.success("Message sent as AI ") asyncio.run(fetch_events(session_id))
