Writing Gears¶
Complete guide to developing Axium Gears — privileged automation extensions with enhanced capabilities.
When to Use Gears¶
Choose Gears over Spokes when you need:
- Tmux pane management - Spawn commands in split panes
- Long-running processes - Background jobs with lifecycle management
- Filesystem access - Read/write files with permissions
- Command prefix registration - Dynamically wrap system commands
- Privileged IPC operations - HUD updates, notifications, file operations
Example use cases: - Ansible playbook runner with tmux output - Terraform workspace manager with state file access - Docker Compose orchestrator with context switching - CI/CD job runner with artifact management
Project Structure¶
Create a gear directory in ~/.config/axium/gears/<name>/:
~/.config/axium/gears/
my-gear/
gear.yaml # Manifest
main.py # Entry point
runner.py # Additional modules (optional)
utils.py # Helper functions (optional)
Step 1: Create Manifest¶
Create gear.yaml with metadata and permissions:
name: my-gear
version: 0.1.0
description: My automation gear
author: Your Name
entrypoint: main:register
permissions:
exec: true # Allow subprocess execution
notify: true # Allow notifications
net: false # Deny network access
ipc:
- tmux_split_run # Create tmux panes
- hud_update # Update HUD
- notify # Send notifications
- read_file # Read files via daemon
fs_read:
- ~/.my-tool/** # Read tool config
- ~/projects/** # Read project files
fs_write:
- ~/.my-tool/cache/** # Write cache
- ~/projects/logs/** # Write logs
# Optional: Register command wrappers
prefixes:
- command: my-tool
wrapper: axium my-gear-run
Manifest Guidelines¶
- Be conservative with permissions - Only request what you need
- Document filesystem patterns - Use comments for clarity
- Limit IPC actions - Only include actions you'll use
- Version semantically - Follow semver for version field
Step 2: Implement register()¶
Create main.py with the registration function:
"""
My Gear - Automation example.
"""
def register(app, events):
"""
Register gear commands and event handlers.
This function is called by Axium when the gear loads.
Args:
app: Typer CLI application for command registration
events: EventBus for event subscriptions
"""
from axium.core import api
import typer
@app.command("my-gear-run")
def my_gear_run(
playbook: str = typer.Argument(..., help="Playbook to run"),
check: bool = typer.Option(False, "--check", help="Dry run mode")
):
"""
Run my-tool automation in tmux pane.
"""
# Build command
cmd = f"my-tool execute {playbook}"
if check:
cmd += " --check"
# Run in tmux pane (requires ipc: [tmux_split_run])
result = api.tmux_split_run(
gear_name="my-gear",
command=cmd,
height=20,
cwd="~/projects"
)
if result["ok"]:
pane_id = result.get("pane_id")
print(f"✓ Running in pane {pane_id}")
# Update HUD (requires ipc: [hud_update])
api.update_hud_segment("my-gear", "[my-gear:RUNNING]")
else:
error = result.get("error", "Unknown error")
print(f"✗ Failed: {error}")
raise typer.Exit(1)
@app.command("my-gear-status")
def my_gear_status():
"""
Show gear status.
"""
# Read config file (requires ipc: [read_file])
result = api.read_file("my-gear", "~/.my-tool/config.yaml")
if result["ok"]:
content = result["content"]
print(f"Config loaded: {len(content)} bytes")
else:
print("✗ Failed to read config")
# React to environment changes
def on_env_change(new_env, old_env):
"""Handle environment switch."""
# Send notification (requires permissions: notify: true, ipc: [notify])
api.notify_send_cli(
"my-gear",
"Environment Changed",
f"Switched from {old_env} to {new_env}"
)
events.on("env_change", on_env_change)
# React to gear loaded event
def on_gear_loaded(gear_name):
"""Initialize when gear loads."""
if gear_name == "my-gear":
print("[my-gear] Initialized")
events.on("gear_loaded", on_gear_loaded)
Step 3: Using the API¶
Gears access privileged operations through axium.core.api:
Tmux Pane Management¶
from axium.core import api
# Create pane and run command
result = api.tmux_split_run(
gear_name="my-gear",
command="long-running-process",
height=20, # Percentage of window height
cwd="~/projects", # Working directory
env_vars={"FOO": "bar"} # Additional environment vars
)
# result: {"ok": True, "pane_id": "%3"}
Requires: ipc: [tmux_split_run]
HUD Updates¶
from axium.core import api
# Update HUD segment
api.update_hud_segment("my-gear", "[my-gear:ACTIVE]")
# Clear HUD segment
api.update_hud_segment("my-gear", "")
Requires: ipc: [hud_update]
Notifications¶
from axium.core import api
# Send notification
api.notify_send_cli(
spoke_name="my-gear",
title="Job Complete",
body="Playbook finished successfully",
level="info" # info, warning, error
)
Requires: permissions: {notify: true} and ipc: [notify]
File Operations¶
from axium.core import api
# Read file (via daemon)
result = api.read_file("my-gear", "~/.my-tool/config.yaml")
if result["ok"]:
content = result["content"] # str
# Write file (via daemon)
result = api.write_file("my-gear", "~/projects/output.txt", "Hello\n")
if result["ok"]:
print("File written")
Requires: ipc: [read_file] and/or ipc: [write_file]
Requires: Matching fs_read or fs_write patterns
Environment Access¶
from axium.core import api
# Get environment data
result = api.get_env_data("my-gear")
if result["ok"]:
env_data = result["env_data"]
prefix = env_data.get("prefix")
region = env_data.get("region")
Requires: ipc: [get_env_data]
Step 4: Event Handling¶
Subscribe to Axium events:
Available Events¶
| Event | Args | Description |
|---|---|---|
env_change |
new_env, old_env |
Environment switched |
gear_loaded |
gear_name |
Gear finished loading |
spoke_loaded |
spoke_name |
Spoke finished loading |
hud_refresh |
None | HUD needs refresh |
daemon_ready |
None | Daemon startup complete |
Example Handlers¶
def register(app, events):
# Environment changes
def on_env_change(new_env, old_env):
print(f"Env: {old_env} → {new_env}")
# Reload gear config for new environment
load_config_for_env(new_env)
events.on("env_change", on_env_change)
# Daemon startup
def on_daemon_ready():
# Initialize gear state
initialize_gear_state()
events.on("daemon_ready", on_daemon_ready)
# Other gear loaded
def on_gear_loaded(gear_name):
if gear_name == "terraform":
# Coordinate with terraform gear
setup_terraform_integration()
events.on("gear_loaded", on_gear_loaded)
Step 5: Prefix Registration¶
Gears can register command prefixes in their manifest:
# gear.yaml
prefixes:
- command: ansible-playbook
wrapper: axium ansible-run
- command: terraform
wrapper: axium tf-run
When these commands are executed, they'll be intercepted and wrapped:
Prefix Conflict Detection¶
If multiple gears register the same command, Axium detects the conflict:
[WARNING] Prefix conflict: command 'terraform' already registered by gear 'terraform-gear'
[WARNING] Gear 'my-gear' attempted to register 'terraform' but was rejected
First gear to register wins. Users can resolve conflicts by: - Disabling one gear - Removing prefix from one manifest - Using permission overrides
Step 6: Testing¶
Manual Testing¶
# Reload daemon to pick up gear changes
axium daemon reload
# Test gear commands
axium my-gear-run playbook.yml
axium my-gear-status
# Check gear loaded correctly
axium gear list
# View permissions
axium gear perms-show my-gear
# Check logs
axium daemon logs -f
Unit Testing¶
# tests/test_my_gear.py
import pytest
from pathlib import Path
def test_gear_manifest():
"""Test manifest is valid YAML."""
gear_path = Path("~/.config/axium/gears/my-gear").expanduser()
manifest = gear_path / "gear.yaml"
assert manifest.exists()
import yaml
data = yaml.safe_load(manifest.read_text())
assert data["name"] == "my-gear"
assert "permissions" in data
assert "entrypoint" in data
def test_register_function():
"""Test register function exists."""
from gears.my_gear import main
assert hasattr(main, "register")
assert callable(main.register)
Step 7: Permission Overrides¶
Allow users to customize permissions:
Create Override Template¶
This creates ~/.config/axium/overrides/permissions/my-gear.yaml:
# Permission overrides for gear: my-gear
# See base permissions in: ~/.config/axium/gears/my-gear/gear.yaml
# Uncomment to override:
# exec: true
# notify: false
# net: true
# ipc:
# - tmux_split_run
# - hud_update
# fs_read:
# - ~/custom/path/**
# fs_write:
# - ~/custom/output/**
Test Overrides¶
# View effective permissions (base + overrides)
axium gear perms-show my-gear
# Reload to apply changes
axium daemon reload
Best Practices¶
1. Security¶
- Minimal permissions - Only request what you need
- Validate inputs - Sanitize user input before executing
- Error handling - Gracefully handle permission denials
- Audit logging - Log privileged operations
2. User Experience¶
- Clear command names - Use
<gear>-<action>pattern - Help text - Document all commands and options
- Error messages - Provide actionable error messages
- Progress feedback - Show status for long operations
3. Development¶
- Type hints - Use type annotations for clarity
- Docstrings - Document all public functions
- Testing - Write unit tests for core logic
- Versioning - Follow semver for releases
4. Integration¶
- Event coordination - Use events to coordinate with other gears
- HUD updates - Keep HUD in sync with gear state
- Configuration - Use Axium's environment system
- Logging - Use Python logging module
Example: Complete Ansible Gear¶
# ~/.config/axium/gears/ansible/main.py
"""
Ansible Gear - Run playbooks in tmux with environment context.
"""
import typer
from pathlib import Path
from axium.core import api
def register(app, events):
"""Register ansible gear commands."""
@app.command("ansible-run")
def ansible_run(
playbook: str = typer.Argument(..., help="Playbook file"),
inventory: str = typer.Option(None, "-i", help="Inventory file"),
check: bool = typer.Option(False, "--check", help="Dry run"),
diff: bool = typer.Option(False, "--diff", help="Show diffs")
):
"""
Run Ansible playbook in tmux pane with environment context.
"""
# Get current environment data
env_result = api.get_env_data("ansible")
if not env_result["ok"]:
print("✗ Failed to get environment data")
raise typer.Exit(1)
env_data = env_result["env_data"]
# Build command
cmd = f"ansible-playbook {playbook}"
if inventory:
cmd += f" -i {inventory}"
elif "ansible_inventory" in env_data:
# Use environment-specific inventory
cmd += f" -i {env_data['ansible_inventory']}"
if check:
cmd += " --check"
if diff:
cmd += " --diff"
# Add environment variables
env_vars = {}
if "ansible_vault_password_file" in env_data:
env_vars["ANSIBLE_VAULT_PASSWORD_FILE"] = env_data["ansible_vault_password_file"]
print(f"Running: {cmd}")
# Execute in tmux pane
result = api.tmux_split_run(
gear_name="ansible",
command=cmd,
height=30,
cwd=Path.cwd(),
env_vars=env_vars
)
if result["ok"]:
pane_id = result["pane_id"]
print(f"✓ Playbook running in pane {pane_id}")
# Update HUD
api.update_hud_segment("ansible", "[ansible:RUNNING]")
# Send notification
api.notify_send_cli(
"ansible",
"Playbook Started",
f"Running {playbook} in {pane_id}"
)
else:
error = result.get("error", "Unknown error")
print(f"✗ Failed to start playbook: {error}")
raise typer.Exit(1)
@app.command("ansible-check")
def ansible_check():
"""Check Ansible installation and configuration."""
# Read ansible.cfg
config_result = api.read_file("ansible", "~/ansible/ansible.cfg")
if config_result["ok"]:
print("✓ Ansible config found")
else:
print("✗ No ansible.cfg found")
# Get environment data
env_result = api.get_env_data("ansible")
if env_result["ok"]:
env_data = env_result["env_data"]
print(f"✓ Environment: {env_data.get('name', 'unknown')}")
if "ansible_inventory" in env_data:
print(f" Inventory: {env_data['ansible_inventory']}")
else:
print("✗ Failed to get environment")
# Event handlers
def on_env_change(new_env, old_env):
"""Clear HUD when environment changes."""
api.update_hud_segment("ansible", "")
events.on("env_change", on_env_change)
With manifest:
# ~/.config/axium/gears/ansible/gear.yaml
name: ansible
version: 1.0.0
description: Ansible playbook automation with tmux
author: Axium Team
entrypoint: main:register
permissions:
exec: true
notify: true
net: false
ipc:
- tmux_split_run
- hud_update
- notify
- read_file
- get_env_data
fs_read:
- ~/.ansible/**
- ~/ansible/**
- ~/.ssh/config
fs_write:
- ~/.ansible/tmp/**
- ~/ansible/logs/**
prefixes:
- command: ansible-playbook
wrapper: axium ansible-run
Troubleshooting¶
Permission Denied Errors¶
Solution: Add permission to gear.yaml:
Gear Not Loading¶
Check daemon logs:
Common issues:
- Syntax error in gear.yaml
- Missing entrypoint function
- Import errors in gear code
Prefix Not Working¶
# Check if prefix registered
axium wrapper list
# Check for conflicts
axium daemon logs | grep "Prefix conflict"
Official Gears¶
The Axium project maintains several official gears that demonstrate best practices and provide common functionality:
axium-gear-ansible¶
Profile-based ansible-playbook execution with environment awareness. Simplifies Ansible operations with preconfigured profiles and environment-specific inventory management.
Features: - Profile-based playbook execution - Environment-aware inventory selection - Automatic verbosity control - Tag filtering support - Dry-run mode - Custom variables per profile
Permissions:
- exec: Run ansible-playbook commands
- fs_read: Access inventory and playbook files
- notify: Send completion notifications
axium-gear-web¶
Web dashboard for Axium state, permissions, and notifications. Provides a modern web interface for monitoring and managing Axium.
Features: - Real-time state visualization - Permission management UI - Notification history viewer - Environment switching - Gear and spoke status monitoring - RESTful API for external integrations
Permissions:
- exec: Start embedded web server
- fs_read: Read Axium configuration
- ipc: Access daemon state and notifications
See Also¶
- Gears Concept - Architecture and permissions
- Gears API Reference - Complete API docs
- Spokes Guide - Alternative plugin system
- Permissions - Permission system details