Backend Tool System API

Technical documentation for the backend tool creation and prompt management system based on tools/model_tools.py and tools/tool_prompts.py.

Overview

The backend tool system provides a sophisticated framework for creating, configuring, and managing AI model tools. It integrates with the VAC Service Architecture and Tool Context to provide dynamic tool capabilities for AI assistants.

Key Components:

  • Dynamic tool creation for Google Gemini models
  • Langfuse-based prompt management and caching
  • UI/non-UI tool filtering and configuration
  • Markdown formatting for tool contexts
  • Integration with permission and access control systems

Model Tools (tools/model_tools.py)

Handles creation of Google GenAI Tool objects for AI model integration.

Core Function

create_model_tools(tools: List[str]) -> List[types.Tool]

Creates Google GenAI Tool objects from tool name strings.

Supported Tools:

  • google_search_retrieval → Creates Google Search tool
  • code_execution → Creates Code Execution tool
  • url_processing → Creates URL Context tool

Implementation:

def create_model_tools(tools):
    """
    Create Google GenAI Tool objects from tool names.
    
    Args:
        tools: List of tool names or None
        
    Returns:
        List of types.Tool objects
    """
    if not tools:
        return []
    
    model_tools = []
    
    for tool in tools:
        if tool == "google_search_retrieval":
            model_tools.append(types.Tool(google_search={}))
        elif tool == "code_execution": 
            model_tools.append(types.Tool(code_execution={}))
        elif tool == "url_processing":
            model_tools.append(types.Tool(url_context={}))
        # Unknown tools are silently ignored
    
    log.info(f"Created model_tools={model_tools}")
    return model_tools

Usage Example:

# Create tools for assistant
requested_tools = ["google_search_retrieval", "code_execution"]
model_tools = create_model_tools(requested_tools)

# Pass to VAC service
result = await vac_stream(
    question=user_question,
    tools_to_use=model_tools,
    # ... other parameters
)

Error Handling:

  • Unknown tools are silently ignored (no exceptions thrown)
  • Empty or None input returns empty list
  • Duplicates are preserved (no deduplication)
  • Logs created tools for debugging

Integration Points:

  • VAC Service: Tools passed directly to AI model calls
  • Permission System: Tool names validated against user permissions
  • Tool Context: Tool configurations merged with model tools

Tool Prompts (tools/tool_prompts.py)

Manages dynamic loading and formatting of tool-specific prompts from Langfuse.

Core Functions

add_tool_prompts(tools, prefix="aitana3", ui_only=None) -> str

Loads and formats tool-specific prompts with caching and filtering.

Parameters:

  • tools: List of tool names (strings) or tool objects (dicts)
  • prefix: Langfuse prompt prefix (default: “aitana3”)
  • ui_only: Filter for UI tools (True=UI only, False=non-UI only, None=all)

Process:

  1. Tool Name Extraction: Handle both string and dict tool formats
  2. UI Filtering: Apply UI-only filter if specified
  3. Prompt Loading: Load from Langfuse with caching (300s TTL)
  4. Error Handling: Continue on individual prompt failures
  5. Content Assembly: Join successful prompts with double newlines

Implementation:

def add_tool_prompts(tools, prefix="aitana3", ui_only=None):
    """
    Load tool prompts from Langfuse and format them.
    
    Args:
        tools: List of tool names or tool objects
        prefix: Langfuse prompt prefix  
        ui_only: Filter for UI tools (True/False/None)
        
    Returns:
        Formatted prompt string
    """
    if not tools:
        return ""
    
    prompts = []
    
    for tool in tools:
        # Extract tool name
        if isinstance(tool, str):
            tool_name = tool
        elif isinstance(tool, dict):
            tool_name = tool.get("name")
        else:
            raise ValueError(f"What is tool? {tool}")
        
        # Apply UI filter
        if ui_only is True and tool_name not in UI_PROMPTS:
            continue
        elif ui_only is False and tool_name in UI_PROMPTS:
            continue
        
        # Load prompt from Langfuse
        try:
            prompt = langfuse.get_prompt(f"{prefix}-{tool_name}", cache_ttl_seconds=300)
            compiled_prompt = prompt.compile()
            prompts.append(compiled_prompt)
        except Exception as e:
            log.warning(f"Failed to load prompt for tool {tool_name}: {e}")
            continue
    
    return "\n\n".join(prompts)

