API Reference

Complete reference for axium.core.api - the high-level Python API for spokes and gears.

Use API over IPC

Spokes and gears should always use this API instead of calling IPC commands directly. The API provides:

• Error handling - Graceful failures with logging
• Type safety - Proper type hints and validation
• Better ergonomics - Simpler function signatures
• Permission checks - Automatic validation for gears
• Future compatibility - IPC protocol changes won't break your code

Quick Start

from axium.core import api

# Environment management
env = api.get_active_env()
api.set_active_env("prod")

# Configuration
config = api.load_config("myspoke", "config.yaml")

# Notifications
api.send_notification("myspoke", "Alert", "Something happened", level="warning")

# Daemon control
api.reload_daemon()
status = api.daemon_status()

API vs IPC

Use Case	Recommended Approach
Spokes	Use api.* functions exclusively
Gears	Use api.* functions (permission-checked)
Core daemon	Direct IPC for internal operations
CLI commands	Use api.* for most operations
Testing	Mock api functions, not IPC

See IPC Commands Reference for low-level protocol details.

---

Core Functions

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

get_active_env()

Get the name of the currently active environment.

Returns:

Type	Description
str | None	Active environment name, or None if not set or unreachable.

Example

>>> api.get_active_env()
'prod'

Source code in axium/core/api.py

def get_active_env() -> str | None:
    """
    Get the name of the currently active environment.

    Returns:
        Active environment name, or None if not set or unreachable.

    Example:
        ```python
        >>> api.get_active_env()
        'prod'
        ```
    """
    try:
        from axium.core.env import get_active_env_name

        return get_active_env_name()
    except Exception as e:
        logger.debug("Failed to get active environment: %s", e)
        return None

get_env_data(name)

Get configuration data for a specific environment.

Returns all configuration for the named environment, including variables, inventory paths, and other environment-specific settings.

Parameters:

Name	Type	Description	Default
name	str	Environment name to query	required

Returns:

Type	Description
dict[str, Any] | None	Environment configuration dict, or None if environment doesn't exist

Example

>>> data = api.get_env_data('production')
>>> data['vars']['app_version']
'1.2.3'
>>> data.get('ansible_inventory')
'~/ansible/inventories/production.ini'

Source code in axium/core/api.py

def get_env_data(name: str) -> dict[str, Any] | None:
    """
    Get configuration data for a specific environment.

    Returns all configuration for the named environment, including
    variables, inventory paths, and other environment-specific settings.

    Args:
        name: Environment name to query

    Returns:
        Environment configuration dict, or None if environment doesn't exist

    Example:
        ```python
        >>> data = api.get_env_data('production')
        >>> data['vars']['app_version']
        '1.2.3'
        >>> data.get('ansible_inventory')
        '~/ansible/inventories/production.ini'
        ```
    """
    try:
        from axium.core.env import load_environments

        envs = load_environments()
        return envs.get(name)
    except Exception as e:
        logger.debug("Failed to get environment data for %s: %s", name, e)
        return None

set_active_env(name, pane_id=None)

Set the active environment (global or pane-specific).

Parameters:

Name	Type	Description	Default
name	str	Environment name to activate	required
pane_id	str | None	Optional tmux pane ID for pane-specific environment	None

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.set_active_env("prod")
True
>>> api.set_active_env("dev", pane_id="%1")
True

Source code in axium/core/api.py

def set_active_env(name: str, pane_id: str | None = None) -> bool:
    """
    Set the active environment (global or pane-specific).

    Args:
        name: Environment name to activate
        pane_id: Optional tmux pane ID for pane-specific environment

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.set_active_env("prod")
        True
        >>> api.set_active_env("dev", pane_id="%1")
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        if pane_id:
            resp = send_request_sync(
                {"cmd": "set_pane_env", "pane": pane_id, "env": name}
            )
        else:
            resp = send_request_sync({"cmd": "set_env", "env": name})

        # Log error if daemon rejected the request
        if not resp.get("ok"):
            error = resp.get("error", "Unknown error")
            logger.error("Daemon rejected environment change: %s", error)

        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to set active environment: %s", e)
        return False

get_pane_env(pane_id)

Get environment for specific tmux pane.

Parameters:

Name	Type	Description	Default
pane_id	str	Tmux pane ID	required

Returns:

Type	Description
str | None	Environment name or None if not set

Example

>>> api.get_pane_env("%1")
'dev'

Source code in axium/core/api.py

def get_pane_env(pane_id: str) -> str | None:
    """
    Get environment for specific tmux pane.

    Args:
        pane_id: Tmux pane ID

    Returns:
        Environment name or None if not set

    Example:
        ```python
        >>> api.get_pane_env("%1")
        'dev'
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "get_pane_env", "pane": pane_id})
        if resp.get("ok"):
            return resp.get("env")
        return None
    except Exception as e:
        logger.debug("Failed to get pane environment: %s", e)
        return None

clear_pane_env(pane_id)

Clear pane-specific environment (falls back to global).

Parameters:

Name	Type	Description	Default
pane_id	str	Tmux pane ID	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.clear_pane_env("%1")
True

Source code in axium/core/api.py

def clear_pane_env(pane_id: str) -> bool:
    """
    Clear pane-specific environment (falls back to global).

    Args:
        pane_id: Tmux pane ID

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.clear_pane_env("%1")
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "clear_pane_env", "pane": pane_id})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to clear pane environment: %s", e)
        return False

list_environments()

List all defined environments.

Returns:

Type	Description
list[str]	List of environment names (empty list on error)

Example

>>> api.list_environments()
['dev', 'staging', 'prod']

Source code in axium/core/api.py

def list_environments() -> list[str]:
    """
    List all defined environments.

    Returns:
        List of environment names (empty list on error)

    Example:
        ```python
        >>> api.list_environments()
        ['dev', 'staging', 'prod']
        ```
    """
    try:
        from axium.core import env as env_module

        envs = env_module.load_envs()
        return list(envs.keys()) if envs else []
    except Exception as e:
        logger.debug("Failed to list environments: %s", e)
        return []

get_state()

Get current daemon state.

Returns:

Type	Description
dict[str, Any] | None	State dictionary containing active_env, started time, etc.
dict[str, Any] | None	Returns None if daemon unreachable.

Example

>>> api.get_state()
{'active_env': 'prod', 'started': '2024-10-07T10:30:00', ...}

Source code in axium/core/api.py

def get_state() -> dict[str, Any] | None:
    """
    Get current daemon state.

    Returns:
        State dictionary containing active_env, started time, etc.
        Returns None if daemon unreachable.

    Example:
        ```python
        >>> api.get_state()
        {'active_env': 'prod', 'started': '2024-10-07T10:30:00', ...}
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "get_state"})
        if resp.get("ok"):
            return resp.get("state")
        return None
    except Exception as e:
        logger.debug("Failed to get daemon state: %s", e)
        return None

