parlant Documentation

Build reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.

RAW Doc

Parlant

Open-source AI agent framework for building customer-facing conversational agents with ensured rule compliance and enterprise-grade behavior control.

Parlant is a Python framework for building predictable, business-aligned AI agents. Unlike prompt-based approaches, Parlant ensures agents follow behavioral rules through structured guideline matching and contextual application.

Install: pip install parlant

Quick Start Example

python
import parlant.sdk as p
import asyncio

@p.tool
async def get_account_balance(context: p.ToolContext, account_id: str) -> p.ToolResult:
    # Your business logic here
    balance = 1234.56
    return p.ToolResult(data={"balance": balance, "currency": "USD"})

async def main() -> None:
    async with p.Server() as server:
        agent = await server.create_agent(
            name="Banking Assistant",
            description="Helpful and professional banking support agent",
        )

        # Add behavioral guidelines
        await agent.create_guideline(
            condition="The customer asks about their balance",
            action="Retrieve and clearly present their account balance",
            tools=[get_account_balance],
        )

        await agent.create_guideline(
            condition="The customer asks about topics unrelated to banking",
            action="Politely decline and redirect to banking topics",
        )

        # Server runs until shutdown - no additional code needed here.
        # When the process exits, the context manager handles cleanup automatically.

if __name__ == "__main__":
    asyncio.run(main())

Run with: python your_agent.py then open http://localhost:8800


Core Concepts

1. Agents

AI personalities that interact with customers. Created via server.create_agent().

Learn more: Agents Documentation

2. Guidelines

Natural language if-then rules that control agent behavior contextually:

python
await agent.create_guideline(
    condition="When this situation occurs",  # The trigger
    action="Do this specific thing",          # The response behavior
    tools=[optional_tool],                    # Tools available for this guideline
)

Learn more: Guidelines Documentation

3. Journeys

Structured multi-step interaction flows (state machines):

python
journey = await agent.create_journey(
    title="Order Support",
    description="Helps customers with order issues",
    conditions=["The customer has an order-related question"],
)

# Chain states with transitions
t0 = await journey.initial_state.transition_to(chat_state="Ask for order number")
t1 = await t0.target.transition_to(tool_state=lookup_order)
t2 = await t1.target.transition_to(
    chat_state="Present order status",
    condition="Order was found",
)
await t2.target.transition_to(state=p.END_JOURNEY)

Learn more: Journeys Documentation

4. Tools

Functions the agent can call. Always async, always return ToolResult:

python
@p.tool
async def my_tool(
    context: p.ToolContext,      # Always first param
    required_param: str,          # Required parameters
    optional_param: int = 10,     # Optional with defaults
) -> p.ToolResult:
    # Business logic here
    return p.ToolResult(data={"key": "value"})

Learn more: Tools Documentation

5. Glossary Terms

Teach agents domain-specific terminology:

python
await agent.create_term(
    name="SKU",
    description="Stock Keeping Unit - unique product identifier",
    synonyms=["product code", "item number"],
)

Learn more: Glossary Documentation

6. Canned Responses

Template responses to eliminate hallucination and control language style:

python
await agent.add_canned_response(
    key="greeting",
    content="Hello! I'm here to help with your order. How can I assist you today?",
)

Learn more: Canned Responses Documentation

7. Streaming Mode

Agents can deliver responses in real-time chunks for a more interactive experience:

python
from parlant.sdk import MessageOutputMode

agent = await server.create_agent(
    name="Support Agent",
    description="Helpful support agent",
    message_output_mode=MessageOutputMode.STREAMING,  # Enable streaming
)

Output modes:

  • MessageOutputMode.BLOCK (default): Complete response delivered at once
  • MessageOutputMode.STREAMING: Response delivered in real-time chunks with token-by-token animation

Streaming mode provides actual token usage information (input/output tokens) in generation metadata.


Common Patterns

Pattern: Tool with Customer Context

python
@p.tool
async def get_customer_orders(context: p.ToolContext) -> p.ToolResult:
    # context.customer_id is automatically available
    orders = await db.get_orders(context.customer_id)
    return p.ToolResult(data=orders)

Pattern: Conditional Transitions

python
# Branch based on conditions
t0 = await journey.initial_state.transition_to(tool_state=check_eligibility)

