
Kotlin Flow
- 17 installs
- 7 repo stars
- Updated January 5, 2026
- pluginagentmarketplace/custom-plugin-kotlin
kotlin-flow is a Claude Code skill for ai & agent building.
About
kotlin-flow is a Claude Code skill for ai & agent building. It helps solo builders move faster with AI-assisted development.
- kotlin-flow
- AI & Agent Building
- AI-coding skill
Kotlin Flow by the numbers
- 17 all-time installs (skills.sh)
- Ranked #10,886 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 4, 2026 (Skillselion catalog sync)
npx skills add https://github.com/pluginagentmarketplace/custom-plugin-kotlin --skill kotlin-flowAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 17 |
|---|---|
| repo stars | ★ 7 |
| Last updated | January 5, 2026 |
| Repository | pluginagentmarketplace/custom-plugin-kotlin ↗ |
How do I helps with ai & agent building tasks.?
Helps with ai & agent building tasks.
Who is it for?
Best when you're working on ai & agent building and need structured help with kotlin flow.
Skip if: Teams with no ai & agent building needs, or anyone wanting a generic chat assistant without this specific workflow.
When should I use this skill?
When you need to helps with ai & agent building tasks., or when kotlin-flow is a claude code skill for ai & agent building.
What you get
Structured output aligned to kotlin-flow: kotlin-flow, AI & Agent Building.
Files
Kotlin Flow Skill
Reactive programming with Kotlin Flow.
Topics Covered
Cold vs Hot Flows
// Cold Flow - starts fresh for each collector
fun loadData(): Flow<Data> = flow {
emit(fetchData())
}
// Hot Flow - shared state
private val _state = MutableStateFlow(State())
val state: StateFlow<State> = _state.asStateFlow()Flow Operators
fun searchUsers(query: Flow<String>): Flow<List<User>> =
query
.debounce(300)
.filter { it.length >= 2 }
.distinctUntilChanged()
.flatMapLatest { term -> userRepository.search(term) }
.catch { emit(emptyList()) }Combining Flows
val dashboard: Flow<Dashboard> = combine(
userFlow,
ordersFlow,
notificationsFlow
) { user, orders, notifications ->
Dashboard(user, orders.size, notifications.count())
}Testing with Turbine
@Test
fun `flow emits values`() = runTest {
viewModel.state.test {
assertThat(awaitItem().isLoading).isFalse()
viewModel.load()
assertThat(awaitItem().isLoading).isTrue()
advanceUntilIdle()
assertThat(awaitItem().data).isNotNull()
}
}Troubleshooting
| Issue | Resolution |
|---|---|
| Flow never emits | Add terminal operator (collect, first) |
| Stale data in UI | Use stateIn or shareIn properly |
Usage
Skill("kotlin-flow")# kotlin-flow Configuration
# Category: general
# Generated: 2025-12-30
skill:
name: kotlin-flow
version: "1.0.0"
category: general
settings:
# Default settings for kotlin-flow
enabled: true
log_level: info
# Category-specific defaults
validation:
strict_mode: false
auto_fix: false
output:
format: markdown
include_examples: true
# Environment-specific overrides
environments:
development:
log_level: debug
validation:
strict_mode: false
production:
log_level: warn
validation:
strict_mode: true
# Integration settings
integrations:
# Enable/disable integrations
git: true
linter: true
formatter: true
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "kotlin-flow Configuration Schema",
"type": "object",
"properties": {
"skill": {
"type": "object",
"properties": {
"name": {
"type": "string"
},
"version": {
"type": "string",
"pattern": "^\\d+\\.\\d+\\.\\d+$"
},
"category": {
"type": "string",
"enum": [
"api",
"testing",
"devops",
"security",
"database",
"frontend",
"algorithms",
"machine-learning",
"cloud",
"containers",
"general"
]
}
},
"required": [
"name",
"version"
]
},
"settings": {
"type": "object",
"properties": {
"enabled": {
"type": "boolean",
"default": true
},
"log_level": {
"type": "string",
"enum": [
"debug",
"info",
"warn",
"error"
]
}
}
}
},
"required": [
"skill"
]
}Kotlin Flow Guide
StateFlow
class ViewModel {
private val _state = MutableStateFlow(UiState())
val state: StateFlow<UiState> = _state.asStateFlow()
fun updateState(newState: UiState) {
_state.update { newState }
}
}SharedFlow
class EventBus {
private val _events = MutableSharedFlow<Event>()
val events: SharedFlow<Event> = _events.asSharedFlow()
suspend fun emit(event: Event) {
_events.emit(event)
}
}Flow Operators
repository.observeUsers()
.map { it.filter { user -> user.isActive } }
.distinctUntilChanged()
.debounce(300)
.catch { emit(emptyList()) }
.collect { users -> updateUI(users) }Kotlin Flow Patterns
Design Patterns
Pattern 1: Input Validation
Always validate input before processing:
def validate_input(data):
if data is None:
raise ValueError("Data cannot be None")
if not isinstance(data, dict):
raise TypeError("Data must be a dictionary")
return TruePattern 2: Error Handling
Use consistent error handling:
try:
result = risky_operation()
except SpecificError as e:
logger.error(f"Operation failed: {e}")
handle_error(e)
except Exception as e:
logger.exception("Unexpected error")
raisePattern 3: Configuration Loading
Load and validate configuration:
import yaml
def load_config(config_path):
with open(config_path) as f:
config = yaml.safe_load(f)
validate_config(config)
return configAnti-Patterns to Avoid
❌ Don't: Swallow Exceptions
# BAD
try:
do_something()
except:
pass✅ Do: Handle Explicitly
# GOOD
try:
do_something()
except SpecificError as e:
logger.warning(f"Expected error: {e}")
return default_valueCategory-Specific Patterns: General
Recommended Approach
1. Start with the simplest implementation 2. Add complexity only when needed 3. Test each addition 4. Document decisions
Common Integration Points
- Configuration:
assets/config.yaml - Validation:
scripts/validate.py - Documentation:
references/GUIDE.md
---
Pattern library for kotlin-flow skill
#!/usr/bin/env python3
"""
Validation script for kotlin-flow skill.
Category: general
"""
import os
import sys
import yaml
import json
from pathlib import Path
def validate_config(config_path: str) -> dict:
"""
Validate skill configuration file.
Args:
config_path: Path to config.yaml
Returns:
dict: Validation result with 'valid' and 'errors' keys
"""
errors = []
if not os.path.exists(config_path):
return {"valid": False, "errors": ["Config file not found"]}
try:
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
except yaml.YAMLError as e:
return {"valid": False, "errors": [f"YAML parse error: {e}"]}
# Validate required fields
if 'skill' not in config:
errors.append("Missing 'skill' section")
else:
if 'name' not in config['skill']:
errors.append("Missing skill.name")
if 'version' not in config['skill']:
errors.append("Missing skill.version")
# Validate settings
if 'settings' in config:
settings = config['settings']
if 'log_level' in settings:
valid_levels = ['debug', 'info', 'warn', 'error']
if settings['log_level'] not in valid_levels:
errors.append(f"Invalid log_level: {settings['log_level']}")
return {
"valid": len(errors) == 0,
"errors": errors,
"config": config if not errors else None
}
def validate_skill_structure(skill_path: str) -> dict:
"""
Validate skill directory structure.
Args:
skill_path: Path to skill directory
Returns:
dict: Structure validation result
"""
required_dirs = ['assets', 'scripts', 'references']
required_files = ['SKILL.md']
errors = []
# Check required files
for file in required_files:
if not os.path.exists(os.path.join(skill_path, file)):
errors.append(f"Missing required file: {file}")
# Check required directories
for dir in required_dirs:
dir_path = os.path.join(skill_path, dir)
if not os.path.isdir(dir_path):
errors.append(f"Missing required directory: {dir}/")
else:
# Check for real content (not just .gitkeep)
files = [f for f in os.listdir(dir_path) if f != '.gitkeep']
if not files:
errors.append(f"Directory {dir}/ has no real content")
return {
"valid": len(errors) == 0,
"errors": errors,
"skill_name": os.path.basename(skill_path)
}
def main():
"""Main validation entry point."""
skill_path = Path(__file__).parent.parent
print(f"Validating kotlin-flow skill...")
print(f"Path: {skill_path}")
# Validate structure
structure_result = validate_skill_structure(str(skill_path))
print(f"\nStructure validation: {'PASS' if structure_result['valid'] else 'FAIL'}")
if structure_result['errors']:
for error in structure_result['errors']:
print(f" - {error}")
# Validate config
config_path = skill_path / 'assets' / 'config.yaml'
if config_path.exists():
config_result = validate_config(str(config_path))
print(f"\nConfig validation: {'PASS' if config_result['valid'] else 'FAIL'}")
if config_result['errors']:
for error in config_result['errors']:
print(f" - {error}")
else:
print("\nConfig validation: SKIPPED (no config.yaml)")
# Summary
all_valid = structure_result['valid']
print(f"\n==================================================")
print(f"Overall: {'VALID' if all_valid else 'INVALID'}")
return 0 if all_valid else 1
if __name__ == "__main__":
sys.exit(main())
Related skills
FAQ
What does kotlin-flow do?
kotlin-flow is a Claude Code skill for ai & agent building.
When should I use kotlin-flow?
When you need to helps with ai & agent building tasks., or when kotlin-flow is a claude code skill for ai & agent building.
What are the main capabilities?
kotlin-flow; AI & Agent Building; AI-coding skill.