reload_spokes()

Reload all spokes from disk.

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.reload_spokes()
True

Source code in axium/core/api.py

def reload_spokes() -> bool:
    """
    Reload all spokes from disk.

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.reload_spokes()
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "reload_spokes"})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to reload spokes: %s", e)
        return False

reload_daemon()

Reload daemon configuration from disk.

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.reload_daemon()
True

Source code in axium/core/api.py

def reload_daemon() -> bool:
    """
    Reload daemon configuration from disk.

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.reload_daemon()
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "reload"})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to reload daemon: %s", e)
        return False

---

Notifications & HUD

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

send_notification(spoke, title, body, level='info')

Send notification via daemon.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke name	required
title	str	Notification title	required
body	str	Notification body	required
level	str	Notification level (info, warning, error)	'info'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.send_notification("creds", "Alert", "Credentials expired", level="warning")
True

Source code in axium/core/api.py

def send_notification(spoke: str, title: str, body: str, level: str = "info") -> bool:
    """
    Send notification via daemon.

    Args:
        spoke: Spoke name
        title: Notification title
        body: Notification body
        level: Notification level (info, warning, error)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.send_notification("creds", "Alert", "Credentials expired", level="warning")
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {
                "cmd": "notify",
                "spoke": spoke,
                "title": title,
                "body": body,
                "level": level,
            }
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to send notification: %s", e)
        return False

show_toast(title, body)

Show a simple toast notification (info level).

Parameters:

Name	Type	Description	Default
title	str	Notification title	required
body	str	Notification body	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.show_toast("Success", "Operation completed")
True

Source code in axium/core/api.py

def show_toast(title: str, body: str) -> bool:
    """
    Show a simple toast notification (info level).

    Args:
        title: Notification title
        body: Notification body

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.show_toast("Success", "Operation completed")
        True
        ```
    """
    return send_notification("system", title, body, level="info")

drain_notifications()

Get and clear all queued notifications from daemon.

This drains the notification queue - notifications are removed after reading.

Returns:

Type	Description
list[dict[str, Any]]	List of notification dicts, each containing:
list[dict[str, Any]]	spoke: Source spoke name
list[dict[str, Any]]	title: Notification title
list[dict[str, Any]]	body: Notification body
list[dict[str, Any]]	level: Notification level (info/warning/error)
list[dict[str, Any]]	timestamp: ISO 8601 timestamp

Example

>>> notifications = api.drain_notifications()
>>> for notif in notifications:
...     print(f"{notif['level']}: {notif['title']}")
>>> print(f"Processed {len(notifications)} notifications")
warning: Credentials expired
Processed 3 notifications

Source code in axium/core/api.py

def drain_notifications() -> list[dict[str, Any]]:
    """
    Get and clear all queued notifications from daemon.

    This drains the notification queue - notifications are removed after reading.

    Returns:
        List of notification dicts, each containing:
        - spoke: Source spoke name
        - title: Notification title
        - body: Notification body
        - level: Notification level (info/warning/error)
        - timestamp: ISO 8601 timestamp

    Example:
        ```python
        >>> notifications = api.drain_notifications()
        >>> for notif in notifications:
        ...     print(f"{notif['level']}: {notif['title']}")
        >>> print(f"Processed {len(notifications)} notifications")
        warning: Credentials expired
        Processed 3 notifications
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "notify_drain"})
        if resp.get("ok"):
            return resp.get("notifications", [])
        return []
    except Exception as e:
        logger.debug("Failed to drain notifications: %s", e)
        return []

clear_notifications()

Clear all queued notifications without reading them.

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.clear_notifications()
True

Source code in axium/core/api.py

def clear_notifications() -> bool:
    """
    Clear all queued notifications without reading them.

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.clear_notifications()
        True
        ```
    """
    try:
        # Drain and discard
        drain_notifications()
        return True
    except Exception:
        return False

hud_get(pane_id=None)

Get HUD status line.

Parameters:

Name	Type	Description	Default
pane_id	str | None	Optional tmux pane ID for pane-specific HUD	None

Returns:

Type	Description
str | None	HUD status line string or None on error

Example

>>> hud = api.hud_get()
>>> print(hud)
'[env:prod] [creds:Y] [aws:us-east-1]'

Source code in axium/core/api.py

