Kailash Code Templates

Kailash Code Templates

Production-ready code templates and boilerplate for common Kailash development tasks.

Overview

Complete templates for:

• Basic workflows
• Cyclic workflows
• Custom node development
• MCP server creation
• Unit tests (Tier 1)
• Integration tests (Tier 2)
• End-to-end tests (Tier 3)

Reference Documentation

Workflow Templates

Basic Workflow

• template-workflow-basic - Standard workflow template
  • WorkflowBuilder setup
  • Node addition pattern
  • Connection pattern
  • Runtime execution
  • Result access
  • Error handling

Cyclic Workflow

• template-cyclic-workflow - Cyclic workflow template
  • CycleNode setup
  • Convergence checking
  • State persistence
  • Iteration limits
  • Result aggregation

Custom Development Templates

Custom Node

• template-custom-node - Custom node template
  • Node class structure
  • Parameter definition
  • Execute method implementation
  • Input validation
  • Output formatting
  • Error handling
  • Registration pattern

MCP Server

• template-mcp-server - MCP server template
  • MCPServer initialization
  • Tool registration
  • Resource definition
  • Transport configuration
  • Authentication setup
  • Progress reporting

Test Templates

Unit Tests (Tier 1)

• template-test-unit - Unit test template
  • Test class structure
  • Fixture setup
  • Mock setup (allowed in Tier 1)
  • Assertion patterns
  • Teardown cleanup
  • Fast execution pattern

Integration Tests (Tier 2)

• template-test-integration - Integration test template
  • Real database setup
  • Workflow execution testing
  • NO MOCKING policy
  • Real infrastructure fixtures
  • Resource cleanup
  • Result validation

End-to-End Tests (Tier 3)

• template-test-e2e - E2E test template
  • Full system setup
  • API testing pattern
  • Real HTTP requests
  • Complete user flows
  • Multi-component testing
  • Production-like environment

Template Usage

Quick Start Process

1. Select Template: Choose relevant template for your task
2. Copy Code: Copy template as starting point
3. Customize: Adapt to your specific needs
4. Test: Verify with real data
5. Refine: Iterate based on results

Template Categories

Workflow Development:

• template-workflow-basic → Standard workflows
• template-cyclic-workflow → Iterative workflows

Custom Development:

• template-custom-node → New node types
• template-mcp-server → MCP integration

Testing:

• template-test-unit → Fast unit tests
• template-test-integration → Real infrastructure tests
• template-test-e2e → Complete system tests

Template Examples

Basic Workflow Template

from kailash.workflow.builder import WorkflowBuilder
from kailash.runtime.local import LocalRuntime

def create_workflow():
    """Create a basic workflow."""
    workflow = WorkflowBuilder()

    # Add nodes
    workflow.add_node("PythonCodeNode", "node1", {
        "code": "result = input_data * 2"
    })

    workflow.add_node("PythonCodeNode", "node2", {
        "code": "result = input_data + 10"
    })

    # Add connections
    workflow.add_connection("node1", "result", "node2", "input_data")

    return workflow

# Execute
runtime = LocalRuntime()
results, run_id = runtime.execute(create_workflow().build())
print(results["node2"]["result"])

Custom Node Template

from kailash.nodes.base import BaseNode

class MyCustomNode(BaseNode):
    """Custom node implementation."""

    def __init__(self, node_id: str, **params):
        super().__init__(node_id)
        self.my_param = params.get("my_param")

    def execute(self, inputs: dict) -> dict:
        """Execute node logic."""
        # Validate inputs
        if "input_data" not in inputs:
            raise ValueError("Missing input_data")

        # Process
        result = self.process(inputs["input_data"])

        # Return output
        return {"result": result}

    def process(self, data):
        """Custom processing logic."""
        return data * 2

Integration Test Template

import pytest
from kailash.workflow.builder import WorkflowBuilder
from kailash.runtime.local import LocalRuntime
from dataflow import DataFlow

@pytest.fixture
def db():
    """Real database for testing."""
    db = DataFlow("postgresql://test:test@localhost:5433/​test_db")
    db.create_tables()
    yield db
    db.drop_tables()

@pytest.fixture
def runtime():
    """Real runtime."""
    return LocalRuntime()

def test_workflow_execution(runtime):
    """Tier 2: Integration test with real execution."""
    # Arrange
    workflow = WorkflowBuilder()
    workflow.add_node("PythonCodeNode", "calc", {
        "code": "result = 2 + 2"
    })

    # Act
    results, run_id = runtime.execute(workflow.build())

    # Assert
    assert results["calc"]["result"] == 4
    assert run_id is not None

Best Practices

Using Templates

• ✅ Start with template, then customize
• ✅ Keep template structure
• ✅ Follow naming conventions
• ✅ Add comments for customization points
• ✅ Test template modifications
• ❌ NEVER skip error handling
• ❌ NEVER remove validation
• ❌ NEVER ignore resource cleanup

Template Customization

• Keep core structure intact
• Add custom logic in designated areas
• Maintain error handling patterns
• Preserve type hints
• Update docstrings

Testing Templates

• Test with real data
• Verify all code paths
• Check error handling
• Validate cleanup
• Performance test if needed

When to Use This Skill

Use this skill when you need:

• Starting point for new code
• Boilerplate for common tasks
• Reference implementation
• Best practice examples
• Quick scaffolding
• Test structure guidance

Template Selection Guide

Related Skills

• 01-core-sdk - Core patterns
• 06-cheatsheets - Pattern examples
• 07-development-guides - Development guides
• 12-testing-strategies - Testing strategies
• 17-gold-standards - Best practices

Support

For template help, invoke:

• pattern-expert - Pattern selection
• tdd-implementer - Test-first development
• sdk-navigator - Find examples