Workflow Handlers

Workflow handlers allow you to intercept user messages and execute deterministic workflows before they reach the LLM. This is the first extensibility point in the agent’s message processing pipeline.

WorkflowHandler Interface

All workflow handlers extend the WorkflowHandler base class:

from vanna.core.workflow import WorkflowHandler, TriggerResult
from vanna.core.user import User
from vanna.core.storage import Conversation
from vanna.components import UiComponent

class WorkflowHandler(ABC):
    async def try_handle(
        self,
        agent: Agent,
        user: User,
        conversation: Conversation,
        message: str
    ) -> TriggerResult:
        """Attempt to handle a workflow for the given message"""
        pass
    
    async def get_starter_ui(
        self,
        agent: Agent,
        user: User,
        conversation: Conversation
    ) -> Optional[List[UiComponent]]:
        """Provide UI components when a conversation starts"""
        return None

TriggerResult

When a workflow is handled, it returns a TriggerResult:

from dataclasses import dataclass
from typing import List, Optional, Union, AsyncGenerator, Callable, Awaitable

@dataclass
class TriggerResult:
    triggered: bool  # If True, skip LLM processing
    components: Optional[Union[List[UiComponent], AsyncGenerator[UiComponent, None]]] = None
    conversation_mutation: Optional[Callable[[Conversation], Awaitable[None]]] = None

Registering Handlers

Add handlers when creating your agent:

from vanna import Agent, DefaultWorkflowHandler

# Option 1: Use the default handler (recommended for most users)
agent = Agent(
    llm_service=llm,
    tool_registry=tool_registry,
    user_resolver=user_resolver
    # workflow_handler defaults to DefaultWorkflowHandler()
)

# Option 2: Use a custom handler
agent = Agent(
    llm_service=llm,
    tool_registry=tool_registry,
    user_resolver=user_resolver,
    workflow_handler=MyCustomHandler()
)

# Option 3: Disable workflow handling entirely
agent = Agent(
    llm_service=llm,
    tool_registry=tool_registry,
    user_resolver=user_resolver,
    workflow_handler=None
)

Default Workflow Handler

Vanna includes a DefaultWorkflowHandler that provides:

🔍 Setup Health Checking

• Automatically detects critical tools (run_sql)
• Warns about missing memory tools
• Suggests improvements for incomplete setups

🎨 Smart Starter UI

• Complete Setup: Welcome message with quick actions
• Functional Setup: SQL available, suggestions for memory/viz tools
• Incomplete Setup: Clear error messages and setup guidance

💬 Built-in Commands

• /help - Show help and available commands
• /status - Detailed setup status report

📊 Tool Analysis

The default handler checks for:

• Critical: run_sql, sql_query, execute_sql (required)
• Memory: search_saved_correct_tool_uses, save_question_tool_args (recommended)
• Visualization: visualize_data, create_chart, plot_data (optional)
• Utilities: calculator, etc.

Example Starter UI

# Complete setup - shows success message
StatusCardComponent(
    title="SQL Connection", 
    status="success",
    description="Database connection configured and ready"
)

# Missing SQL - shows error
StatusCardComponent(
    title="SQL Connection",
    status="error", 
    description="No SQL tool detected - this is required"
)

# Missing memory - shows warning  
StatusCardComponent(
    title="Memory System",
    status="warning",
    description="Memory tools not configured - I won't remember patterns"
)

Use Cases

1. Command Handling

Handle slash commands and quick actions:

class CommandHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        if message.startswith("/help"):
            return TriggerResult(
                triggered=True,
                components=[
                    RichTextComponent(
                        content="## Available Commands\n\n- `/help` - Show this help\n- `/reset` - Clear conversation\n- `/report` - Generate report",
                        markdown=True
                    )
                ]
            )
        
        if message.startswith("/reset"):
            async def clear_conversation(conv):
                conv.messages.clear()
            
            return TriggerResult(
                triggered=True,
                components=[
                    StatusCardComponent(
                        title="Conversation Reset",
                        status="success",
                        description="Your conversation has been cleared."
                    )
                ],
                conversation_mutation=clear_conversation
            )
        
        # Continue to LLM
        return TriggerResult(triggered=False)

2. Rate Limiting with Upgrade UI

Enforce quotas and show upgrade prompts:

from vanna.components import StatusCardComponent, ButtonComponent, ButtonGroupComponent