def hud_get(pane_id: str | None = None) -> str | None:
    """
    Get HUD status line.

    Args:
        pane_id: Optional tmux pane ID for pane-specific HUD

    Returns:
        HUD status line string or None on error

    Example:
        ```python
        >>> hud = api.hud_get()
        >>> print(hud)
        '[env:prod] [creds:Y] [aws:us-east-1]'
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        req = {"cmd": "hud"}
        if pane_id:
            req["pane"] = pane_id

        resp = send_request_sync(req)
        if resp.get("ok"):
            return resp.get("hud", "")
        return None
    except Exception as e:
        logger.debug("Failed to get HUD: %s", e)
        return None

---

Daemon Control

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

daemon_exec(spoke, command, mode='background')

Execute command via daemon.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke name requesting execution	required
command	str	Shell command to execute	required
mode	str	Execution mode (background, foreground)	'background'

Returns:

Type	Description
bool	True if request accepted, False otherwise

Example

>>> api.daemon_exec("creds", "aws sso login")
True

Source code in axium/core/api.py

def daemon_exec(spoke: str, command: str, mode: str = "background") -> bool:
    """
    Execute command via daemon.

    Args:
        spoke: Spoke name requesting execution
        command: Shell command to execute
        mode: Execution mode (background, foreground)

    Returns:
        True if request accepted, False otherwise

    Example:
        ```python
        >>> api.daemon_exec("creds", "aws sso login")
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {"cmd": "daemon_exec", "spoke": spoke, "command": command, "mode": mode}
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to execute command via daemon: %s", e)
        return False

daemon_status()

Get daemon status information.

Returns:

Type	Description
dict[str, Any] | None	Status dictionary or None if unreachable

Example

>>> api.daemon_status()
{'running': True, 'uptime': 3600, ...}

Source code in axium/core/api.py

def daemon_status() -> dict[str, Any] | None:
    """
    Get daemon status information.

    Returns:
        Status dictionary or None if unreachable

    Example:
        ```python
        >>> api.daemon_status()
        {'running': True, 'uptime': 3600, ...}
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "daemon_status"})
        if resp.get("ok"):
            return resp.get("status")
        return None
    except Exception as e:
        logger.debug("Failed to get daemon status: %s", e)
        return None

daemon_logs(tail=False, lines=100)

Get daemon logs.

Parameters:

Name	Type	Description	Default
tail	bool	If True, follow logs (streaming)	False
lines	int	Number of lines to return	100

Returns:

Type	Description
str | None	Log content as string, or None if unreachable

Example

>>> api.daemon_logs(lines=50)
'[2024-10-07 10:30:00] Daemon started\n...'

Source code in axium/core/api.py

def daemon_logs(tail: bool = False, lines: int = 100) -> str | None:
    """
    Get daemon logs.

    Args:
        tail: If True, follow logs (streaming)
        lines: Number of lines to return

    Returns:
        Log content as string, or None if unreachable

    Example:
        ```python
        >>> api.daemon_logs(lines=50)
        '[2024-10-07 10:30:00] Daemon started\\n...'
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "daemon_logs", "tail": tail, "lines": lines})
        if resp.get("ok"):
            return resp.get("logs")
        return None
    except Exception as e:
        logger.debug("Failed to get daemon logs: %s", e)
        return None

daemon_start(debug=False)

Start the Axium daemon.

Parameters:

Name	Type	Description	Default
debug	bool	If True, run daemon in debug mode	False

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.daemon_start()
True

Source code in axium/core/api.py

def daemon_start(debug: bool = False) -> bool:
    """
    Start the Axium daemon.

    Args:
        debug: If True, run daemon in debug mode

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.daemon_start()
        True
        ```
    """
    try:
        from axium.core import daemon as axiumd

        return axiumd.start(debug=debug)
    except Exception as e:
        logger.debug("Failed to start daemon: %s", e)
        return False

daemon_stop(force=True)

Stop the daemon gracefully (or forcefully if needed).

Parameters:

Name	Type	Description	Default
force	bool	If True and graceful stop fails, kill daemon process	True

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.daemon_stop()
True

Source code in axium/core/api.py

def daemon_stop(force: bool = True) -> bool:
    """
    Stop the daemon gracefully (or forcefully if needed).

    Args:
        force: If True and graceful stop fails, kill daemon process

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.daemon_stop()
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "stop"})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to stop daemon gracefully: %s", e)
        if force:
            try:
                from axium.core.daemon import kill_daemon_pid

                kill_daemon_pid()
                return True
            except Exception as kill_err:
                logger.debug("Failed to force-kill daemon: %s", kill_err)
        return False

---

Tmux Integration (Gears Only)

Permission Required

These functions require explicit permissions in your gear's manifest.yaml:

permissions:
  ipc:
    - tmux_split_run
    - tmux_send_keys
    - tmux_capture_pane

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

tmux_split_run(spoke, command, height=20)

Create tmux split pane and run command.

Requires 'tmux_split_run' in spoke/gear's ipc permissions.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (for permission check)	required
command	str	Shell command to execute in split	required
height	int	Split height as percentage (default 20%, range 1-99)	20

Returns:

Type	Description
dict[str, Any]	Response dict with:
dict[str, Any]	ok: True if successful, False if failed
dict[str, Any]	pane_id: Tmux pane ID (e.g., "%42") if successful
dict[str, Any]	error: Error message if failed

Example

>>> result = api.tmux_split_run("ansible", "ansible-playbook site.yml", height=25)
>>> if result["ok"]:
...     print(f"Running in pane: {result['pane_id']}")
Running in pane: %42

Permission

Requires spoke/gear to have 'tmux_split_run' in permissions.ipc list.

Note

Command runs with shell=True in the split pane. Returns immediately (does not wait for command completion).

Source code in axium/core/api.py

def tmux_split_run(spoke: str, command: str, height: int = 20) -> dict[str, Any]:
    """
    Create tmux split pane and run command.

    Requires 'tmux_split_run' in spoke/gear's ipc permissions.

    Args:
        spoke: Spoke/gear name (for permission check)
        command: Shell command to execute in split
        height: Split height as percentage (default 20%, range 1-99)

    Returns:
        Response dict with:
        - ok: True if successful, False if failed
        - pane_id: Tmux pane ID (e.g., "%42") if successful
        - error: Error message if failed

    Example:
        ```python
        >>> result = api.tmux_split_run("ansible", "ansible-playbook site.yml", height=25)
        >>> if result["ok"]:
        ...     print(f"Running in pane: {result['pane_id']}")
        Running in pane: %42
        ```

    Permission:
        Requires spoke/gear to have 'tmux_split_run' in permissions.ipc list.

    Note:
        Command runs with shell=True in the split pane.
        Returns immediately (does not wait for command completion).
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {
                "cmd": "tmux_split_run",
                "spoke": spoke,
                "command": command,
                "height": height,
            }
        )
        return resp
    except Exception as e:
        logger.debug("tmux_split_run failed: %s", e)
        return {"ok": False, "error": str(e)}

tmux_send_keys(spoke, pane_id, keys)

Send keys to specific tmux pane.

Requires 'tmux_send_keys' in spoke/gear's ipc permissions.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (for permission check)	required
pane_id	str	Target pane ID (e.g., "%42")	required
keys	str	Keys to send to pane	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.tmux_send_keys("ansible", "%42", "q")  # Send 'q' to quit
True

Source code in axium/core/api.py

def tmux_send_keys(spoke: str, pane_id: str, keys: str) -> bool:
    """
    Send keys to specific tmux pane.

    Requires 'tmux_send_keys' in spoke/gear's ipc permissions.

    Args:
        spoke: Spoke/gear name (for permission check)
        pane_id: Target pane ID (e.g., "%42")
        keys: Keys to send to pane

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.tmux_send_keys("ansible", "%42", "q")  # Send 'q' to quit
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {"cmd": "tmux_send_keys", "spoke": spoke, "pane_id": pane_id, "keys": keys}
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("tmux_send_keys failed: %s", e)
        return False

tmux_capture_pane(spoke, pane_id)

Capture visible contents of tmux pane.

Requires 'tmux_capture_pane' in spoke/gear's ipc permissions.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (for permission check)	required
pane_id	str	Target pane ID (e.g., "%42")	required

Returns:

Type	Description
str | None	Pane contents as string, or None if failed

Example

>>> output = api.tmux_capture_pane("ansible", "%42")
>>> if "PLAY RECAP" in output:
...     print("Playbook completed!")

Source code in axium/core/api.py

def tmux_capture_pane(spoke: str, pane_id: str) -> str | None:
    """
    Capture visible contents of tmux pane.

    Requires 'tmux_capture_pane' in spoke/gear's ipc permissions.

    Args:
        spoke: Spoke/gear name (for permission check)
        pane_id: Target pane ID (e.g., "%42")

    Returns:
        Pane contents as string, or None if failed

    Example:
        ```python
        >>> output = api.tmux_capture_pane("ansible", "%42")
        >>> if "PLAY RECAP" in output:
        ...     print("Playbook completed!")
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {"cmd": "tmux_capture_pane", "spoke": spoke, "pane_id": pane_id}
        )
        return resp.get("content") if resp.get("ok") else None
    except Exception as e:
        logger.debug("tmux_capture_pane failed: %s", e)
        return None