UI Tools Configuration:

UI_PROMPTS = [
    "tooltips",
    "alerts", 
    "preview",
    "plots",
    "artifact",
    "dynamic-ui",
    "assistantresponse"
]

Usage Examples:

Basic Usage:

# Load all tool prompts
tools = ["google_search_retrieval", "code_execution", "tooltips"]
prompts = add_tool_prompts(tools)

UI-Only Filtering:

# Load only UI tool prompts
ui_tools = ["tooltips", "alerts", "search", "plots"]
ui_prompts = add_tool_prompts(ui_tools, ui_only=True)
# Returns prompts for: tooltips, alerts, plots (excludes search)

Non-UI Filtering:

# Load only non-UI tool prompts  
mixed_tools = ["google_search_retrieval", "tooltips", "code_execution"]
backend_prompts = add_tool_prompts(mixed_tools, ui_only=False)
# Returns prompts for: google_search_retrieval, code_execution (excludes tooltips)

Custom Prefix:

# Use custom Langfuse prompt prefix
prompts = add_tool_prompts(tools, prefix="custom-assistant")
# Loads prompts: custom-assistant-tool1, custom-assistant-tool2, etc.

Prompt Management Features

Langfuse Integration

Caching Strategy:

  • 300-second TTL for prompt caching
  • Reduces API calls and improves performance
  • Cache key based on prompt name

Prompt Naming Convention:

{prefix}-{tool_name}

Examples:

  • aitana3-google_search_retrieval
  • aitana3-code_execution
  • custom-assistant-tooltips

Error Resilience

Individual Failure Handling:

# If one prompt fails, others continue loading
tools = ["valid_tool", "invalid_tool", "another_valid_tool"]
prompts = add_tool_prompts(tools)
# Returns prompts for valid tools only, logs warning for invalid_tool

Graceful Degradation:

  • Failed prompts are logged but don’t stop processing
  • Empty result if all prompts fail
  • Partial success returns available prompts

Format Utilities

format_as_markdown(context: Dict) -> str

Converts context dictionaries to structured markdown format for AI consumption.

Features:

  • Recursive processing of nested dictionaries
  • List handling with item enumeration
  • Section headers with markdown formatting
  • XML-style tag wrapping for structured data

Example Usage:

context = {
    "user_info": {
        "name": "John Doe",
        "email": "john@example.com"
    },
    "search_results": [
        {"title": "Result 1", "url": "https://example.com/1"},
        {"title": "Result 2", "url": "https://example.com/2"}
    ],
    "settings": {
        "theme": "dark",
        "notifications": True
    }
}

formatted = format_as_markdown(context)

Output:

### user_info

<name>
John Doe
</name>

<email>
john@example.com
</email>

---

### search_results

<item>

<title>
Result 1
</title>

<url>
https://example.com/1
</url>

</item>

<item>

<title>
Result 2
</title>

<url>
https://example.com/2
</url>

</item>

---

### settings

<theme>
dark
</theme>

<notifications>
True
</notifications>

Processing Rules:

  1. Top-level keys → ### {key} headers
  2. Dictionary values → Nested <key>value</key> tags
  3. List values → <item> wrappers with nested structure
  4. Sections → Separated by --- dividers
  5. Primitives → String conversion with proper formatting

Integration with Core Systems

VAC Service Integration

Tool system integrates with VAC Service Architecture:

