Skip to main content

Triggered Responses

Triggered responses allow your agent to send messages outside the normal request/reply cycle. This enables proactive, human-like interactions where the agent can reach out to customers without waiting for them to speak first, or to follow-up during long-running operations, or when the customer is idle.

Why Triggered Responses?​

We've all experienced situations where a customer service representative says "Let me check that for you" while looking something up, or a store employee approaches us asking "Is there anything I can help you with?" without us initiating the conversation.

Triggered responses let your AI agents replicate these natural interactions by sending messages whenever your application decides it's appropriate, not just in response to customer input.

Triggering Agent Responses​

Triggering from Within a Tool​

To trigger a response internally from within a tool's long-running background task, see Deferred Follow-Ups below.

Triggering from an External System​

From an external system, you can trigger an agent response by creating a message event with source: "ai_agent" via the client SDK or REST API.

from parlant.client import AsyncParlantClient

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

# Trigger the agent to generate a response
await client.sessions.create_event(
    session_id=SESSION_ID,
    kind="message",
    source="ai_agent",
)

When you trigger an agent response this way, the agent will generate a contextually appropriate message based on the current conversation state and active guidelines.

Guided Responses​

You can guide what the agent says by providing guidelines with the trigger. This is useful when you want the agent to communicate something specific.

await client.sessions.create_event(
    session_id=SESSION_ID,
    kind="message",
    source="ai_agent",
    guidelines=[
        {
            "action": "Ask if the customer needs any more help",
        }
    ],
)

Common Use Cases​

Inactivity Follow-ups​

Trigger a follow-up when the customer hasn't responded for a while:

import asyncio

async def monitor_session(session_id: str):
    while True:
        await asyncio.sleep(30)  # Check every 30 seconds

        session = client.sessions.get(session_id)
        last_event = get_last_customer_message(session)

        if seconds_since(last_event) > 180:
            client.sessions.create_event(
                session_id=session_id,
                kind="message",
                source="ai_agent",
                guidelines=[{
                    "action": "Gently ask if the customer is still there or needs any help",
                }],
            )
            break  # Don't spam follow-ups

External Event Notifications​

Notify the customer when something happens in your system:

async def on_order_shipped(order_id: str, session_id: str):
    client.sessions.create_event(
        session_id=session_id,
        kind="message",
        source="ai_agent",
        guidelines=[{
            "action": f"Inform the customer that order {order_id} has shipped",
        }],
    )

Background Operations with Deferred Follow-up​

For operations that take too long to keep the customer waiting, you can spawn a background task and use Agent.utter() to have the agent follow up when it completes. This pattern lets the tool return immediately while the actual work continues asynchronously:

import asyncio

@p.tool
async def purchase_flight_ticket(
    context: p.ToolContext,
    flight_id: str,
    passenger_name: str,
) -> p.ToolResult:
    """Purchase a flight ticket for the customer."""

    # Important: capture agent and session here, before the background task.
    # .current accessors are only available during tool execution.
    agent = p.Agent.current
    session = p.Session.current

    async def finish_ticket_purchase_in_background():
        try:
            # Perform the actual purchase (may take a while)
            confirmation = await flight_api.purchase(flight_id, passenger_name)

            await agent.utter(
                session=session,
                guidelines=[
                    {"action": "Inform the customer that their flight ticket purchase succeeded."},
                ],
            )
        except Exception as e:
            await agent.utter(
                session=session,
                guidelines=[
                    {"action": "Apologize and explain that the flight ticket purchase failed."},
                ],
            )

    # Create background task with auto-cleanup on completion
    task = asyncio.create_task(finish_ticket_purchase_in_background())
    task.add_done_callback(lambda t: t.result() if not t.cancelled() else None)

    return p.ToolResult(
        data="Flight ticket purchase initiated. I'll let you know once it's confirmed!",
        metadata={"status": "processing", "flight_id": flight_id},
    )

This pattern:

  1. Captures the current agent and session before spawning the background task — .current accessors are only available during tool execution
  2. Defines an async function that performs the long-running operation
  3. Uses agent.utter() to have the agent generate a follow-up message with the provided guidelines
  4. Uses asyncio.create_task() to run the operation without blocking
  5. Returns immediately with an initial update so the agent can respond to the customer

The agent will respond with something like "I've started processing your flight ticket purchase. I'll let you know as soon as it's confirmed!" and then follow up automatically when the background operation completes, either with the confirmation number or an apology if something went wrong.

github Need help with triggered responses?