---

File Operations (Gears Only)

Permission Required

These functions require explicit file path permissions in your gear's manifest.yaml:

permissions:
  fs_read:
    - ~/.my-tool/**
    - /etc/my-tool/*.conf
  fs_write:
    - ~/.my-tool/cache/**
    - ~/.my-tool/logs/**

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

read_file(spoke, path)

Read file contents (permission checked).

Daemon checks path against spoke/gear's fs_read glob patterns.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (for permission check)	required
path	str	File path to read	required

Returns:

Type	Description
str | None	File contents as string, or None if denied/failed

Example

>>> content = api.read_file("ansible", "~/.ansible/config")
>>> if content:
...     print(f"Config size: {len(content)} bytes")

Permission

Requires path to match a pattern in spoke/gear's fs_read list. Supports glob patterns (*, ?, []).

Note

Max file size: 10MB (larger files return error). Tilde (~) expansion performed automatically.

Source code in axium/core/api.py

def read_file(spoke: str, path: str) -> str | None:
    """
    Read file contents (permission checked).

    Daemon checks path against spoke/gear's fs_read glob patterns.

    Args:
        spoke: Spoke/gear name (for permission check)
        path: File path to read

    Returns:
        File contents as string, or None if denied/failed

    Example:
        ```python
        >>> content = api.read_file("ansible", "~/.ansible/config")
        >>> if content:
        ...     print(f"Config size: {len(content)} bytes")
        ```

    Permission:
        Requires path to match a pattern in spoke/gear's fs_read list.
        Supports glob patterns (*, ?, []).

    Note:
        Max file size: 10MB (larger files return error).
        Tilde (~) expansion performed automatically.
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "read_file", "spoke": spoke, "path": path})
        return resp.get("content") if resp.get("ok") else None
    except Exception as e:
        logger.debug("read_file failed: %s", e)
        return None

write_file(spoke, path, content)

Write file contents (permission checked).

Daemon checks path against spoke/gear's fs_write glob patterns.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (for permission check)	required
path	str	File path to write	required
content	str	Content to write	required

Returns:

Type	Description
bool	True if successful, False if denied/failed

Example

>>> success = api.write_file("ansible", "~/ansible/logs/last_run.log", log_data)
>>> if success:
...     print("Log written successfully")

Permission

Requires path to match a pattern in spoke/gear's fs_write list. Supports glob patterns (*, ?, []).

Note

Parent directories created automatically if they don't exist. Existing files are overwritten. Tilde (~) expansion performed automatically.

Source code in axium/core/api.py

def write_file(spoke: str, path: str, content: str) -> bool:
    """
    Write file contents (permission checked).

    Daemon checks path against spoke/gear's fs_write glob patterns.

    Args:
        spoke: Spoke/gear name (for permission check)
        path: File path to write
        content: Content to write

    Returns:
        True if successful, False if denied/failed

    Example:
        ```python
        >>> success = api.write_file("ansible", "~/ansible/logs/last_run.log", log_data)
        >>> if success:
        ...     print("Log written successfully")
        ```

    Permission:
        Requires path to match a pattern in spoke/gear's fs_write list.
        Supports glob patterns (*, ?, []).

    Note:
        Parent directories created automatically if they don't exist.
        Existing files are overwritten.
        Tilde (~) expansion performed automatically.
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {"cmd": "write_file", "spoke": spoke, "path": path, "content": content}
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("write_file failed: %s", e)
        return False

write_log(name, message, level='info')

Write to gear/spoke log file at ~/.config/axium/logs/.log.

Parameters:

Name	Type	Description	Default
name	str	Gear/spoke name	required
message	str	Log message	required
level	str	Log level (debug, info, warning, error)	'info'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.write_log("ansible", "Starting playbook execution")
True
>>> api.write_log("ansible", "Playbook failed!", level="error")
True

Note

Log files automatically created in ~/.config/axium/logs/. Logs are timestamped and include level indicator.

Source code in axium/core/api.py

def write_log(name: str, message: str, level: str = "info") -> bool:
    """
    Write to gear/spoke log file at ~/.config/axium/logs/<name>.log.

    Args:
        name: Gear/spoke name
        message: Log message
        level: Log level (debug, info, warning, error)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.write_log("ansible", "Starting playbook execution")
        True
        >>> api.write_log("ansible", "Playbook failed!", level="error")
        True
        ```

    Note:
        Log files automatically created in ~/.config/axium/logs/.
        Logs are timestamped and include level indicator.
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {"cmd": "write_log", "spoke": name, "message": message, "level": level}
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("write_log failed: %s", e)
        return False

---

Prefix System

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

register_prefix(spoke, command, wrapper)

Register command prefix rule programmatically.

Allows gears/spokes to intercept shell commands and wrap them. Includes conflict detection - returns False if command already registered.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke/gear name (owner of this rule)	required
command	str	Command to intercept (e.g., "ansible-playbook")	required
wrapper	str	Replacement command (e.g., "axium ansible-run")	required

Returns:

Type	Description
bool	True if registered, False if conflict detected

Example

>>> success = api.register_prefix("ansible", "ansible-playbook", "axium ansible-run")
>>> if success:
...     print("Prefix rule registered!")
... else:
...     print("Conflict: command already registered")

Note

Conflict occurs when another spoke/gear already registered same command. Use this during gear's register() function. Rules are automatically unregistered when gear is unloaded.

Source code in axium/core/api.py

def register_prefix(spoke: str, command: str, wrapper: str) -> bool:
    """
    Register command prefix rule programmatically.

    Allows gears/spokes to intercept shell commands and wrap them.
    Includes conflict detection - returns False if command already registered.

    Args:
        spoke: Spoke/gear name (owner of this rule)
        command: Command to intercept (e.g., "ansible-playbook")
        wrapper: Replacement command (e.g., "axium ansible-run")

    Returns:
        True if registered, False if conflict detected

    Example:
        ```python
        >>> success = api.register_prefix("ansible", "ansible-playbook", "axium ansible-run")
        >>> if success:
        ...     print("Prefix rule registered!")
        ... else:
        ...     print("Conflict: command already registered")
        ```

    Note:
        Conflict occurs when another spoke/gear already registered same command.
        Use this during gear's register() function.
        Rules are automatically unregistered when gear is unloaded.
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync(
            {
                "cmd": "register_prefix",
                "spoke": spoke,
                "command": command,
                "wrapper": wrapper,
            }
        )
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("register_prefix failed: %s", e)
        return False

---

Configuration

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

load_config(name, filename, env_aware=True, component_type='spoke')

Load spoke or gear configuration.

Wraps axium.core.config.load_spoke_config / load_gear_config with simplified interface.

Parameters:

Name	Type	Description	Default
name	str	Component name (spoke or gear)	required
filename	str	Config filename (e.g., "myspoke.yaml", "ansible.yaml")	required
env_aware	bool	Whether to apply environment-specific overrides	True
component_type	str	"spoke" or "gear" (default: "spoke")	'spoke'

Returns:

Type	Description
dict[str, Any]	Configuration dictionary (empty dict on error)

Example

>>> api.load_config("creds", "creds.yaml")
{'check': {'type': 'mtime', 'path': '~/.aws/credentials'}, ...}

>>> api.load_config("ansible", "ansible.yaml", component_type="gear")
{'defaults': {'ansible_bin': 'ansible-playbook'}, 'profiles': {...}}

Source code in axium/core/api.py

def load_config(
    name: str, filename: str, env_aware: bool = True, component_type: str = "spoke"
) -> dict[str, Any]:
    """
    Load spoke or gear configuration.

    Wraps axium.core.config.load_spoke_config / load_gear_config with simplified interface.

    Args:
        name: Component name (spoke or gear)
        filename: Config filename (e.g., "myspoke.yaml", "ansible.yaml")
        env_aware: Whether to apply environment-specific overrides
        component_type: "spoke" or "gear" (default: "spoke")

    Returns:
        Configuration dictionary (empty dict on error)

    Example:
        ```python
        >>> api.load_config("creds", "creds.yaml")
        {'check': {'type': 'mtime', 'path': '~/.aws/credentials'}, ...}

        >>> api.load_config("ansible", "ansible.yaml", component_type="gear")
        {'defaults': {'ansible_bin': 'ansible-playbook'}, 'profiles': {...}}
        ```
    """
    try:
        if component_type == "gear":
            from axium.core.config import load_gear_config

            return load_gear_config(name, filename, env_aware=env_aware)
        else:
            from axium.core.config import load_spoke_config

            return load_spoke_config(name, filename, env_aware=env_aware)
    except Exception as e:
        logger.debug("Failed to load config for %s: %s", name, e)
        return {}

get_permissions()

Get current daemon permissions.

Returns:

Type	Description
dict[str, Any] | None	Permissions dictionary or None if unreachable

Example

>>> api.get_permissions()
{'exec': ['creds'], 'network': [], ...}

Source code in axium/core/api.py

def get_permissions() -> dict[str, Any] | None:
    """
    Get current daemon permissions.

    Returns:
        Permissions dictionary or None if unreachable

    Example:
        ```python
        >>> api.get_permissions()
        {'exec': ['creds'], 'network': [], ...}
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "get_permissions"})
        if resp.get("ok"):
            return resp.get("permissions")
        return None
    except Exception as e:
        logger.debug("Failed to get permissions: %s", e)
        return None

check_permission(spoke, permission)

Check if spoke has specific permission.

Parameters:

Name	Type	Description	Default
spoke	str	Spoke name	required
permission	str	Permission type (exec, network, etc.)	required

Returns:

Type	Description
bool	True if permission granted, False otherwise

Example

>>> api.check_permission("creds", "exec")
True

Source code in axium/core/api.py

def check_permission(spoke: str, permission: str) -> bool:
    """
    Check if spoke has specific permission.

    Args:
        spoke: Spoke name
        permission: Permission type (exec, network, etc.)

    Returns:
        True if permission granted, False otherwise

    Example:
        ```python
        >>> api.check_permission("creds", "exec")
        True
        ```
    """
    perms = get_permissions()
    if not perms:
        return False

    perm_list = perms.get(permission, [])
    return spoke in perm_list

config_show()

Get current configuration as YAML string.

Returns:

Type	Description
str | None	Configuration YAML or None on error

Example

>>> config = api.config_show()
>>> print(config)

Source code in axium/core/api.py

def config_show() -> str | None:
    """
    Get current configuration as YAML string.

    Returns:
        Configuration YAML or None on error

    Example:
        ```python
        >>> config = api.config_show()
        >>> print(config)
        ```
    """
    try:
        from axium.core import config as config_module

        return config_module.show_config()
    except Exception as e:
        logger.debug("Failed to show config: %s", e)
        return None

config_edit()

Open configuration in editor.

Returns:

Type	Description
bool	True if editor launched successfully, False otherwise

Example

>>> api.config_edit()
True

Source code in axium/core/api.py

def config_edit() -> bool:
    """
    Open configuration in editor.

    Returns:
        True if editor launched successfully, False otherwise

    Example:
        ```python
        >>> api.config_edit()
        True
        ```
    """
    try:
        from axium.core import config as config_module

        config_module.edit_config()
        return True
    except Exception as e:
        logger.debug("Failed to edit config: %s", e)
        return False

---

Spoke Management

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

list_spokes()

List all installed spokes.

Returns:

Type	Description
list[str]	List of spoke names (empty list on error)

Example

>>> api.list_spokes()
['creds', 'aws', 'k8s']

Source code in axium/core/api.py

def list_spokes() -> list[str]:
    """
    List all installed spokes.

    Returns:
        List of spoke names (empty list on error)

    Example:
        ```python
        >>> api.list_spokes()
        ['creds', 'aws', 'k8s']
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "spoke_list"})
        if resp.get("ok"):
            return resp.get("spokes", [])
        return []
    except Exception as e:
        logger.debug("Failed to list spokes: %s", e)
        return []

install_spoke(name)

Install a spoke.

Parameters:

Name	Type	Description	Default
name	str	Spoke name	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.install_spoke("aws")
True

Note

Stub for future implementation

Source code in axium/core/api.py

def install_spoke(name: str) -> bool:
    """
    Install a spoke.

    Args:
        name: Spoke name

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.install_spoke("aws")
        True
        ```

    Note:
        Stub for future implementation
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "spoke_install", "spoke": name})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to install spoke: %s", e)
        return False

validate_spoke(name)

Validate spoke schema and structure.

Parameters:

Name	Type	Description	Default
name	str	Spoke name	required

Returns:

Type	Description
bool	True if valid, False otherwise

Example

>>> api.validate_spoke("creds")
True

Note

Stub for future implementation

Source code in axium/core/api.py

def validate_spoke(name: str) -> bool:
    """
    Validate spoke schema and structure.

    Args:
        name: Spoke name

    Returns:
        True if valid, False otherwise

    Example:
        ```python
        >>> api.validate_spoke("creds")
        True
        ```

    Note:
        Stub for future implementation
    """
    # TODO: Implement local validation logic
    logger.debug("validate_spoke not yet implemented")
    return False

list_commands(sort_by='grouped')

List all registered commands with full metadata.

Returns all commands from the command registry, including core commands and spoke-registered commands. Each command includes name, summary, description, category, source, and group information.

Parameters:

Name	Type	Description	Default
sort_by	str	Sort method - "alpha" (alphabetical), "grouped" (source grouped), or "usage" (future: by usage frequency). Default: "grouped"	'grouped'

Returns:

Type	Description
list[dict[str, Any]]	List of command dictionaries with complete metadata, or empty list on error

Example

>>> commands = api.list_commands()
>>> for cmd in commands:
...     print(f"{cmd['name']}: {cmd['summary']}")
daemon start: Start the Axium daemon
env set: Set the active environment
creds-status: Check credential status

Note

• Used by palette for command discovery
• Sorted by category by default (core first, then spokes)
• Safe to call even if registry is empty (returns [])

Source code in axium/core/api.py

def list_commands(sort_by: str = "grouped") -> list[dict[str, Any]]:
    """
    List all registered commands with full metadata.

    Returns all commands from the command registry, including core commands
    and spoke-registered commands. Each command includes name, summary,
    description, category, source, and group information.

    Args:
        sort_by: Sort method - "alpha" (alphabetical), "grouped" (source grouped),
                or "usage" (future: by usage frequency). Default: "grouped"

    Returns:
        List of command dictionaries with complete metadata, or empty list on error

    Example:
        ```python
        >>> commands = api.list_commands()
        >>> for cmd in commands:
        ...     print(f"{cmd['name']}: {cmd['summary']}")
        daemon start: Start the Axium daemon
        env set: Set the active environment
        creds-status: Check credential status
        ```

    Note:
        - Used by palette for command discovery
        - Sorted by category by default (core first, then spokes)
        - Safe to call even if registry is empty (returns [])
    """
    try:
        from axium.core import registry

        return registry.get_all_commands(sort_by=sort_by)
    except Exception as e:
        logger.debug("Failed to list commands: %s", e)
        return []

spoke_new(name, prefix=None, check_type=None, check_command=None)

Create new spoke from template.

Parameters:

Name	Type	Description	Default
name	str	Spoke name	required
prefix	str | None	Command prefix (optional)	None
check_type	str | None	Check type (command, mtime, etc.)	None
check_command	str | None	Check command if check_type is 'command'	None

Returns:

Type	Description
str | None	Path to created spoke directory, or None on error

Example

>>> spoke_path = api.spoke_new("myspoke", prefix="myapp")
>>> print(spoke_path)
'/home/user/.config/axium/spokes/myspoke'

Source code in axium/core/api.py

def spoke_new(
    name: str,
    prefix: str | None = None,
    check_type: str | None = None,
    check_command: str | None = None,
) -> str | None:
    """
    Create new spoke from template.

    Args:
        name: Spoke name
        prefix: Command prefix (optional)
        check_type: Check type (command, mtime, etc.)
        check_command: Check command if check_type is 'command'

    Returns:
        Path to created spoke directory, or None on error

    Example:
        ```python
        >>> spoke_path = api.spoke_new("myspoke", prefix="myapp")
        >>> print(spoke_path)
        '/home/user/.config/axium/spokes/myspoke'
        ```
    """
    try:
        from axium.core import spokes as spokes_module

        return spokes_module.create_spoke(
            name=name,
            prefix=prefix,
            check_type=check_type,
            check_command=check_command,
        )
    except Exception as e:
        logger.debug("Failed to create spoke: %s", e)
        return None

spoke_install_from_path(path)

Install spoke from local path or git URL.

Parameters:

Name	Type	Description	Default
path	str	Local directory path or git URL	required

Returns:

Type	Description
dict[str, Any] | None	Spoke metadata dict or None on error

Example

>>> metadata = api.spoke_install_from_path("./my-spoke")
>>> print(metadata['name'])
'my-spoke'

Source code in axium/core/api.py

def spoke_install_from_path(path: str) -> dict[str, Any] | None:
    """
    Install spoke from local path or git URL.

    Args:
        path: Local directory path or git URL

    Returns:
        Spoke metadata dict or None on error

    Example:
        ```python
        >>> metadata = api.spoke_install_from_path("./my-spoke")
        >>> print(metadata['name'])
        'my-spoke'
        ```
    """
    try:
        from axium.core import spokes as spokes_module

        return spokes_module.install_spoke(path)
    except Exception as e:
        logger.debug("Failed to install spoke: %s", e)
        return None

spoke_reload_all()

Reload all spokes from disk.

Returns:

Type	Description
list[str]	List of reloaded spoke names (empty list on error)

Example

>>> reloaded = api.spoke_reload_all()
>>> print(f"Reloaded {len(reloaded)} spokes")

Source code in axium/core/api.py

def spoke_reload_all() -> list[str]:
    """
    Reload all spokes from disk.

    Returns:
        List of reloaded spoke names (empty list on error)

    Example:
        ```python
        >>> reloaded = api.spoke_reload_all()
        >>> print(f"Reloaded {len(reloaded)} spokes")
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "reload_spokes"})
        if resp.get("ok"):
            return resp.get("spokes", [])
        return []
    except Exception as e:
        logger.debug("Failed to reload spokes: %s", e)
        return []

spoke_reload_one(name)

Reload specific spoke from disk.

Parameters:

Name	Type	Description	Default
name	str	Spoke name	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.spoke_reload_one("creds")
True

Source code in axium/core/api.py

def spoke_reload_one(name: str) -> bool:
    """
    Reload specific spoke from disk.

    Args:
        name: Spoke name

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.spoke_reload_one("creds")
        True
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "reload_spoke", "spoke": name})
        return resp.get("ok", False)
    except Exception as e:
        logger.debug("Failed to reload spoke: %s", e)
        return False

---

Command Execution

Axium Core API - High-level interface for spokes.

This module provides a simplified, public API for spokes to interact with the Axium daemon, environment state, configuration, and other core services.

Spokes should import from this module instead of using axium.core.ipc directly. All functions handle errors gracefully and return None on failure.

Example

from axium.core import api

# Get current environment
env = api.get_active_env()

# Load spoke config
config = api.load_config("myspoke", "myspoke.yaml")

# Update HUD
api.update_hud_segment("myspoke", "[myspoke:OK]")

# Send notification
api.send_notification("myspoke", "Alert", "Something happened")

# Execute command via daemon
api.daemon_exec("myspoke", "some-command")

run_command(command, args)

Execute prefixed command via Axium run system.

Parameters:

Name	Type	Description	Default
command	str	Command name (e.g., "aws", "terraform")	required
args	list[str]	Command arguments	required

Returns:

Type	Description
bool	True if command executed successfully, False otherwise

Example

>>> api.run_command("aws", ["s3", "ls"])
True

Source code in axium/core/api.py

def run_command(command: str, args: list[str]) -> bool:
    """
    Execute prefixed command via Axium run system.

    Args:
        command: Command name (e.g., "aws", "terraform")
        args: Command arguments

    Returns:
        True if command executed successfully, False otherwise

    Example:
        ```python
        >>> api.run_command("aws", ["s3", "ls"])
        True
        ```
    """
    try:
        from axium.core import run as run_module

        return run_module.execute_command(command, args)
    except Exception as e:
        logger.debug("Failed to run command: %s", e)
        return False

notify_send_cli(spoke, title, body, level='info')

Send notification via CLI (not from spoke).

Parameters:

Name	Type	Description	Default
spoke	str	Spoke name (for categorization)	required
title	str	Notification title	required
body	str	Notification body	required
level	str	Notification level (info, warning, error)	'info'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

>>> api.notify_send_cli("system", "Test", "Hello", level="info")
True

Source code in axium/core/api.py

def notify_send_cli(spoke: str, title: str, body: str, level: str = "info") -> bool:
    """
    Send notification via CLI (not from spoke).

    Args:
        spoke: Spoke name (for categorization)
        title: Notification title
        body: Notification body
        level: Notification level (info, warning, error)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        >>> api.notify_send_cli("system", "Test", "Hello", level="info")
        True
        ```
    """
    return send_notification(spoke, title, body, level)

notify_drain()

Drain all pending notifications.

Returns:

Type	Description
list[dict[str, Any]]	List of notification dictionaries (empty list on error)

Example

>>> notifications = api.notify_drain()
>>> for notif in notifications:
...     print(f"{notif['title']}: {notif['body']}")

Source code in axium/core/api.py

def notify_drain() -> list[dict[str, Any]]:
    """
    Drain all pending notifications.

    Returns:
        List of notification dictionaries (empty list on error)

    Example:
        ```python
        >>> notifications = api.notify_drain()
        >>> for notif in notifications:
        ...     print(f"{notif['title']}: {notif['body']}")
        ```
    """
    try:
        from axium.core.ipc import send_request_sync

        resp = send_request_sync({"cmd": "notify_drain"})
        if resp.get("ok"):
            return resp.get("notifications", [])
        return []
    except Exception as e:
        logger.debug("Failed to drain notifications: %s", e)
        return []

---

Usage Examples

Environment-Aware Configuration

from axium.core import api

def register(app, events):
    """Spoke registration with environment-aware config."""

    @app.command("mycommand")
    def my_command():
        # Get current environment
        env = api.get_active_env()

        # Load config with env-specific overrides
        config = api.load_config("myspoke", "config.yaml", env_aware=True)

        # Access environment-specific settings
        endpoint = config.get("endpoint")
        timeout = config.get("timeout", 30)

        print(f"Running in {env} with endpoint: {endpoint}")

Notification Patterns

from axium.core import api

# Simple info notification
api.show_toast("Success", "Operation completed")

# Categorized notification with level
api.send_notification(
    spoke="creds",
    title="Credentials Expired",
    body="Run 'axium creds-refresh' to renew",
    level="warning"
)

# Error notification
api.send_notification(
    spoke="deploy",
    title="Deployment Failed",
    body=f"Error: {error_message}",
    level="error"
)

Gear with Permissions

# manifest.yaml
permissions:
  exec: true
  notify: true
  ipc:
    - tmux_split_run
  fs_write:
    - ~/.my-gear/logs/**

# main.py
from axium.core import api

def register(app, events):
    @app.command("my-gear-run")
    def run_command():
        # Execute command (requires exec permission)
        api.daemon_exec("my-gear", "long-running-task")

        # Create split pane (requires tmux_split_run permission)
        result = api.tmux_split_run(
            spoke="my-gear",
            command="tail -f /var/log/app.log",
            height=25
        )

        if result["ok"]:
            pane_id = result["pane_id"]

            # Write log (requires fs_write permission)
            api.write_log(
                "my-gear",
                f"Created monitoring pane: {pane_id}",
                level="info"
            )

            # Notify user (requires notify permission)
            api.send_notification(
                "my-gear",
                "Monitoring Started",
                f"Logs visible in pane {pane_id}"
            )

Event Handlers with State Updates

from axium.core import api

def register(app, events):
    def on_env_change(new_env: str, old_env: str, **kwargs):
        """React to environment changes."""

        # Load new environment config
        config = api.load_config("myspoke", "config.yaml", env_aware=True)

        # Validate credentials for new environment
        is_valid = check_credentials(new_env, config)

        if not is_valid:
            api.send_notification(
                "myspoke",
                "Credentials Required",
                f"Please authenticate for {new_env}",
                level="warning"
            )

    events.on("env_change", on_env_change)

---

Error Handling

All API functions handle errors gracefully and return sensible defaults:

# Returns None on failure (never raises)
env = api.get_active_env()
if env is None:
    print("Daemon not reachable or env not set")

# Returns False on failure
success = api.set_active_env("prod")
if not success:
    print("Failed to set environment")

# Returns empty dict/list on failure
config = api.load_config("myspoke", "config.yaml")  # {} on error
envs = api.list_environments()  # [] on error

Check logs for detailed error information:

tail -f ~/.config/axium/daemon.log

---

See Also

• IPC Commands Reference - Low-level IPC protocol
• Events Reference - Event system documentation
• Gears Guide - Gear permissions and development
• Spokes Guide - Spoke development guide