Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions docs/plugins.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# Plugins

mini-agent loads external plugins via the `mini_agent.plugins` [entry point](https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/) group.

## Interface

```python
from mini_agent import MiniAgentPlugin

class MyPlugin(MiniAgentPlugin):
def on_session_start(self, session_id: str): ...
```

| Hook | When |
|------|------|
| `on_agent_init()` | Startup, before CLI loop |
| `on_session_start(session_id)` | New session, `/new`, `/resume` |
| `on_turn_complete(session_id, history, round_usages)` | After each assistant response saved |
| `on_session_end(session_id, history, round_usages)` | Interactive loop exits |

## Creating a Plugin

```toml
[project.entry-points."mini_agent.plugins"]
my-plugin = "my_plugin.plugin:create_plugin"
```

```python
# src/my_plugin/plugin.py
from mini_agent import MiniAgentPlugin

class MyPlugin(MiniAgentPlugin):
def on_session_start(self, session_id: str) -> None:
print(f"Session started: {session_id}")

def create_plugin():
return MyPlugin()
```

```bash
pip install my-plugin
```

## Secrets

Plugins can store API keys in `~/.mini-agent/.env` (loaded automatically). Shell env vars take precedence.

```bash
# ~/.mini-agent/.env
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
```
1 change: 1 addition & 0 deletions docs/quickstart.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,3 +49,4 @@ Summarize this repository
- [Usage](usage.md) - CLI flags, slash commands, and keyboard shortcuts.
- [Providers](providers.md) - authentication and gateway setup.
- [Config](config.md) - model and reasoning effort defaults.
- [Plugins](plugins.md) - extending mini-agent with external plugins.
1 change: 1 addition & 0 deletions docs/usage.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ mini /resume
| `/new` | Start a new session |
| `/resume` | Pick from previous sessions |
| `/copy` | Copy last assistant message to clipboard |
| `/plugins` | List active plugins |
| `/exit`, `q` | Quit |

## Keyboard Shortcuts
Expand Down
8 changes: 7 additions & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,14 @@

sysconfig._CONFIG_VARS["MACOSX_DEPLOYMENT_TARGET"] = "11.0"

EXCLUDED = {"__init__.py"}
EXCLUDED_PATHS = {"plugin/base.py"}

modules = [
str(f) for f in Path("src/mini_agent").rglob("*.py") if f.name != "__init__.py"
str(f)
for f in Path("src/mini_agent").rglob("*.py")
if f.name not in EXCLUDED
and str(f.relative_to("src/mini_agent")) not in EXCLUDED_PATHS
]

setup(
Expand Down
2 changes: 2 additions & 0 deletions src/mini_agent/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
"""Mini-agent: a minimal agent."""

from .cli.main import main

__all__ = ["main"]
Comment on lines +1 to 5

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 MiniAgentPlugin not importable from the top-level package

The plugin docs show from mini_agent import MiniAgentPlugin as the canonical import, but __init__.py only re-exports main — no reference to MiniAgentPlugin exists. Any plugin author following the documented example will get ImportError: cannot import name 'MiniAgentPlugin' from 'mini_agent' at install time.

Suggested change
"""Mini-agent: a minimal agent."""
from .cli.main import main
__all__ = ["main"]
"""Mini-agent: a minimal agent."""
from .cli.main import main
from .plugin.base import MiniAgentPlugin
__all__ = ["main", "MiniAgentPlugin"]

1 change: 1 addition & 0 deletions src/mini_agent/cli/display/completion.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
"/model": "Select a model",
"/copy": "Copy the last assistant response to clipboard",
"/status": "Show current session configuration and token usage",
"/plugins": "List currently active plugins",
"/exit": "exit the session",
}