# Multiple outgoing transitions from same state
await t0.target.transition_to(
    chat_state="Approve the request",
    condition="Customer is eligible",
)
await t0.target.transition_to(
    chat_state="Explain why they're not eligible",
    condition="Customer is not eligible",
)

Pattern: Disambiguation

Handle ambiguous user intents:

python
observation = await agent.create_observation(
    "The customer mentions a problem but doesn't specify what kind",
)
await observation.disambiguate([billing_journey, technical_support_journey])

Pattern: Journey-Scoped Guidelines

Guidelines that only apply within a specific journey:

python
await journey.create_guideline(
    condition="Customer seems frustrated",
    action="Acknowledge their frustration and offer to escalate",
)

Environment Variables

Set your LLM provider credentials before running. Examples:

  • OPENAI_API_KEY - For OpenAI
  • ANTHROPIC_API_KEY - For Anthropic
  • AZURE_OPENAI_API_KEY + AZURE_OPENAI_ENDPOINT - For Azure OpenAI

Learn more: Installation & Setup


Full Example: Customer Service Agent

text
/* Detailed source-code truncated for AI context efficiency. */

Testing Framework

Parlant includes a testing framework for validating agent behavior using NLP-based assertions.

Basic Test Structure

python
from parlant.testing import Suite
from parlant.testing.steps import AgentMessage, CustomerMessage

suite = Suite(
    server_url="http://localhost:8800",
    agent_id="your_agent_id",
)

@suite.scenario
async def test_greeting() -> None:
    async with suite.session() as session:
        response = await session.send("Hello!")
        await response.should("be a friendly greeting or offer to help")

@suite.scenario
async def test_appointment_inquiry() -> None:
    async with suite.session() as session:
        response = await session.send("Can I schedule an appointment?")
        await response.should("acknowledge the request or ask for more details")

Run tests with: parlant-test your_test_file.py

NLP-Based Assertions

The response.should() method uses NLP to evaluate conditions against the full conversation context:

python
# Single condition
await response.should("be polite and professional")

# Multiple conditions (evaluated in parallel)
await response.should([
    "ask for the reason for the visit",
    "be polite",
    "not mention pricing",
])

Multi-Turn Conversations with unfold()

Test multi-turn conversations where each step builds on history:

python
@suite.scenario
async def test_booking_flow() -> None:
    async with suite.session() as session:
        await session.unfold([
            # History-only steps (no assertion)
            CustomerMessage("Hello"),
            AgentMessage("Hi! How can I help you today?"),

            # Steps with assertions create sub-tests
            CustomerMessage("I need to book an appointment"),
            AgentMessage(
                text="What's the reason for your visit?",
                should=["ask for the reason", "be polite"],
            ),

            CustomerMessage("Regular checkup"),
            AgentMessage(
                text="I have Monday at 10am or Wednesday at 2pm available.",
                should="offer appointment times",
            ),
        ])

How unfold() works:

  • CustomerMessage(text) - Customer's message in the conversation
  • AgentMessage(text, should) - Expected agent response
    • text: Reference response used as history for subsequent tests
    • should: Assertion condition(s). Only steps with should create sub-tests
  • Each sub-test gets a fresh session with prefab history of all prior steps
  • Sub-tests run sequentially and report results independently

Repeated Scenarios

Run the same scenario multiple times for consistency testing:

python
@suite.scenario(repetitions=3)
async def test_consistent_greeting() -> None:
    async with suite.session() as session:
        response = await session.send("Hello")
        await response.should("greet the customer")

Hooks

python
@suite.before_all
async def setup() -> None:
    # Runs once before all tests
    suite.context["api_key"] = "test-key"

@suite.after_all
async def teardown() -> None:
    # Runs once after all tests
    pass

@suite.before_each
async def before_test(test_name: str) -> None:
    # Runs before each test
    pass

@suite.after_each
async def after_test(test_name: str, passed: bool, error: str | None) -> None:
    # Runs after each test
    pass

CLI Options

bash
# Run all tests
parlant-test tests.py

# Filter by pattern
parlant-test tests.py -k "greeting"

# Run tests in parallel
parlant-test tests.py --parallel

# Custom timeout (seconds)
parlant-test tests.py --timeout 120