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
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:
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):
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:
@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:
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:
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:
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 onceMessageOutputMode.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
@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
# 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:
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:
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 OpenAIANTHROPIC_API_KEY- For AnthropicAZURE_OPENAI_API_KEY+AZURE_OPENAI_ENDPOINT- For Azure OpenAI
Learn more: Installation & Setup
Full Example: Customer Service Agent
/* 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
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:
# 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:
@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 conversationAgentMessage(text, should)- Expected agent responsetext: Reference response used as history for subsequent testsshould: Assertion condition(s). Only steps withshouldcreate 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:
@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
@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
passCLI Options
# 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 120Links
- Documentation: https://parlant.io/
- GitHub: https://github.com/emcie-co/parlant
- PyPI: https://pypi.org/project/parlant/
- Discord: https://discord.gg/duxWqxKk6J