Building human handover interfaces using Parlant and simplified AI-powered insurance agents

by · October 7, 2025

Human handover is a key component of customer service automation – ensuring that skilled people can seamlessly take over when AI reaches its limits. In this tutorial, we will implement a human handover system for AI-powered insurance agents using Parlant. You will learn how to create a simplified-based interface that allows human operators (layer 2) to view real-time customer messages and respond directly in the same session, thereby bridging the gap between automation and human expertise. Check The complete code is here.

Set up dependencies

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

OPENAI_API_KEY=your_api_key_here

This ensures your credentials are secure and prevents them from being hardcoded into your code base.

pip install parlant dotenv streamlit

Insurance Agent (Agent)

We will first build a proxy script that defines the behavior of AI, dialogue travel, glossary, and human handover mechanisms. This will form the core logic that supports our insurance assistants. Once the agent is ready and able to upgrade to manual mode, we will continue to develop a simplified human handover interface where human operators can view ongoing sessions, read customer messages and respond in real time – establishing a seamless collaboration between AI automation and human expertise. Check The complete code is here.

Load the required library

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"
    })

Code blocks introduce three tools that simulate interactions that insurance assistants may need.

  • this get_open_claims The tool represents an asynchronous feature that retrieves an open insurance claim list, allowing agents to provide users with the latest information about pending or approved claims.
  • this file_claim The tool accepts claim details as input and simulates the process of filing a new insurance claim and returns a confirmation message to the user.

at last, get_policy_details The tool provides basic policy information, such as policy numbers and coverage, allowing agents to answer questions about insurance coverage accurately. 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 allow AI agents to gracefully transfer conversations to human operators when they discover problems require human intervention. By switching the conversation to manual mode, it pauses all automatic responses to ensure that the human agent has full control. The tool helps maintain a smooth transition between AI and human assistance to ensure complex or sensitive customer inquiries are handled with the right expertise.

Define the glossary

The glossary defines the key terms and phrases that AI agents should always recognize and respond to. It helps maintain accuracy and brand unity by providing agents with clear, predefined, domain-specific query answers. 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 Claims Journey Guide clients to file new insurance claims. It collects accident details, triggers the claim filing tool, confirms successful submission, and then ends the journey – automatically starts the entire claim startup flow.

this Policy Journey Help clients understand their coverage by searching for policy details and clearly explaining their coverage. It also includes a guide to ensure that AI avoids providing legal interpretations, maintaining compliance and professionalism. Check The complete code is 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())

Run the Agent

This will start the Parlant Agent locally, which will handle all conversation logic and session management.

In the next step, we will connect this running agent to a simplified human handover interface, allowing human operators to seamlessly connect and manage real-time conversations using Parlant Session ID. 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

Setting up the Parlant client

After the AI ​​proxy script is run, Parlant will host its server locally (usually) .

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

client = AsyncParlantClient(base_url="")

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

Session State Management

STRAMLIT’s SESSION_STATE is used to interact across users, such as storing received messages and tracking the latest event offsets to effectively acquire 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, distinguish between customers, AI and human agents. 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 a conversation – whether it is a customer, an AI, or a message sent by a 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 a message as a human or artificial intelligence

Two accessibility features are defined to send messages:

  • A as a human operator (source = “human_agent”)
  • Another seems to be sent by AI, but triggered manually by 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

Simplify interfaces

Finally, we build a simple interactive simplified UI:

  • Enter the session ID (from Parlant UI)
  • View chat history
  • Send a message as a human or artificial intelligence
  • Refresh new news
  • 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. Check out ours anytime Tutorials, codes and notebooks for github pages. Also, please stay tuned for us twitter And don’t forget to join us 100K+ ml reddit And subscribe Our newsletter. wait! Are you on the telegram? Now, you can also join us on Telegram.


I am a civil engineering graduate in Islamic Islam in Jamia Milia New Delhi (2022) and I am very interested in data science, especially neural networks and their applications in various fields.

🙌Follow Marktechpost: Add us as the preferred source on Google.

You may also like...

live chat