# Tool preparation for VAC service
def prepare_vac_tools(emissary_config):
    """Prepare tools for VAC service processing."""
    
    # Extract tool names from config
    tool_names = emissary_config.get("tools", [])
    
    # Create model tools for AI
    model_tools = create_model_tools(tool_names)
    
    # Load tool prompts for context
    tool_prompts = add_tool_prompts(tool_names, ui_only=False)
    
    # Combine for VAC service
    return {
        "model_tools": model_tools,
        "tool_context": tool_prompts,
        "tool_configs": emissary_config.get("toolConfigs", {})
    }

# Usage in VAC service call
vac_tools = prepare_vac_tools(emissary_config)
result = await vac_stream(
    question=user_question,
    tools_to_use=vac_tools["model_tools"],
    instructions=f"{base_instructions}\n\n{vac_tools['tool_context']}",
    emissaryConfig=emissary_config
)

Email Integration

Tool system supports Email Integration processing:

# Email-specific tool processing
async def process_email_tools(assistant_config, current_user):
    """Process tools for email context."""
    
    # Get permitted tools for user
    requested_tools = assistant_config.get('tools', [])
    tool_configs = assistant_config.get('toolConfigs', {})
    
    user_permissions, allowed_configs = permitted_tools(
        current_user, requested_tools, tool_configs
    )
    
    # Create model tools for permitted tools only
    email_model_tools = create_model_tools(user_permissions)
    
    # Load non-UI prompts (UI tools not relevant for email)
    email_tool_prompts = add_tool_prompts(user_permissions, ui_only=False)
    
    return {
        "tools": user_permissions,
        "model_tools": email_model_tools,
        "tool_prompts": email_tool_prompts,
        "configs": allowed_configs
    }

Permission System Integration

Tool system respects Tool Context permissions:

# Permission-aware tool loading
def load_user_tools(user, assistant_tools):
    """Load tools based on user permissions."""
    
    # Check permissions (from tool_permissions.py)
    permitted_tool_names, permitted_configs = permitted_tools(
        user, assistant_tools, {}
    )
    
    # Filter tools by permissions
    filtered_tools = [
        tool for tool in assistant_tools 
        if tool in permitted_tool_names
    ]
    
    # Create model tools for permitted tools
    user_model_tools = create_model_tools(filtered_tools)
    
    # Load prompts for permitted tools
    user_tool_prompts = add_tool_prompts(filtered_tools)
    
    return {
        "permitted_tools": filtered_tools,
        "model_tools": user_model_tools,
        "tool_prompts": user_tool_prompts
    }

Testing and Validation

Model Tools Testing

Tool Creation Validation:

def test_create_model_tools_google_search():
    """Test Google Search tool creation."""
    tools = ["google_search_retrieval"]
    result = create_model_tools(tools)
    
    assert len(result) == 1
    assert isinstance(result[0], types.Tool)
    assert hasattr(result[0], 'google_search')

def test_create_model_tools_unknown_tool():
    """Test unknown tool handling."""
    tools = ["unknown_tool", "google_search_retrieval"]
    result = create_model_tools(tools)
    
    # Should only create known tools
    assert len(result) == 1
    assert hasattr(result[0], 'google_search')

Prompt System Testing

Prompt Loading with Mocking:

@patch('tools.tool_prompts.langfuse')
def test_add_tool_prompts_string_tools(mock_langfuse):
    """Test prompt loading for string tools."""
    mock_prompt = MagicMock()
    mock_prompt.compile.return_value = "Tool prompt content"
    mock_langfuse.get_prompt.return_value = mock_prompt
    
    tools = ["tool1", "tool2"]
    result = add_tool_prompts(tools)
    
    assert "Tool prompt content" in result
    assert mock_langfuse.get_prompt.call_count == 2

Error Handling Testing:

@patch('tools.tool_prompts.langfuse')
def test_add_tool_prompts_error_handling(mock_langfuse):
    """Test error handling for failed prompt loads."""
    mock_langfuse.get_prompt.side_effect = Exception("Prompt not found")
    
    tools = ["tool1", "tool2"]
    result = add_tool_prompts(tools)
    
    # Should return empty string when all prompts fail
    assert result == ""

