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:
OPENAI_API_KEY=your_api_key_hereThis keeps your credentials safe and prevents them from being hardcoded into your codebase.
pip install parlant dotenv streamlitInsurance Agent (agent.py)
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.
Loading the required libraries
import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
import parlant.sdk as p
load_dotenv()Defining the Agent’s Tools
@p.tool
async def get_open_claims(context: p.ToolContext) -> p.ToolResult:
return p.ToolResult(data=["Claim #123 - Pending", "Claim #456 - Approved"])
@p.tool
async def file_claim(context: p.ToolContext, claim_details: str) -> p.ToolResult:
return p.ToolResult(data=f"New claim filed: {claim_details}")
@p.tool
async def get_policy_details(context: p.ToolContext) -> p.ToolResult:
return p.ToolResult(data={
"policy_number": "POL-7788",
"coverage": "Covers accidental damage and theft up to $50,000"
})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.tool
async 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",
)Defining the Journeys
# ---------------------------
# 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 journeyThe 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.
Defining the Main Runner
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())
Running the Agent
python agent.pyThis 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.
Human Handoff (handoff.py)
Importing Libraries
import asyncio
import streamlit as st
from datetime import datetime
from parlant.client import AsyncParlantClientSetting Up the Parlant Client
Once the AI agent script is running, Parlant will host its server locally (usually at http://localhost:8800).
Here, we connect to that running instance by creating an asynchronous client. Check out the FULL CODES here.
client = AsyncParlantClient(base_url="http://localhost:8800")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 = 0Message 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.
def render_message(message, source, participant_name, timestamp):
if source == "customer":
st.markdown(f"**
Customer [{timestamp}]:** {message}")
elif source == "ai_agent":
st.markdown(f"**
AI [{timestamp}]:** {message}")
elif source == "human_agent":
st.markdown(f"**
{participant_name} [{timestamp}]:** {message}")
elif source == "human_agent_on_behalf_of_ai_agent":
st.markdown(f"**
(Human as AI) [{timestamp}]:** {message}")Fetching Events from Parlant
This asynchronous function retrieves new messages (events) from Parlant for the given session.
Each event represents a message in the conversation — whether sent by the customer, AI, or human operator. Check out the FULL CODES here.
async def fetch_events(session_id):
try:
events = await client.sessions.list_events(
session_id=session_id,
kinds="message",
min_offset=st.session_state.last_offset,
wait_for_data=5
)
for event in events:
message = event.data.get("message")
source = event.source
participant_name = event.data.get("participant", {}).get("display_name", "Unknown")
timestamp = getattr(event, "created", None) or event.data.get("created", "Unknown Time")
event_id = getattr(event, "id", "Unknown ID")
st.session_state.events.append(
(message, source, participant_name, timestamp, event_id)
)
st.session_state.last_offset = max(st.session_state.last_offset, event.offset + 1)
except Exception as e:
st.error(f"Error fetching events: {e}")Sending Messages as Human or AI
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.
async def send_human_message(session_id: str, message: str, operator_name: str = "Tier-2 Operator"):
event = await client.sessions.create_event(
session_id=session_id,
kind="message",
source="human_agent",
message=message,
participant={
"id": "operator-001",
"display_name": operator_name
}
)
return event
async def send_message_as_ai(session_id: str, message: str):
event = await client.sessions.create_event(
session_id=session_id,
kind="message",
source="human_agent_on_behalf_of_ai_agent",
message=message
)
return event
Streamlit Interface
Finally, we build a simple, interactive Streamlit UI:
- Enter a session ID (from the Parlant UI)
- View chat history
- Send messages as either Human or AI
- Refresh to pull new messages
- 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))
Check out the FULL CODES here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post Building a Human Handoff Interface for AI-Powered Insurance Agent Using Parlant and Streamlit appeared first on MarkTechPost.