class RateLimitHandler(WorkflowHandler):
    def __init__(self, quota_service):
        self.quota_service = quota_service
    
    async def try_handle(self, agent, user, conversation, message):
        # Check user's quota
        remaining = await self.quota_service.get_remaining(user.id)
        
        if remaining <= 0:
            return TriggerResult(
                triggered=True,
                components=[
                    StatusCardComponent(
                        title="Monthly Limit Reached",
                        status="warning",
                        description=f"You've used all {await self.quota_service.get_quota(user.id)} messages this month",
                        icon="⚠️"
                    ),
                    RichTextComponent(
                        content="## Upgrade to Continue\n\n✨ **Pro Plan Benefits:**\n- Unlimited messages\n- Priority support\n- Advanced features\n- Custom integrations",
                        markdown=True
                    ),
                    ButtonGroupComponent(
                        buttons=[
                            {
                                "label": "Upgrade Now",
                                "action": f"https://billing.example.com/upgrade?user={user.id}",
                                "variant": "primary"
                            },
                            {
                                "label": "View Usage",
                                "action": "/usage",
                                "variant": "secondary"
                            }
                        ],
                        orientation="horizontal"
                    )
                ]
            )
        
        # Decrement quota and continue
        await self.quota_service.decrement(user.id)
        return TriggerResult(triggered=False)

3. Pattern-Based Routing

Route specific patterns to tools:

import re

class ReportHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        # Match report generation patterns
        report_patterns = [
            r"generate (w+) report",
            r"create (w+) analysis",
            r"show me (w+) data"
        ]
        
        for pattern in report_patterns:
            match = re.search(pattern, message.lower())
            if match:
                report_type = match.group(1)
                
                # Execute report tool directly
                tool = await agent.tool_registry.get_tool("generate_report")
                context = ToolContext(
                    user=user,
                    conversation_id=conversation.id,
                    request_id=str(uuid.uuid4())
                )
                
                result = await tool.execute(context, {"type": report_type})
                
                return TriggerResult(
                    triggered=True,
                    components=[result.ui_component] if result.ui_component else []
                )
        
        return TriggerResult(triggered=False)

4. State-Based Workflows

Handle onboarding and multi-step processes:

class OnboardingHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        # Check if user needs onboarding
        if user.metadata.get("needs_onboarding"):
            step = user.metadata.get("onboarding_step", 1)
            
            if step == 1:
                # Welcome step
                async def update_step(conv):
                    user.metadata["onboarding_step"] = 2
                
                return TriggerResult(
                    triggered=True,
                    components=[
                        RichTextComponent(
                            content="# Welcome to Vanna! 👋\n\nLet's get you set up in 3 quick steps.",
                            markdown=True
                        ),
                        ButtonComponent(
                            label="Get Started",
                            action="Continue onboarding",
                            variant="primary"
                        )
                    ],
                    conversation_mutation=update_step
                )
            
            elif step == 2:
                # Configuration step
                return TriggerResult(
                    triggered=True,
                    components=[
                        StatusCardComponent(
                            title="Step 2: Database Connection",
                            status="pending",
                            description="Let's connect to your database"
                        ),
                        # ... configuration UI
                    ]
                )
        
        return TriggerResult(triggered=False)

5. Permission-Based Access

Control access to features:

class PermissionHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        # Check for admin commands
        if message.startswith("/admin"):
            if "admin" not in user.group_memberships:
                return TriggerResult(
                    triggered=True,
                    components=[
                        StatusCardComponent(
                            title="Access Denied",
                            status="error",
                            description="You don't have permission to use admin commands.",
                            icon="🚫"
                        )
                    ]
                )
            
            # Handle admin commands
            if "/admin users" in message:
                # Show user management interface
                return TriggerResult(
                    triggered=True,
                    components=[
                        # Admin UI components
                    ]
                )
        
        return TriggerResult(triggered=False)

Starter UI

Provide welcome messages and quick actions:

class StarterUIHandler(WorkflowHandler):
    async def get_starter_ui(self, agent, user, conversation):
        # Role-based starter UI
        if "analyst" in user.group_memberships:
            # Get available report tools dynamically
            tools = await agent.tool_registry.get_schemas(user)
            report_tools = [t for t in tools if t.name.startswith("report_")]
            
            buttons = [
                {
                    "label": f"📊 {tool.name.replace('report_', '').title()} Report",
                    "action": f"/{tool.name}",
                    "variant": "secondary"
                }
                for tool in report_tools[:3]  # Show top 3
            ]
            
            return [
                RichTextComponent(
                    content=f"# Welcome back, {user.username}! 👋\n\nReady to analyze some data?",
                    markdown=True
                ),
                ButtonGroupComponent(
                    buttons=buttons,
                    orientation="vertical"
                )
            ]
        
        elif user.metadata.get("is_new_user"):
            return [
                RichTextComponent(
                    content="# Welcome to Vanna! 🎉\n\nI'm your AI data assistant. Try one of these to get started:",
                    markdown=True
                ),
                ButtonGroupComponent(
                    buttons=[
                        {"label": "📖 Show Tutorial", "action": "/tutorial", "variant": "primary"},
                        {"label": "💡 Example Query", "action": "/example", "variant": "secondary"},
                        {"label": "🔧 Setup Guide", "action": "/setup", "variant": "secondary"}
                    ],
                    orientation="vertical"
                )
            ]
        
        return None  # No starter UI for other users
    
    async def try_handle(self, agent, user, conversation, message):
        # Handle starter UI actions
        if message == "/tutorial":
            return TriggerResult(
                triggered=True,
                components=[
                    # Tutorial content
                ]
            )
        
        return TriggerResult(triggered=False)

Execution Flow

1. User sends message → Agent receives message
2. User resolution → Identify user and load context
3. Conversation loading → Load or create conversation
4. Workflow handler → try_handle() called before message added to conversation
5. If triggered=True → Stream components and exit (skip LLM)
6. If triggered=False → Add message to conversation and continue to LLM

# Agent execution flow (simplified)
async def send_message(self, user, message, conversation_id=None):
    # 1. Load conversation
    conversation = await self.conversation_store.get_conversation(conversation_id, user)
    
    # 2. Try workflow handler FIRST
    if self.workflow_handler:
        result = await self.workflow_handler.try_handle(self, user, conversation, message)
        
        if result.triggered:
            # Stream components and exit
            for component in result.components:
                yield component
            return  # Skip LLM
    
    # 3. Add message to conversation
    conversation.add_message(Message(role="user", content=message))
    
    # 4. Continue to LLM...

Combining with Other Features

Workflow handlers work alongside other extensibility points:

agent = Agent(
    llm_service=llm,
    tool_registry=tool_registry,
    user_resolver=user_resolver,
    workflow_handler=CommandHandler(),  # First: handle commands
    lifecycle_hooks=[QuotaHook()],      # Then: track usage
    llm_middlewares=[CachingMiddleware()],  # Finally: LLM processing
)

Customizing the Default Handler

You can customize the DefaultWorkflowHandler:

from vanna import DefaultWorkflowHandler

# Custom welcome message
handler = DefaultWorkflowHandler(
    welcome_message="# Welcome to AcmeCorp Data Assistant!\n\nI'm here to help with your analytics needs."
)