Expand Down
31 changes: 30 additions & 1 deletion src/mini_agent/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
REASONING_EFFORT_LEVELS,
config,
)
from ..plugin import PluginManager
from .clipboard import copy_last_assistant_text
from .display import (
ACCENT_COLOR,
Expand Down Expand Up @@ -39,18 +40,24 @@
console = Console()

session_manager = SessionManager()
plugin_manager = PluginManager.discover()


def _run_non_interactive(prompt: str) -> None:
history: list[MessageParam] = [{"role": "user", "content": prompt}]
session_id = session_manager.new_id()
plugin_manager.on_session_start(session_id)
history_len = len(history)
agent_loop(history)
if len(history) > history_len and token_tracker.get() is not None:
session_manager.save(session_id, history, token_tracker.round_usages)
plugin_manager.on_turn_complete(session_id, history, token_tracker.round_usages)
Comment on lines 46 to +54

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

In non-interactive mode (_run_non_interactive), the session is initialized and on_session_start is triggered, but on_session_end is never called when the run completes. This causes an inconsistent lifecycle state for plugins that rely on on_session_end to finalize tasks, persist data, or clean up resources.

Suggested change
def _run_non_interactive(prompt: str) -> None:
history: list[MessageParam] = [{"role": "user", "content": prompt}]
session_id = session_manager.new_id()
plugin_manager.on_session_start(session_id)
history_len = len(history)
agent_loop(history)
if len(history) > history_len and token_tracker.get() is not None:
session_manager.save(session_id, history, token_tracker.round_usages)
plugin_manager.on_turn_complete(session_id, history, token_tracker.round_usages)
def _run_non_interactive(prompt: str) -> None:
history: list[MessageParam] = [{"role": "user", "content": prompt}]
session_id = session_manager.new_id()
plugin_manager.on_session_start(session_id)
history_len = len(history)
agent_loop(history)
if len(history) > history_len and token_tracker.get() is not None:
session_manager.save(session_id, history, token_tracker.round_usages)
plugin_manager.on_turn_complete(session_id, history, token_tracker.round_usages)
plugin_manager.on_session_end(session_id, history, token_tracker.round_usages)

Comment on lines 46 to +54

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 on_session_end never fires in non-interactive mode

_run_non_interactive calls on_session_start and on_turn_complete but never on_session_end. Plugins that implement cleanup, flushing, or finalization logic in on_session_end will silently skip it for every mini -p "..." invocation, leaving e.g. buffered telemetry unsent or open resources unclosed.



def _run_interactive(prompt: str | None = None, session_id: str | None = None) -> None:
def _run_interactive(
prompt: str | None = None,
session_id: str | None = None,
) -> None:
print_welcome_banner()
history: list[MessageParam] = []
current_session_id = session_manager.new_id()
Expand All @@ -73,6 +80,8 @@ def _run_interactive(prompt: str | None = None, session_id: str | None = None) -
except StopIteration:
print("Session ID not found.\n")

plugin_manager.on_session_start(current_session_id)

while True:
try:
query = session.prompt(pre_run=pre_run)
Expand All @@ -93,6 +102,7 @@ def _run_interactive(prompt: str | None = None, session_id: str | None = None) -
if command == "/new":
history.clear()
current_session_id = session_manager.new_id()
plugin_manager.on_session_start(current_session_id)
Comment on lines 102 to +105

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

When starting a new session via the /new command in interactive mode, the current active session is discarded without triggering the on_session_end lifecycle hook. This leads to resource leaks or incomplete state tracking in plugins. We should explicitly end the current session before starting the new one.

Suggested change
if command == "/new":
history.clear()
current_session_id = session_manager.new_id()
plugin_manager.on_session_start(current_session_id)
if command == "/new":
plugin_manager.on_session_end(
current_session_id, history, token_tracker.round_usages
)
history.clear()
current_session_id = session_manager.new_id()
plugin_manager.on_session_start(current_session_id)

sent_image_count[0] = 0
next_indicator[0] = 1
token_tracker.reset()
Expand All @@ -104,6 +114,7 @@ def _run_interactive(prompt: str | None = None, session_id: str | None = None) -
current_session_id, history, _ = prompt_resume(
session_manager, current_session_id, history
)
plugin_manager.on_session_start(current_session_id)
Comment on lines 114 to +117

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similarly to the /new command, when resuming a session via /resume, the current active session is replaced without triggering the on_session_end lifecycle hook. We should explicitly end the current session before prompting to resume another one.

Suggested change
current_session_id, history, _ = prompt_resume(
session_manager, current_session_id, history
)
plugin_manager.on_session_start(current_session_id)
plugin_manager.on_session_end(
current_session_id, history, token_tracker.round_usages
)
current_session_id, history, _ = prompt_resume(
session_manager, current_session_id, history
)
plugin_manager.on_session_start(current_session_id)

sent_image_count[0] = count_images_in_history(history)
next_indicator[0] = max_indicator_in_history(history) + 1
attached_images.clear()
Expand All @@ -122,6 +133,15 @@ def _run_interactive(prompt: str | None = None, session_id: str | None = None) -
print()
attached_images.clear()
continue
if command == "/plugins":
plugins = plugin_manager.list_plugins()
if plugins:
print(f"Active plugins: {', '.join(plugins)}")
else:
print("No plugins loaded.")
print()
attached_images.clear()
continue

content = build_user_content(query, attached_images, sent_image_count)

Expand All @@ -135,6 +155,13 @@ def _run_interactive(prompt: str | None = None, session_id: str | None = None) -
session_manager.save(
current_session_id, history, token_tracker.round_usages
)
plugin_manager.on_turn_complete(
current_session_id, history, token_tracker.round_usages
)

plugin_manager.on_session_end(
current_session_id, history, token_tracker.round_usages
)

if session_manager.exists(current_session_id):
usage_report = format_usage_report(token_tracker.get())
Expand Down Expand Up @@ -192,6 +219,8 @@ def main() -> None:
)
args = parser.parse_args()

