# my_topics_plugin/plugin.py
from pathlib import Path
from typing import Any, Optional
from kernle.protocols import (
PluginContext,
PluginHealth,
ToolDefinition,
)
class TopicsPlugin:
"""Tracks conversation topics and their frequency."""
@property
def name(self) -> str:
return "topics"
@property
def version(self) -> str:
return "0.1.0"
@property
def protocol_version(self) -> int:
return 1
@property
def description(self) -> str:
return "Track conversation topics"
def capabilities(self) -> list[str]:
return ["topic-tracking", "analytics"]
def activate(self, context: PluginContext) -> None:
self._context = context
self._topics: dict[str, int] = {}
# Restore state from data dir if it exists
state_file = context.get_data_dir() / "topics.json"
if state_file.exists():
import json
self._topics = json.loads(state_file.read_text())
def deactivate(self) -> None:
# Persist state before shutdown
import json
state_file = self._context.get_data_dir() / "topics.json"
state_file.write_text(json.dumps(self._topics))
self._context = None
def register_cli(self, subparsers: Any) -> None:
parser = subparsers.add_parser(
"topics", help="Show tracked topics"
)
parser.set_defaults(func=self._cli_topics)
def _cli_topics(self, args: Any) -> None:
for topic, count in sorted(
self._topics.items(), key=lambda x: x[1], reverse=True
):
print(f" {topic}: {count}")
def register_tools(self) -> list[ToolDefinition]:
return [
ToolDefinition(
name="track_topic",
description="Record a conversation topic",
input_schema={
"type": "object",
"properties": {
"topic": {
"type": "string",
"description": "The topic to track",
}
},
"required": ["topic"],
},
handler=self._handle_track,
),
ToolDefinition(
name="get_topics",
description="Get all tracked topics with counts",
input_schema={"type": "object", "properties": {}},
handler=self._handle_get,
),
]
def _handle_track(self, topic: str) -> dict:
self._topics[topic] = self._topics.get(topic, 0) + 1
# Record as an episode through the context
self._context.episode(
f"Discussed topic: {topic}",
f"Topic '{topic}' tracked ({self._topics[topic]} times)",
tags=["topic-tracking"],
)
return {"tracked": topic, "count": self._topics[topic]}
def _handle_get(self) -> dict:
return {"topics": self._topics}
def on_load(self, load_context: dict[str, Any]) -> None:
if self._topics:
load_context["topics"] = {
"top_topics": sorted(
self._topics.items(),
key=lambda x: x[1],
reverse=True,
)[:5]
}
def on_status(self, status: dict[str, Any]) -> None:
status["topics"] = {
"total_topics": len(self._topics),
"total_mentions": sum(self._topics.values()),
}
def health_check(self) -> PluginHealth:
return PluginHealth(
healthy=True,
message=f"Tracking {len(self._topics)} topics",
)
