parlant

# 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](https://parlant.io/docs/concepts/agents)

### 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](https://parlant.io/docs/concepts/customization/guidelines)

### 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](https://parlant.io/docs/concepts/customization/journeys)

### 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](https://parlant.io/docs/concepts/customization/tools)

### 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](https://parlant.io/docs/concepts/customization/glossary)

### 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](https://parlant.io/docs/concepts/customization/canned-responses)

### 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](https://parlant.io/docs/quickstart/installation)

---

## Full Example: Customer Service Agent

```python
import parlant.sdk as p
import asyncio

# Define tools
@p.tool
async def lookup_order(context: p.ToolContext, order_id: str) -> p.ToolResult:
    # Simulated order lookup
    return p.ToolResult(data={
        "order_id": order_id,
        "status": "shipped",
        "tracking": "1Z999AA10123456784",
    })

@p.tool
async def request_refund(context: p.ToolContext, order_id: str, reason: str) -> p.ToolResult:
    return p.ToolResult(data={"refund_id": "REF-12345", "status": "processing"})

async def create_order_journey(agent: p.Agent) -> p.Journey:
    journey = await agent.create_journey(
        title="Order Support",
        description="Helps customers check order status or request refunds",
        conditions=["Customer asks about an order"],
    )

    # Step 1: Get order number
    t0 = await journey.initial_state.transition_to(
        chat_state="Ask the customer for their order number"
    )

    # Step 2: Look up the order
    t1 = await t0.target.transition_to(tool_state=lookup_order)

    # Step 3a: Order found - show status
    t2 = await t1.target.transition_to(
        chat_state="Present the order status and tracking information",
        condition="Order was found",
    )

    # Step 3b: Order not found
    await t1.target.transition_to(
        chat_state="Apologize and ask them to verify the order number",
        condition="Order was not found",
    )

    # Step 4: Check if they need anything else
    t3 = await t2.target.transition_to(
        chat_state="Ask if they need help with anything else regarding this order"
    )

    # Step 5a: They want a refund
    t4 = await t3.target.transition_to(
        tool_state=request_refund,
        condition="Customer requests a refund",
    )
    await t4.target.transition_to(
        chat_state="Confirm the refund has been initiated"
    )

    # Step 5b: They're satisfied
    await t3.target.transition_to(
        state=p.END_JOURNEY,
        condition="Customer has no more questions",
    )

    # Journey-specific guidelines
    await journey.create_guideline(
        condition="Customer is upset about a delayed order",
        action="Apologize sincerely and offer expedited shipping on their next order",
    )

    return journey

async def main() -> None:
    async with p.Server() as server:
        agent = await server.create_agent(
            name="Support Agent",
            description="Friendly and efficient customer support representative",
        )

        # Domain knowledge
        await agent.create_term(
            name="Express Shipping",
            description="2-day delivery, costs $9.99",
        )

        # Create journey
        await create_order_journey(agent)

        # Global guidelines (apply everywhere)
        await agent.create_guideline(
            condition="Customer uses profanity or is abusive",
            action="Calmly ask them to be respectful, or offer to end the conversation",
        )

        await agent.create_guideline(
            condition="Customer asks to speak to a human",
            action="Provide the support phone number: 1-800-555-0123",
        )

        # 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())
```

---

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

---

## Links

- Documentation: https://parlant.io/
- GitHub: https://github.com/emcie-co/parlant
- PyPI: https://pypi.org/project/parlant/
- Discord: https://discord.gg/duxWqxKk6J