plugin_manager.on_agent_init()

if args.model:
config.set_session_model(args.model)

Expand Down
4 changes: 4 additions & 0 deletions src/mini_agent/plugin/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
from .base import MiniAgentPlugin
from .manager import PluginManager

__all__ = ["MiniAgentPlugin", "PluginManager"]
26 changes: 26 additions & 0 deletions src/mini_agent/plugin/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
"""Base class for mini-agent plugins.

Zero internal dependencies — safe to import before the rest of
the mini-agent package is fully initialised.
"""

from typing import Any


class MiniAgentPlugin:
"""Override lifecycle methods as needed. All default to no-ops."""

def on_agent_init(self) -> None: ...
def on_session_start(self, session_id: str) -> None: ...
def on_turn_complete(
self,
session_id: str,
history: list[dict[str, Any]],
round_usages: list[Any] | None,
) -> None: ...
def on_session_end(
self,
session_id: str,
history: list[dict[str, Any]],
round_usages: list[Any] | None,
) -> None: ...
57 changes: 57 additions & 0 deletions src/mini_agent/plugin/manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
"""Plugin discovery and lifecycle dispatch."""

import contextlib
import importlib.metadata
import warnings
from dataclasses import dataclass, field
from typing import Any


@dataclass
class PluginManager:
"""Discovers plugins and dispatches lifecycle events."""

plugins: list[Any] = field(default_factory=list)

@staticmethod
def discover() -> PluginManager:
plugins: list[Any] = []
for ep in importlib.metadata.entry_points(group="mini_agent.plugins"):
try:
plugins.append(ep.load()())
except Exception as exc:
warnings.warn(f"Failed to load plugin '{ep.name}': {exc}", stacklevel=2)
return PluginManager(plugins=plugins)

def on_agent_init(self) -> None:
for p in self.plugins:
with contextlib.suppress(Exception):
p.on_agent_init()

def on_session_start(self, session_id: str) -> None:
for p in self.plugins:
with contextlib.suppress(Exception):
p.on_session_start(session_id)

def on_turn_complete(
self,
session_id: str,
history: list[dict[str, Any]],
round_usages: list[Any] | None,
) -> None:
for p in self.plugins:
with contextlib.suppress(Exception):
p.on_turn_complete(session_id, history, round_usages)

def on_session_end(
self,
session_id: str,
history: list[dict[str, Any]],
round_usages: list[Any] | None,
) -> None:
for p in self.plugins:
with contextlib.suppress(Exception):
p.on_session_end(session_id, history, round_usages)
Comment on lines +26 to +54

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Silently suppressing all exceptions (Exception) during plugin lifecycle hook execution makes debugging extremely difficult for plugin developers, as any internal errors (like NameError, TypeError, or AttributeError) will be swallowed without any trace.

Instead of duplicating try-except blocks with silent suppression, we can introduce a centralized _dispatch helper method that uses warnings.warn to report hook execution failures while keeping the application robust.

    def _dispatch(self, hook_name: str, *args: Any, **kwargs: Any) -> None:
        for p in self.plugins:
            hook = getattr(p, hook_name, None)
            if hook is not None:
                try:
                    hook(*args, **kwargs)
                except Exception as exc:
                    warnings.warn(
                        f"Error executing hook '{hook_name}' on plugin '{type(p).__name__}': {exc}",
                        stacklevel=3,
                    )

    def on_agent_init(self) -> None:
        self._dispatch("on_agent_init")

    def on_session_start(self, session_id: str) -> None:
        self._dispatch("on_session_start", session_id)

    def on_turn_complete(
        self,
        session_id: str,
        history: list[dict[str, Any]],
        round_usages: list[Any] | None,
    ) -> None:
        self._dispatch("on_turn_complete", session_id, history, round_usages)

    def on_session_end(
        self,
        session_id: str,
        history: list[dict[str, Any]],
        round_usages: list[Any] | None,
    ) -> None:
        self._dispatch("on_session_end", session_id, history, round_usages)

Comment on lines +26 to +54

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Plugin errors silently swallowed after load — no observable signal

contextlib.suppress(Exception) in every lifecycle dispatch discards all plugin exceptions with no log line, warning, or counter. A plugin that raises on every call (e.g. network error, missing config key) will appear to work fine. At minimum, a warnings.warn (matching the load-failure path) would let operators notice misbehaving plugins without crashing the agent.


def list_plugins(self) -> list[str]:
return [type(p).__name__ for p in self.plugins]
Loading