forked from Rockachopa/Timmy-time-dashboard
This commit is contained in:
90
src/timmy/tools/_base.py
Normal file
90
src/timmy/tools/_base.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""Base types, shared state, and tracking for the Timmy tool system."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Lazy imports to handle test mocking
|
||||
_ImportError = None
|
||||
try:
|
||||
from agno.tools import Toolkit
|
||||
from agno.tools.file import FileTools
|
||||
from agno.tools.python import PythonTools
|
||||
from agno.tools.shell import ShellTools
|
||||
|
||||
_AGNO_TOOLS_AVAILABLE = True
|
||||
except ImportError as e:
|
||||
_AGNO_TOOLS_AVAILABLE = False
|
||||
_ImportError = e
|
||||
|
||||
# Track tool usage stats
|
||||
_TOOL_USAGE: dict[str, list[dict]] = {}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ToolStats:
|
||||
"""Statistics for a single tool."""
|
||||
|
||||
tool_name: str
|
||||
call_count: int = 0
|
||||
last_used: str | None = None
|
||||
errors: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentTools:
|
||||
"""Tools assigned to an agent."""
|
||||
|
||||
agent_id: str
|
||||
agent_name: str
|
||||
toolkit: "Toolkit"
|
||||
available_tools: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
# Backward-compat alias
|
||||
PersonaTools = AgentTools
|
||||
|
||||
|
||||
def _track_tool_usage(agent_id: str, tool_name: str, success: bool = True) -> None:
|
||||
"""Track tool usage for analytics."""
|
||||
if agent_id not in _TOOL_USAGE:
|
||||
_TOOL_USAGE[agent_id] = []
|
||||
_TOOL_USAGE[agent_id].append(
|
||||
{
|
||||
"tool": tool_name,
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"success": success,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def get_tool_stats(agent_id: str | None = None) -> dict:
|
||||
"""Get tool usage statistics.
|
||||
|
||||
Args:
|
||||
agent_id: Optional agent ID to filter by. If None, returns stats for all agents.
|
||||
|
||||
Returns:
|
||||
Dict with tool usage statistics.
|
||||
"""
|
||||
if agent_id:
|
||||
usage = _TOOL_USAGE.get(agent_id, [])
|
||||
return {
|
||||
"agent_id": agent_id,
|
||||
"total_calls": len(usage),
|
||||
"tools_used": list(set(u["tool"] for u in usage)),
|
||||
"recent_calls": usage[-10:] if usage else [],
|
||||
}
|
||||
|
||||
# Return stats for all agents
|
||||
all_stats = {}
|
||||
for aid, usage in _TOOL_USAGE.items():
|
||||
all_stats[aid] = {
|
||||
"total_calls": len(usage),
|
||||
"tools_used": list(set(u["tool"] for u in usage)),
|
||||
}
|
||||
return all_stats
|
||||
Reference in New Issue
Block a user