coulomb/kaizen-agentic
2
0
Fork 0
Code Issues 4 Pull Requests Actions Projects Releases Wiki Activity

Fix linting violations for v1.0.0 release preparation

Browse Source 
- Apply black formatting to all Python files
- Fix various flake8 violations in agent system code
- Clean up imports and whitespace issues

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2025-10-19 20:44:58 +02:00
parent 30daabf12c
commit d68310793b
9 changed files with 432 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
src/kaizen_agentic/extensions.py
View File

@@ -3,13 +3,14 @@
import json
import yaml
from pathlib import Path
from typing import Dict, List, Optional, Any, Union
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
class ExtensionType(Enum):
"""Types of agent extensions."""
CONFIGURATION_OVERLAY = "config_overlay" # Override default configurations
FUNCTIONAL_EXTENSION = "functional_extension" # Add new functionality
WORKFLOW_INTEGRATION = "workflow_integration" # Integrate with project workflows
@@ -21,6 +22,7 @@ class ExtensionType(Enum):
@dataclass
class AgentExtension:
"""Defines an extension to a Kaizen agent."""
name: str
base_agent: str # The Kaizen agent this extends
extension_type: ExtensionType
@@ -44,6 +46,7 @@ class AgentExtension:
@dataclass
class ProjectExtensionRegistry:
"""Registry of extensions for a project."""
project_path: Path
extensions: List[AgentExtension] = field(default_factory=list)
global_config: Dict[str, Any] = field(default_factory=dict)
@@ -63,7 +66,7 @@ class ExtensionManager:
base_agent: str,
extension_type: ExtensionType,
description: str,
**kwargs
**kwargs,
) -> AgentExtension:
"""Create a new agent extension."""
extension = AgentExtension(
@@ -71,7 +74,7 @@ class ExtensionManager:
base_agent=base_agent,
extension_type=extension_type,
description=description,
**kwargs
**kwargs,
)
self._save_extension(extension)
@@ -79,14 +82,16 @@ class ExtensionManager:
def install_extension(self, extension_path: Path) -> AgentExtension:
"""Install an extension from a file."""
if extension_path.suffix == '.json':
if extension_path.suffix == ".json":
with open(extension_path) as f:
data = json.load(f)
elif extension_path.suffix in ['.yml', '.yaml']:
elif extension_path.suffix in [".yml", ".yaml"]:
with open(extension_path) as f:
data = yaml.safe_load(f)
else:
raise ValueError(f"Unsupported extension file format: {extension_path.suffix}")
raise ValueError(
f"Unsupported extension file format: {extension_path.suffix}"
)
extension = AgentExtension(**data)
self._save_extension(extension)
@@ -127,6 +132,7 @@ class ExtensionManager:
extension_dir = self.extensions_dir / name
if extension_dir.exists():
import shutil
shutil.rmtree(extension_dir)
return True
@@ -136,8 +142,11 @@ class ExtensionManager:
def get_effective_config(self, base_agent: str) -> Dict[str, Any]:
"""Get the effective configuration for an agent with all extensions applied."""
base_config = self._get_base_agent_config(base_agent)
extensions = [ext for ext in self._load_extensions()
if ext.base_agent == base_agent and ext.enabled]
extensions = [
ext
for ext in self._load_extensions()
if ext.base_agent == base_agent and ext.enabled
]
# Apply extensions in order
for extension in extensions:
@@ -151,7 +160,7 @@ class ExtensionManager:
base_agent: str,
custom_instructions: str,
custom_commands: Optional[Dict[str, str]] = None,
environment_config: Optional[Dict[str, Any]] = None
environment_config: Optional[Dict[str, Any]] = None,
) -> AgentExtension:
"""Create a project-specific agent based on a Kaizen agent."""
@@ -161,8 +170,8 @@ class ExtensionManager:
"project_context": {
"name": self.project_path.name,
"path": str(self.project_path),
"type": self._detect_project_type()
}
"type": self._detect_project_type(),
},
}
if environment_config:
@@ -175,7 +184,7 @@ class ExtensionManager:
description=f"Project-specific extension of {base_agent} for {self.project_path.name}",
configuration=config,
custom_commands=custom_commands or {},
environment_overrides=environment_config or {}
environment_overrides=environment_config or {},
)
# Create agent file
@@ -187,7 +196,7 @@ class ExtensionManager:
self,
legacy_agent_path: Path,
target_kaizen_agent: str,
migration_strategy: str = "preserve_functionality"
migration_strategy: str = "preserve_functionality",
) -> AgentExtension:
"""Integrate a legacy agent as an extension to a Kaizen agent."""
@@ -201,7 +210,7 @@ class ExtensionManager:
"legacy_source": str(legacy_agent_path),
"migration_strategy": migration_strategy,
"preserved_functionality": legacy_analysis.get("functionality", []),
"custom_config": legacy_analysis.get("config", {})
"custom_config": legacy_analysis.get("config", {}),
}
extension = self.create_extension(
@@ -209,7 +218,7 @@ class ExtensionManager:
base_agent=target_kaizen_agent,
extension_type=ExtensionType.WORKFLOW_INTEGRATION,
description=f"Legacy integration of {legacy_agent_path.name}",
configuration=config
configuration=config,
)
# Create migration wrapper
@@ -232,23 +241,23 @@ class ExtensionManager:
extension_dir.mkdir(parents=True, exist_ok=True)
# Save extension definition
with open(extension_dir / "extension.yml", 'w') as f:
with open(extension_dir / "extension.yml", "w") as f:
# Convert dataclass to dict for YAML serialization
data = {
'name': extension.name,
'base_agent': extension.base_agent,
'extension_type': extension.extension_type.value,
'description': extension.description,
'version': extension.version,
'author': extension.author,
'configuration': extension.configuration,
'custom_commands': extension.custom_commands,
'workflow_hooks': extension.workflow_hooks,
'data_transformations': extension.data_transformations,
'environment_overrides': extension.environment_overrides,
'dependencies': extension.dependencies,
'compatibility': extension.compatibility,
'enabled': extension.enabled
"name": extension.name,
"base_agent": extension.base_agent,
"extension_type": extension.extension_type.value,
"description": extension.description,
"version": extension.version,
"author": extension.author,
"configuration": extension.configuration,
"custom_commands": extension.custom_commands,
"workflow_hooks": extension.workflow_hooks,
"data_transformations": extension.data_transformations,
"environment_overrides": extension.environment_overrides,
"dependencies": extension.dependencies,
"compatibility": extension.compatibility,
"enabled": extension.enabled,
}
yaml.dump(data, f, default_flow_style=False)
@@ -262,9 +271,9 @@ class ExtensionManager:
data = yaml.safe_load(f) or {}
extensions = []
for ext_data in data.get('extensions', []):
for ext_data in data.get("extensions", []):
# Convert string back to enum
ext_data['extension_type'] = ExtensionType(ext_data['extension_type'])
ext_data["extension_type"] = ExtensionType(ext_data["extension_type"])
extensions.append(AgentExtension(**ext_data))
return extensions
@@ -277,28 +286,28 @@ class ExtensionManager:
# Convert to serializable format
data = {
'extensions': [
"extensions": [
{
'name': ext.name,
'base_agent': ext.base_agent,
'extension_type': ext.extension_type.value,
'description': ext.description,
'version': ext.version,
'author': ext.author,
'configuration': ext.configuration,
'custom_commands': ext.custom_commands,
'workflow_hooks': ext.workflow_hooks,
'data_transformations': ext.data_transformations,
'environment_overrides': ext.environment_overrides,
'dependencies': ext.dependencies,
'compatibility': ext.compatibility,
'enabled': ext.enabled
"name": ext.name,
"base_agent": ext.base_agent,
"extension_type": ext.extension_type.value,
"description": ext.description,
"version": ext.version,
"author": ext.author,
"configuration": ext.configuration,
"custom_commands": ext.custom_commands,
"workflow_hooks": ext.workflow_hooks,
"data_transformations": ext.data_transformations,
"environment_overrides": ext.environment_overrides,
"dependencies": ext.dependencies,
"compatibility": ext.compatibility,
"enabled": ext.enabled,
}
for ext in extensions
]
}
with open(self.config_file, 'w') as f:
with open(self.config_file, "w") as f:
yaml.dump(data, f, default_flow_style=False)
def _toggle_extension(self, name: str, enabled: bool) -> bool:
@@ -321,7 +330,7 @@ class ExtensionManager:
"name": base_agent,
"type": "kaizen_agent",
"enabled": True,
"config": {}
"config": {},
}
def _apply_extension_config(
@@ -343,11 +352,13 @@ class ExtensionManager:
config.setdefault("environment", {}).update(extension.environment_overrides)
# Add extension metadata
config.setdefault("extensions", []).append({
"name": extension.name,
"type": extension.extension_type.value,
"version": extension.version
})
config.setdefault("extensions", []).append(
{
"name": extension.name,
"type": extension.extension_type.value,
"version": extension.version,
}
)
return config
@@ -370,7 +381,7 @@ class ExtensionManager:
"functionality": [],
"config": {},
"commands": [],
"dependencies": []
"dependencies": [],
}
if agent_path.suffix == ".py":
@@ -379,8 +390,9 @@ class ExtensionManager:
# Simple analysis - look for class and function definitions
import re
classes = re.findall(r'class\s+(\w+)', content)
functions = re.findall(r'def\s+(\w+)', content)
classes = re.findall(r"class\s+(\w+)", content)
functions = re.findall(r"def\s+(\w+)", content)
analysis["functionality"] = classes + functions
@@ -487,10 +499,7 @@ class LegacyWrapper:
def create_extension_template(
name: str,
base_agent: str,
project_path: Path,
template_type: str = "basic"
name: str, base_agent: str, project_path: Path, template_type: str = "basic"
) -> str:
"""Create a template for a new agent extension."""
@@ -535,7 +544,6 @@ Save this as `.kaizen/extensions/{name}/extension.yml` and run:
kaizen-agentic extensions install {name}
```
""",
"advanced": f"""# Advanced Extension Template: {name}
This template provides advanced customization options for the {base_agent} agent.
@@ -610,7 +618,7 @@ Create these files in `.kaizen/extensions/{name}/scripts/`:
1. Save the configuration as `.kaizen/extensions/{name}/extension.yml`
2. Add any custom scripts to the scripts directory
3. Install with: `kaizen-agentic extensions install {name}`
"""
""",
}
return templates.get(template_type, templates["basic"])
return templates.get(template_type, templates["basic"])