// "Production-ready code templates for Kailash SDK including basic workflows, cyclic workflows, custom nodes, MCP servers, and all three test tiers (unit, integration, end-to-end). Use when asking about 'template', 'example code', 'starter code', 'boilerplate', 'scaffold', 'workflow template', 'custom node template', 'MCP server template', 'test template', 'unit test template', 'integration test template', or 'E2E test template'."
| name | code-templates |
| description | Production-ready code templates for Kailash SDK including basic workflows, cyclic workflows, custom nodes, MCP servers, and all three test tiers (unit, integration, end-to-end). Use when asking about 'template', 'example code', 'starter code', 'boilerplate', 'scaffold', 'workflow template', 'custom node template', 'MCP server template', 'test template', 'unit test template', 'integration test template', or 'E2E test template'. |
Production-ready code templates and boilerplate for common Kailash development tasks.
Complete templates for:
Workflow Development:
template-workflow-basic → Standard workflowstemplate-cyclic-workflow → Iterative workflowsCustom Development:
template-custom-node → New node typestemplate-mcp-server → MCP integrationTesting:
template-test-unit → Fast unit teststemplate-test-integration → Real infrastructure teststemplate-test-e2e → Complete system testsfrom 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"])
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
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
Use this skill when you need:
| Task | Template | Why |
|---|---|---|
| New workflow | template-workflow-basic | Standard structure |
| Iterative logic | template-cyclic-workflow | Cycle pattern |
| Custom node | template-custom-node | Node structure |
| MCP integration | template-mcp-server | MCP setup |
| Fast tests | template-test-unit | Unit testing |
| Real infra tests | template-test-integration | Integration |
| Full system | template-test-e2e | End-to-end |
For template help, invoke:
pattern-expert - Pattern selectiontdd-implementer - Test-first developmentsdk-navigator - Find examples