# Extend the default handler
class CustomDefaultHandler(DefaultWorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        # Handle custom commands first
        if message.startswith("/acme"):
            return TriggerResult(
                triggered=True,
                components=[
                    RichTextComponent(content="AcmeCorp specific feature...")
                ]
            )
        
        # Fall back to default behavior
        return await super().try_handle(agent, user, conversation, message)
    
    async def get_starter_ui(self, agent, user, conversation):
        components = await super().get_starter_ui(agent, user, conversation)
        
        # Add company-specific branding
        if components:
            components.insert(0, RichTextComponent(
                content="*Powered by AcmeCorp Analytics Platform*",
                markdown=True
            ))
        
        return components

Observability

The agent automatically creates observability spans:

• agent.workflow_handler.try_handle - Time spent in try_handle()
• agent.workflow_handler.starter_ui - Time spent generating starter UI

# Access observability in your handler
class ObservableHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        if agent.observability_provider:
            span = await agent.observability_provider.create_span(
                "custom.workflow.processing",
                attributes={"message_type": self.classify_message(message)}
            )
            
            # ... processing logic ...
            
            await agent.observability_provider.end_span(span)
        
        return TriggerResult(triggered=False)

Best Practices

1. Keep handlers focused - One responsibility per handler
2. Be efficient - Handlers run on every message
3. Handle errors gracefully - Don’t crash the agent
4. Use proper UI components - Leverage the rich component system
5. Consider conversation state - Think about multi-turn workflows
6. Test thoroughly - Unit test all workflow paths
7. Document patterns - Make message patterns clear
8. Use observability - Monitor handler performance

Advanced Examples

Multi-Step Workflow with State

class SurveyHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        survey_state = user.metadata.get("survey_state")
        
        if message == "/survey" or survey_state:
            if not survey_state:
                # Start survey
                async def start_survey(conv):
                    user.metadata["survey_state"] = {"step": 1, "answers": {}}
                
                return TriggerResult(
                    triggered=True,
                    components=[
                        StatusCardComponent(
                            title="Customer Feedback Survey",
                            status="pending",
                            description="Help us improve - 3 quick questions"
                        ),
                        RichTextComponent(content="**Question 1:** How satisfied are you with our service?"),
                        ButtonGroupComponent(
                            buttons=[
                                {"label": "😍 Very Satisfied", "action": "survey:q1:5"},
                                {"label": "😊 Satisfied", "action": "survey:q1:4"},
                                {"label": "😐 Neutral", "action": "survey:q1:3"},
                                {"label": "😞 Unsatisfied", "action": "survey:q1:2"},
                                {"label": "😡 Very Unsatisfied", "action": "survey:q1:1"}
                            ],
                            orientation="vertical"
                        )
                    ],
                    conversation_mutation=start_survey
                )
            
            elif message.startswith("survey:"):
                # Handle survey responses
                parts = message.split(":")
                question, answer = parts[1], parts[2]
                
                survey_state["answers"][question] = answer
                next_step = survey_state["step"] + 1
                
                if next_step <= 3:
                    # Continue survey
                    survey_state["step"] = next_step
                    return TriggerResult(
                        triggered=True,
                        components=[self._get_survey_question(next_step)]
                    )
                else:
                    # Complete survey
                    del user.metadata["survey_state"]
                    await self._save_survey_results(user.id, survey_state["answers"])
                    
                    return TriggerResult(
                        triggered=True,
                        components=[
                            StatusCardComponent(
                                title="Survey Complete",
                                status="success",
                                description="Thank you for your feedback! 🙏"
                            )
                        ]
                    )
        
        return TriggerResult(triggered=False)

Dynamic Tool Menu

class ToolMenuHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        if message == "/tools" or message == "/menu":
            # Get available tools for user
            tools = await agent.tool_registry.get_schemas(user)
            
            # Group tools by category
            categories = {}
            for tool in tools:
                category = tool.metadata.get("category", "General")
                if category not in categories:
                    categories[category] = []
                categories[category].append(tool)
            
            # Build menu components
            components = [
                RichTextComponent(
                    content="# Available Tools 🛠️\n\nChoose a tool to get started:",
                    markdown=True
                )
            ]
            
            for category, category_tools in categories.items():
                buttons = [
                    {
                        "label": f"{tool.metadata.get('icon', '⚙️')} {tool.name.replace('_', ' ').title()}",
                        "action": f"/tool {tool.name}",
                        "variant": "secondary"
                    }
                    for tool in category_tools[:5]  # Limit per category
                ]
                
                components.extend([
                    RichTextComponent(content=f"## {category}"),
                    ButtonGroupComponent(buttons=buttons, orientation="vertical")
                ])
            
            return TriggerResult(triggered=True, components=components)
        
        return TriggerResult(triggered=False)

Error Handling

Handle errors gracefully in your workflow handlers:

class RobustHandler(WorkflowHandler):
    async def try_handle(self, agent, user, conversation, message):
        try:
            # Your workflow logic
            if message.startswith("/risky-operation"):
                result = await self.perform_risky_operation(user, message)
                return TriggerResult(triggered=True, components=[result])
            
        except PermissionError:
            return TriggerResult(
                triggered=True,
                components=[
                    StatusCardComponent(
                        title="Permission Denied",
                        status="error",
                        description="You don't have permission for this operation."
                    )
                ]
            )
        except ValidationError as e:
            return TriggerResult(
                triggered=True,
                components=[
                    StatusCardComponent(
                        title="Invalid Input",
                        status="error",
                        description=f"Please check your input: {str(e)}"
                    )
                ]
            )
        except Exception as e:
            # Log unexpected errors
            logger.error(f"Workflow handler error: {e}", exc_info=True)
            
            return TriggerResult(
                triggered=True,
                components=[
                    StatusCardComponent(
                        title="Something went wrong",
                        status="error",
                        description="Please try again or contact support if the problem persists."
                    )
                ]
            )
        
        return TriggerResult(triggered=False)

See Also

• Lifecycle Hooks - Hook into agent execution lifecycle
• LLM Middlewares - Intercept LLM requests/responses
• UI Components - Available UI components for handlers
• Context Enrichers - Add data to tool execution context
• Observability - Monitor handler performance