Authentication & Permissions

Vanna 2.0 provides a flexible authentication and permissions system that ensures every tool execution and UI feature is protected by user-based access controls.

Core Concepts

User Model

Every request in Vanna is associated with a User object that contains:

• id: Unique identifier for the user
• username: Display name
• email: User’s email address
• permissions: List of permission strings
• group_memberships: List of groups the user belongs to (used for tool/feature access)
• metadata: Additional custom data

UserResolver

The UserResolver is responsible for extracting user identity from web requests (cookies, JWT tokens, headers, etc.) and creating authenticated User objects.

from vanna.core.user import UserResolver, RequestContext, User

class MyUserResolver(UserResolver):
    async def resolve_user(self, request_context: RequestContext) -> User:
        # Extract authentication from request
        token = request_context.get_header('Authorization')
        
        # Validate and decode token
        user_data = validate_jwt(token)
        
        # Return User object
        return User(
            id=user_data['user_id'],
            username=user_data['username'],
            email=user_data['email'],
            group_memberships=['user', 'analytics_team'],
            permissions=['read', 'write']
        )

Permission System

Tool Access Control

Tool access control is enforced when tools are registered with the agent’s tool registry using register_local_tool(). You specify which groups can access each tool at registration time:

from vanna.core.registry import ToolRegistry
from vanna.tools import RunSqlTool

# Create tool registry
tools = ToolRegistry()

# Register tools with specific access groups
tools.register_local_tool(
    RunSqlTool(sql_runner=sql_runner),
    access_groups=['admin', 'data_team']  # Only these groups can use this tool
)

tools.register_local_tool(
    AnalyticsTool(),
    access_groups=['analyst', 'admin']
)

tools.register_local_tool(
    PublicReportTool(),
    access_groups=[]  # Empty list = accessible to all users
)

# Create agent with configured tool registry
agent = Agent(
    llm_service=llm,
    tool_registry=tools,
    user_resolver=user_resolver
)

• Empty list ([]): Tool is accessible to all users
• Non-empty list: User must be in at least one of the specified groups

The tool registry validates access permissions both at registration time and at execution time, ensuring users can only invoke tools they have permission to use.

UI Feature Access Control

Similarly, UI features can be restricted by group membership via UiFeatures configuration:

from vanna.core.agent import AgentConfig, UiFeatures

config = AgentConfig(
    ui_features=UiFeatures(
        feature_group_access={
            'tool_names': ['admin', 'developer'],
            'tool_arguments': ['admin'],
            'tool_error': ['admin'],
        }
    )
)

Authentication Patterns

Pattern 1: JWT-Based Authentication

import jwt
from vanna.core.user import UserResolver, User, RequestContext

class JwtUserResolver(UserResolver):
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    async def resolve_user(self, request_context: RequestContext) -> User:
        auth_header = request_context.get_header('Authorization')
        
        if not auth_header or not auth_header.startswith('Bearer '):
            return User(id="anonymous", username="guest")
        
        token = auth_header.split(' ')[1]
        
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            
            return User(
                id=payload['sub'],
                username=payload.get('username', ''),
                email=payload.get('email', ''),
                group_memberships=payload.get('groups', []),
                permissions=payload.get('permissions', [])
            )
        except jwt.InvalidTokenError:
            return User(id="anonymous", username="guest")

Pattern 2: Session Cookie Authentication

class SessionUserResolver(UserResolver):
    def __init__(self, session_store):
        self.session_store = session_store
    
    async def resolve_user(self, request_context: RequestContext) -> User:
        session_id = request_context.get_cookie('session_id')
        
        if not session_id:
            return User(id="anonymous", username="guest")
        
        user_data = await self.session_store.get_user(session_id)
        
        if not user_data:
            return User(id="anonymous", username="guest")
        
        return User(
            id=user_data['id'],
            username=user_data['username'],
            email=user_data['email'],
            group_memberships=user_data.get('groups', []),
        )

Pattern 3: Okta/OAuth Integration

from okta_jwt_verifier import JWTVerifier

class OktaUserResolver(UserResolver):
    def __init__(self, issuer: str, client_id: str):
        self.verifier = JWTVerifier(
            issuer=issuer,
            client_id=client_id,
            audience='api://default'
        )
    
    async def resolve_user(self, request_context: RequestContext) -> User:
        auth_header = request_context.get_header('Authorization')
        
        if not auth_header or not auth_header.startswith('Bearer '):
            return User(id="anonymous", username="guest")
        
        token = auth_header.split(' ')[1]
        
        try:
            # Verify token with Okta
            claims = await self.verifier.verify_access_token(token)
            
            # Extract user information from Okta claims
            return User(
                id=claims['uid'],
                username=claims.get('preferred_username', claims['sub']),
                email=claims.get('email'),
                group_memberships=claims.get('groups', []),
                metadata={
                    'okta_org': claims['iss'],
                    'client_id': claims['cid']
                }
            )
        except Exception as e:
            # Token validation failed
            return User(id="anonymous", username="guest")

Deployment Integration

When deploying with FastAPI or Flask, configure the agent with your UserResolver:

from vanna import Agent
from vanna.servers.fastapi import VannaFastAPIServer

# Create agent with authentication
agent = Agent(
    llm_service=llm,
    sql_runner=sql_runner,
    user_resolver=JwtUserResolver(secret_key='your-secret')
)

# Deploy with FastAPI
server = VannaFastAPIServer(agent)
app = server.create_app()

Best Practices

1. Always validate tokens/credentials - Don’t trust client-provided data
2. Use HTTPS in production - Protect tokens and cookies in transit
3. Implement proper session management - Handle timeouts and revocation
4. Log authentication attempts - Track failed logins for security monitoring
5. Use group-based access control - Easier to manage than individual permissions
6. Provide anonymous fallback - Return anonymous user for public endpoints

See Also

• Deployment with FastAPI
• Deployment with Flask
• Custom Tools
• UI Feature Flags
• Audit Logging