UI Filtering Testing:

def test_add_tool_prompts_ui_filter():
    """Test UI-only filtering."""
    tools = ["tooltips", "google_search_retrieval", "alerts"]
    
    # UI-only should include tooltips and alerts
    ui_result = add_tool_prompts(tools, ui_only=True)
    
    # Non-UI should include only google_search_retrieval
    non_ui_result = add_tool_prompts(tools, ui_only=False)

Performance Considerations

Prompt Caching

Langfuse Cache Configuration:

# 300-second TTL reduces API calls
prompt = langfuse.get_prompt(prompt_name, cache_ttl_seconds=300)

Cache Effectiveness:

  • Reduces API latency for repeated tool combinations
  • Improves response times for common assistants
  • Decreases Langfuse API usage costs

Tool Processing Optimization

Batch Processing:

# Process multiple tools in single function call
tools = ["search", "code", "analysis", "ui"]
all_prompts = add_tool_prompts(tools)  # Single batch operation

# Avoid individual calls
# search_prompt = add_tool_prompts(["search"])     # Inefficient
# code_prompt = add_tool_prompts(["code"])         # Inefficient

Memory Efficiency:

  • String concatenation with join() for large prompt sets
  • Lazy evaluation for unused tool combinations
  • Efficient filtering with early returns

Configuration and Deployment

Environment Variables

Langfuse Configuration:

# Required for prompt loading
LANGFUSE_SECRET_KEY=your_secret_key
LANGFUSE_PUBLIC_KEY=your_public_key
LANGFUSE_HOST=https://your-langfuse-instance.com

Tool-Specific Settings:

# Google Search API
GOOGLE_SEARCH_API_KEY=your_api_key
GOOGLE_SEARCH_ENGINE_ID=your_engine_id

# Code execution settings
CODE_EXECUTION_TIMEOUT=30
CODE_EXECUTION_MEMORY_LIMIT=512MB

Prompt Management Best Practices

Naming Conventions:

  • Use consistent prefix patterns: {assistant_type}-{tool_name}
  • Version prompts with suffixes: aitana3-search-v2
  • Environment-specific prefixes: dev-aitana3-search

Content Guidelines:

  • Keep prompts focused and specific to tool functionality
  • Include example usage and expected outputs
  • Maintain backward compatibility when updating prompts
  • Test prompt changes with representative tool combinations

Monitoring:

  • Track prompt loading success rates
  • Monitor cache hit ratios
  • Log prompt compilation times
  • Alert on prompt loading failures

Error Handling and Debugging

Common Issues

Prompt Loading Failures:

# Debug prompt loading
try:
    prompt = langfuse.get_prompt(f"{prefix}-{tool_name}")
    log.info(f"Successfully loaded prompt for {tool_name}")
except Exception as e:
    log.error(f"Failed to load prompt for {tool_name}: {e}")
    # Check: prompt exists in Langfuse, API keys correct, network connectivity

Tool Creation Issues:

# Debug tool creation
model_tools = create_model_tools(tool_names)
log.info(f"Created {len(model_tools)} tools from {len(tool_names)} requested")

# Check for silently ignored tools
created_tool_types = [type(tool).__name__ for tool in model_tools]
log.debug(f"Created tool types: {created_tool_types}")

Monitoring and Observability

Key Metrics:

  • Tool creation success rate
  • Prompt loading latency
  • Cache hit ratios
  • Tool usage patterns by assistant
  • Error rates by tool type

Logging Strategy:

# Structured logging for tool operations
log.info("tool_operation", extra={
    "operation": "create_model_tools",
    "requested_tools": tool_names,
    "created_count": len(model_tools),
    "processing_time": processing_time,
    "user_id": user.get("uid"),
    "assistant_id": assistant_id
})