Building human handoff interfaces for AI-driven insurance agents using Parlant and Simplification

Human handoffs are a key component of customer service automation – ensuring that when AI reaches its limits, skilled humans can take over seamlessly. In this tutorial, we will implement a human handoff system for an AI-driven insurance agent using Parlant. You’ll learn how to create a simplification-based interface that allows human operators (Tier 2) to view live customer messages and respond directly within the same session, bridging the gap between automation and human expertise. Check The complete code is here.

Set dependencies

Before starting, make sure you have a valid OpenAI API key. After generating it from the OpenAI dashboard, create an .ENV file in the root of the project and store the key there:

OPENAI_API_KEY=your_api_key_here

This keeps your credentials secure and prevents them from being hardcoded into your codebase.

pip install parlant dotenv streamlit

Insurance Agency (Agent)

We will start by building the agent script, which defines the AI’s behavior, conversational travel, vocabulary, and human handoff mechanisms. This will form the core logic that powers our insurance assistants. Once agents are ready and able to upgrade to manual mode, we will continue to develop a simplified human-based handoff interface where human operators can view ongoing conversations, read customer messages and respond in real time – establishing seamless collaboration between AI automation and human expertise. Check The complete code is here.

Load required libraries

import asyncio
import os
from datetime import datetime
from dotenv import load_dotenv
import parlant.sdk as p

load_dotenv()

Tools for defining agents

@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 the interactions an insurance assistant might need.

this get_open_claims Tool represents an asynchronous function that retrieves a list of open insurance claims, allowing agents to provide users with up-to-date information on pending or approved claims.
this file_claim The tool accepts claim details as input and simulates the process of filing a new insurance claim, returning a confirmation message to the user.

at last, get_policy_details The tool provides basic policy information, such as policy number and coverage, allowing agents to accurately answer questions about coverage. Check The complete code is 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
        }
    )

this initiate_human_handoff Tools enable AI agents to gracefully transfer conversations to human operators when they identify an issue that requires human intervention. By switching the session to manual mode, it suspends all automated responses to ensure the human agent has full control. The tool helps maintain a smooth transition between AI and human assistance to ensure complex or sensitive customer queries are handled with the appropriate expertise.

definition glossary

The glossary defines key terms and phrases that the AI agent should always recognize and respond to. It helps maintain accuracy and brand consistency by providing agents with clear, predefined answers to domain-specific queries. Check The complete code is 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",
    )

Define the journey

# ---------------------------
# 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

this Claim journey Guide clients in filing new insurance claims. It collects incident details, triggers the claim filing tool, confirms successful submission, and then ends the journey – automatically initiating the entire claim’s initiation flow.

this policy journey Help customers understand their coverage by retrieving policy details and clearly explaining their coverage. It also includes a guide to ensure AI avoids providing legal interpretations and maintains compliance and professionalism. Check The complete code is here.

Define 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())

Run agent

This will start the Parlant agent locally at http://localhost:8800 which will handle all conversation logic and session management.

In the next step, we will connect this running agent to a simplified human-based handoff interface, allowing human operators to seamlessly connect and manage live conversations using Parlant Session IDs. Check The complete code is here.

Human shortening (handoff.py)

Import library

import asyncio
import streamlit as st
from datetime import datetime
from parlant.client import AsyncParlantClient

Set up Parlant client

After the AI agent script is run, Parlant will host its server locally (usually at http://localhost:8800).

Here we connect to this running instance by creating an asynchronous client. Check The complete code is here.

client = AsyncParlantClient(base_url="http://localhost:8800")

When we run the agent and get the session ID, we will use that ID in this UI to connect and manage that specific conversation.

Session state management

SESSION_STATE of STRAMLIT is used for data across user interactions, such as storing received messages and tracking the latest event offset to efficiently obtain new events. Check The complete code is 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 feature controls how messages appear in a simplified interface – for clarity, customers, AI and human agents are distinguished. Check The complete code is 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}")

Extract events from Parlant

This asynchronous function retrieves new messages (events) from Parlant from the given session.

Each event represents a message in the conversation – whether sent by a customer, AI or human operator. Check The complete code is 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}")

Send messages as a human or AI

Two helper functions are defined to send messages:

One as a human operator (source=”human_agent”)
The other one seems to be sent by the AI but manually triggered by a human (source=”human_agent_on_behalf_of_ai_agent”)
Check The complete code is 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

Simplified interface

Finally, we built a simple interactive simplified UI:

Enter session ID (from Parlant UI)
View chat history
Send messages as a human or AI
Refresh new messages
Check The complete code is 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 The complete code is here. Feel free to check out our Github page for tutorials, code and notebooks. Also, please feel free to follow us twitter And don’t forget to join our 100K+ ml subreddit and subscribe our newsletter. wait! Are you on Telegram? Now you can also join us on Telegram.

I am a Civil Engineering graduate (2022) from Jamia Millia Islamia, New Delhi and I am very interested in data science, especially neural networks and their applications in various fields.

🙌 Follow Marktechpost: Add us as your go-to source on Google.

Source link