mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 805af9a29c |
@@ -0,0 +1,254 @@
|
||||
"""Tool call history tracking for MCP operations."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from collections import deque
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any, Callable, Deque, Dict, List, Optional, TypeVar
|
||||
from functools import wraps
|
||||
|
||||
# Type variable for generic function type
|
||||
F = TypeVar("F", bound=Callable[..., Any])
|
||||
|
||||
|
||||
@dataclass
|
||||
class ToolCall:
|
||||
"""Represents a single tool call in history."""
|
||||
|
||||
id: str
|
||||
timestamp: float
|
||||
tool_name: str
|
||||
status: str # "success", "error", "running"
|
||||
execution_time_ms: Optional[float] = None
|
||||
input: Optional[Dict[str, Any]] = None
|
||||
output: Optional[Any] = None
|
||||
error: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Convert to dictionary for JSON serialization."""
|
||||
result = asdict(self)
|
||||
# Convert timestamp to ISO format
|
||||
result["timestamp"] = datetime.fromtimestamp(self.timestamp).isoformat() + "Z"
|
||||
return result
|
||||
|
||||
|
||||
class ToolHistoryTracker:
|
||||
"""Singleton tracker for tool call history."""
|
||||
|
||||
_instance: Optional["ToolHistoryTracker"] = None
|
||||
_lock = asyncio.Lock()
|
||||
|
||||
def __new__(cls) -> "ToolHistoryTracker":
|
||||
"""Ensure singleton instance."""
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
cls._instance._initialized = False
|
||||
return cls._instance
|
||||
|
||||
def __init__(self, max_history: int = 1000):
|
||||
"""Initialize the tracker with a maximum history size."""
|
||||
if self._initialized:
|
||||
return
|
||||
|
||||
self.max_history = max_history
|
||||
self.history: Deque[ToolCall] = deque(maxlen=max_history)
|
||||
self._call_counter = 0
|
||||
self._lock = asyncio.Lock()
|
||||
self._initialized = True
|
||||
|
||||
async def add_call(
|
||||
self,
|
||||
tool_name: str,
|
||||
input_params: Optional[Dict[str, Any]] = None,
|
||||
) -> str:
|
||||
"""Add a new tool call to history and return its ID."""
|
||||
async with self._lock:
|
||||
self._call_counter += 1
|
||||
call_id = f"call_{self._call_counter:06d}_{int(time.time() * 1000)}"
|
||||
|
||||
tool_call = ToolCall(
|
||||
id=call_id,
|
||||
timestamp=time.time(),
|
||||
tool_name=tool_name,
|
||||
status="running",
|
||||
input=input_params,
|
||||
)
|
||||
|
||||
self.history.append(tool_call)
|
||||
return call_id
|
||||
|
||||
async def update_call(
|
||||
self,
|
||||
call_id: str,
|
||||
status: str,
|
||||
execution_time_ms: Optional[float] = None,
|
||||
output: Optional[Any] = None,
|
||||
error: Optional[str] = None,
|
||||
):
|
||||
"""Update an existing tool call with results."""
|
||||
async with self._lock:
|
||||
for call in self.history:
|
||||
if call.id == call_id:
|
||||
call.status = status
|
||||
call.execution_time_ms = execution_time_ms
|
||||
call.output = output
|
||||
call.error = error
|
||||
break
|
||||
|
||||
async def get_history(
|
||||
self,
|
||||
limit: int = 10,
|
||||
tool_name: Optional[str] = None,
|
||||
include_inputs: bool = True,
|
||||
include_outputs: bool = False,
|
||||
since: Optional[str] = None,
|
||||
) -> List[ToolCall]:
|
||||
"""Get filtered tool call history."""
|
||||
async with self._lock:
|
||||
# Convert to list for filtering
|
||||
calls = list(self.history)
|
||||
|
||||
# Filter by time if specified
|
||||
if since:
|
||||
cutoff_time = self._parse_time_filter(since)
|
||||
calls = [c for c in calls if c.timestamp >= cutoff_time]
|
||||
|
||||
# Filter by tool name if specified
|
||||
if tool_name:
|
||||
calls = [c for c in calls if c.tool_name == tool_name]
|
||||
|
||||
# Sort by timestamp (most recent first)
|
||||
calls.sort(key=lambda x: x.timestamp, reverse=True)
|
||||
|
||||
# Limit results
|
||||
calls = calls[:limit]
|
||||
|
||||
# Process outputs based on flags
|
||||
result = []
|
||||
for call in calls:
|
||||
call_copy = ToolCall(
|
||||
id=call.id,
|
||||
timestamp=call.timestamp,
|
||||
tool_name=call.tool_name,
|
||||
status=call.status,
|
||||
execution_time_ms=call.execution_time_ms,
|
||||
input=call.input if include_inputs else None,
|
||||
output=call.output if include_outputs else None,
|
||||
error=call.error,
|
||||
)
|
||||
result.append(call_copy)
|
||||
|
||||
return result
|
||||
|
||||
def _parse_time_filter(self, since: str) -> float:
|
||||
"""Parse time filter string and return timestamp."""
|
||||
now = time.time()
|
||||
|
||||
# Handle relative times like "1h ago", "2d ago"
|
||||
if "ago" in since.lower():
|
||||
parts = since.lower().replace("ago", "").strip().split()
|
||||
if len(parts) == 1:
|
||||
value_unit = parts[0]
|
||||
# Try to parse combined format like "1h"
|
||||
import re
|
||||
match = re.match(r"(\d+)([hdmw])", value_unit)
|
||||
if match:
|
||||
value = int(match.group(1))
|
||||
unit = match.group(2)
|
||||
else:
|
||||
return now
|
||||
elif len(parts) == 2:
|
||||
value = int(parts[0])
|
||||
unit = parts[1][0] # First letter of unit
|
||||
else:
|
||||
return now
|
||||
|
||||
if unit == "h": # hours
|
||||
return now - (value * 3600)
|
||||
elif unit == "d": # days
|
||||
return now - (value * 86400)
|
||||
elif unit == "m": # minutes
|
||||
return now - (value * 60)
|
||||
elif unit == "w": # weeks
|
||||
return now - (value * 604800)
|
||||
|
||||
# Handle absolute dates like "2024-01-20"
|
||||
try:
|
||||
dt = datetime.fromisoformat(since)
|
||||
return dt.timestamp()
|
||||
except:
|
||||
pass
|
||||
|
||||
# Default to current time if parsing fails
|
||||
return now
|
||||
|
||||
async def clear_history(self):
|
||||
"""Clear all tool call history."""
|
||||
async with self._lock:
|
||||
self.history.clear()
|
||||
self._call_counter = 0
|
||||
|
||||
async def get_call_by_id(self, call_id: str) -> Optional[ToolCall]:
|
||||
"""Get a specific tool call by ID."""
|
||||
async with self._lock:
|
||||
for call in self.history:
|
||||
if call.id == call_id:
|
||||
return call
|
||||
return None
|
||||
|
||||
|
||||
def track_tool_call(func: F) -> F:
|
||||
"""Decorator to track MCP tool calls."""
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
"""Wrapper that tracks tool execution."""
|
||||
tracker = ToolHistoryTracker()
|
||||
|
||||
# Extract tool name from function
|
||||
tool_name = func.__name__
|
||||
|
||||
# Filter out Context parameter if present
|
||||
filtered_kwargs = {k: v for k, v in kwargs.items() if k != "context"}
|
||||
|
||||
# Record the call
|
||||
call_id = await tracker.add_call(tool_name, filtered_kwargs)
|
||||
|
||||
# Execute the tool
|
||||
start_time = time.time()
|
||||
try:
|
||||
result = await func(*args, **kwargs)
|
||||
execution_time = (time.time() - start_time) * 1000
|
||||
|
||||
# Update with success
|
||||
await tracker.update_call(
|
||||
call_id,
|
||||
status="success",
|
||||
execution_time_ms=execution_time,
|
||||
output=result if isinstance(result, (str, int, float, bool)) else str(result)[:500],
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
execution_time = (time.time() - start_time) * 1000
|
||||
|
||||
# Update with error
|
||||
await tracker.update_call(
|
||||
call_id,
|
||||
status="error",
|
||||
execution_time_ms=execution_time,
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
raise
|
||||
|
||||
return wrapper # type: ignore
|
||||
|
||||
|
||||
# Singleton instance getter for convenience
|
||||
def get_tracker() -> ToolHistoryTracker:
|
||||
"""Get the singleton ToolHistoryTracker instance."""
|
||||
return ToolHistoryTracker()
|
||||
@@ -26,15 +26,19 @@ from basic_memory.mcp.tools.project_management import (
|
||||
)
|
||||
# ChatGPT-compatible tools
|
||||
from basic_memory.mcp.tools.chatgpt_tools import search, fetch
|
||||
# Tool history tracking
|
||||
from basic_memory.mcp.tools.tool_history import tool_history, get_tool_call, clear_tool_history
|
||||
|
||||
__all__ = [
|
||||
"build_context",
|
||||
"canvas",
|
||||
"clear_tool_history",
|
||||
"create_memory_project",
|
||||
"delete_note",
|
||||
"delete_project",
|
||||
"edit_note",
|
||||
"fetch",
|
||||
"get_tool_call",
|
||||
"list_directory",
|
||||
"list_memory_projects",
|
||||
"move_note",
|
||||
@@ -44,6 +48,7 @@ __all__ = [
|
||||
"search",
|
||||
"search_notes",
|
||||
"sync_status",
|
||||
"tool_history",
|
||||
"view_note",
|
||||
"write_note",
|
||||
]
|
||||
|
||||
@@ -9,6 +9,7 @@ from fastmcp import Context
|
||||
from basic_memory.mcp.async_client import client
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.tool_history import track_tool_call
|
||||
from basic_memory.mcp.tools.search import search_notes
|
||||
from basic_memory.mcp.tools.utils import call_get
|
||||
from basic_memory.schemas.memory import memory_url_path
|
||||
@@ -18,6 +19,7 @@ from basic_memory.utils import validate_project_path
|
||||
@mcp.tool(
|
||||
description="Read a markdown note by title or permalink.",
|
||||
)
|
||||
@track_tool_call
|
||||
async def read_note(
|
||||
identifier: str,
|
||||
project: Optional[str] = None,
|
||||
|
||||
@@ -9,6 +9,7 @@ from fastmcp import Context
|
||||
from basic_memory.mcp.async_client import client
|
||||
from basic_memory.mcp.project_context import get_active_project
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.tool_history import track_tool_call
|
||||
from basic_memory.mcp.tools.utils import call_post
|
||||
from basic_memory.schemas.search import SearchItemType, SearchQuery, SearchResponse
|
||||
|
||||
@@ -199,6 +200,7 @@ Error searching for '{query}': {error_message}
|
||||
@mcp.tool(
|
||||
description="Search across all content in the knowledge base with advanced syntax support.",
|
||||
)
|
||||
@track_tool_call
|
||||
async def search_notes(
|
||||
query: str,
|
||||
project: Optional[str] = None,
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
"""Tool history MCP tool for Basic Memory."""
|
||||
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
from fastmcp import Context
|
||||
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.tool_history import ToolHistoryTracker, get_tracker
|
||||
|
||||
|
||||
@mcp.tool(
|
||||
description="Get the history of previous MCP tool calls. Useful for debugging, auditing workflows, and referencing previous operations."
|
||||
)
|
||||
async def tool_history(
|
||||
limit: int = 10,
|
||||
tool_name: Optional[str] = None,
|
||||
include_inputs: bool = True,
|
||||
include_outputs: bool = False,
|
||||
since: Optional[str] = None,
|
||||
context: Context | None = None,
|
||||
) -> str:
|
||||
"""Get the history of previous MCP tool calls.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of tool calls to return (default: 10)
|
||||
tool_name: Filter by specific tool name (e.g., "write_note", "search")
|
||||
include_inputs: Include input parameters in response (default: True)
|
||||
include_outputs: Include output/results in response (default: False)
|
||||
since: Filter calls since a specific time (e.g., "1h ago", "30m ago", "2024-01-20")
|
||||
context: FastMCP context (automatically provided)
|
||||
|
||||
Returns:
|
||||
A formatted string containing tool call history information.
|
||||
|
||||
Examples:
|
||||
- Get last 5 tool calls: tool_history(limit=5)
|
||||
- Get all write_note calls from last hour: tool_history(tool_name="write_note", since="1h ago")
|
||||
- Get recent searches with outputs: tool_history(tool_name="search", include_outputs=True)
|
||||
"""
|
||||
tracker = get_tracker()
|
||||
|
||||
# Get filtered history
|
||||
calls = await tracker.get_history(
|
||||
limit=limit,
|
||||
tool_name=tool_name,
|
||||
include_inputs=include_inputs,
|
||||
include_outputs=include_outputs,
|
||||
since=since,
|
||||
)
|
||||
|
||||
if not calls:
|
||||
return "No tool calls found matching the specified criteria."
|
||||
|
||||
# Format the response
|
||||
result_lines = [f"## Tool Call History ({len(calls)} calls)\n"]
|
||||
|
||||
if since:
|
||||
result_lines.append(f"**Filter:** Since {since}\n")
|
||||
if tool_name:
|
||||
result_lines.append(f"**Tool:** {tool_name}\n")
|
||||
|
||||
result_lines.append("") # Empty line for spacing
|
||||
|
||||
for call in calls:
|
||||
# Format timestamp
|
||||
call_dict = call.to_dict()
|
||||
timestamp = call_dict["timestamp"]
|
||||
|
||||
result_lines.append(f"### {call.tool_name} - {call.id}")
|
||||
result_lines.append(f"**Time:** {timestamp}")
|
||||
result_lines.append(f"**Status:** {call.status}")
|
||||
|
||||
if call.execution_time_ms is not None:
|
||||
result_lines.append(f"**Duration:** {call.execution_time_ms:.2f}ms")
|
||||
|
||||
if include_inputs and call.input:
|
||||
result_lines.append("\n**Input Parameters:**")
|
||||
result_lines.append("```json")
|
||||
result_lines.append(json.dumps(call.input, indent=2))
|
||||
result_lines.append("```")
|
||||
|
||||
if include_outputs and call.output:
|
||||
result_lines.append("\n**Output:**")
|
||||
if isinstance(call.output, str):
|
||||
# For string outputs, show as-is (truncated if needed)
|
||||
output_str = call.output[:1000] if len(call.output) > 1000 else call.output
|
||||
if len(call.output) > 1000:
|
||||
output_str += "\n... (truncated)"
|
||||
result_lines.append(f"```\n{output_str}\n```")
|
||||
else:
|
||||
# For other types, JSON format
|
||||
result_lines.append("```json")
|
||||
result_lines.append(json.dumps(call.output, indent=2))
|
||||
result_lines.append("```")
|
||||
|
||||
if call.error:
|
||||
result_lines.append(f"\n**Error:** {call.error}")
|
||||
|
||||
result_lines.append("") # Empty line between calls
|
||||
|
||||
# Add summary statistics
|
||||
total_count = len(calls)
|
||||
success_count = len([c for c in calls if c.status == "success"])
|
||||
error_count = len([c for c in calls if c.status == "error"])
|
||||
|
||||
result_lines.append("---")
|
||||
result_lines.append(f"**Summary:** {total_count} calls ({success_count} successful, {error_count} errors)")
|
||||
|
||||
return "\n".join(result_lines)
|
||||
|
||||
|
||||
@mcp.tool(
|
||||
description="Get detailed information about a specific tool call by its ID."
|
||||
)
|
||||
async def get_tool_call(
|
||||
call_id: str,
|
||||
context: Context | None = None,
|
||||
) -> str:
|
||||
"""Get detailed information about a specific tool call.
|
||||
|
||||
Args:
|
||||
call_id: The ID of the tool call to retrieve (e.g., "call_000001_1234567890")
|
||||
context: FastMCP context (automatically provided)
|
||||
|
||||
Returns:
|
||||
Detailed information about the specified tool call, or error message if not found.
|
||||
"""
|
||||
tracker = get_tracker()
|
||||
call = await tracker.get_call_by_id(call_id)
|
||||
|
||||
if not call:
|
||||
return f"Tool call with ID '{call_id}' not found in history."
|
||||
|
||||
# Format detailed response
|
||||
call_dict = call.to_dict()
|
||||
timestamp = call_dict["timestamp"]
|
||||
|
||||
result_lines = [
|
||||
f"## Tool Call Details: {call.id}\n",
|
||||
f"**Tool:** {call.tool_name}",
|
||||
f"**Time:** {timestamp}",
|
||||
f"**Status:** {call.status}",
|
||||
]
|
||||
|
||||
if call.execution_time_ms is not None:
|
||||
result_lines.append(f"**Duration:** {call.execution_time_ms:.2f}ms")
|
||||
|
||||
if call.input:
|
||||
result_lines.append("\n### Input Parameters")
|
||||
result_lines.append("```json")
|
||||
result_lines.append(json.dumps(call.input, indent=2))
|
||||
result_lines.append("```")
|
||||
|
||||
if call.output:
|
||||
result_lines.append("\n### Output")
|
||||
if isinstance(call.output, str):
|
||||
result_lines.append(f"```\n{call.output}\n```")
|
||||
else:
|
||||
result_lines.append("```json")
|
||||
result_lines.append(json.dumps(call.output, indent=2))
|
||||
result_lines.append("```")
|
||||
|
||||
if call.error:
|
||||
result_lines.append(f"\n### Error\n```\n{call.error}\n```")
|
||||
|
||||
return "\n".join(result_lines)
|
||||
|
||||
|
||||
@mcp.tool(
|
||||
description="Clear all tool call history. Use with caution as this cannot be undone."
|
||||
)
|
||||
async def clear_tool_history(
|
||||
context: Context | None = None,
|
||||
) -> str:
|
||||
"""Clear all tool call history.
|
||||
|
||||
Args:
|
||||
context: FastMCP context (automatically provided)
|
||||
|
||||
Returns:
|
||||
Confirmation message.
|
||||
"""
|
||||
tracker = get_tracker()
|
||||
await tracker.clear_history()
|
||||
return "Tool call history has been cleared."
|
||||
@@ -7,6 +7,7 @@ from loguru import logger
|
||||
from basic_memory.mcp.async_client import client
|
||||
from basic_memory.mcp.project_context import get_active_project, add_project_metadata
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.tool_history import track_tool_call
|
||||
from basic_memory.mcp.tools.utils import call_put
|
||||
from basic_memory.schemas import EntityResponse
|
||||
from fastmcp import Context
|
||||
@@ -23,6 +24,7 @@ TagType = Union[List[str], str, None]
|
||||
@mcp.tool(
|
||||
description="Create or update a markdown note. Returns a markdown formatted summary of the semantic content.",
|
||||
)
|
||||
@track_tool_call
|
||||
async def write_note(
|
||||
title: str,
|
||||
content: str,
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
"""Tests for tool call history tracking."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import time
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from basic_memory.mcp.tool_history import (
|
||||
ToolCall,
|
||||
ToolHistoryTracker,
|
||||
track_tool_call,
|
||||
get_tracker,
|
||||
)
|
||||
from basic_memory.mcp.tools.tool_history import (
|
||||
tool_history,
|
||||
get_tool_call,
|
||||
clear_tool_history,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def tracker():
|
||||
"""Get a fresh tracker instance for testing."""
|
||||
tracker = ToolHistoryTracker()
|
||||
await tracker.clear_history()
|
||||
return tracker
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_singleton_tracker():
|
||||
"""Test that ToolHistoryTracker is a singleton."""
|
||||
tracker1 = ToolHistoryTracker()
|
||||
tracker2 = ToolHistoryTracker()
|
||||
assert tracker1 is tracker2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_tracker():
|
||||
"""Test the get_tracker convenience function."""
|
||||
tracker1 = get_tracker()
|
||||
tracker2 = get_tracker()
|
||||
assert tracker1 is tracker2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_and_update_call(tracker):
|
||||
"""Test adding and updating a tool call."""
|
||||
# Add a call
|
||||
call_id = await tracker.add_call(
|
||||
tool_name="test_tool",
|
||||
input_params={"param1": "value1", "param2": 42}
|
||||
)
|
||||
|
||||
assert call_id.startswith("call_")
|
||||
|
||||
# Get the call
|
||||
calls = await tracker.get_history(limit=1)
|
||||
assert len(calls) == 1
|
||||
assert calls[0].tool_name == "test_tool"
|
||||
assert calls[0].status == "running"
|
||||
assert calls[0].input == {"param1": "value1", "param2": 42}
|
||||
|
||||
# Update the call
|
||||
await tracker.update_call(
|
||||
call_id=call_id,
|
||||
status="success",
|
||||
execution_time_ms=123.45,
|
||||
output="Result data"
|
||||
)
|
||||
|
||||
# Check update
|
||||
updated_call = await tracker.get_call_by_id(call_id)
|
||||
assert updated_call is not None
|
||||
assert updated_call.status == "success"
|
||||
assert updated_call.execution_time_ms == 123.45
|
||||
assert updated_call.output == "Result data"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_track_tool_call_decorator():
|
||||
"""Test the track_tool_call decorator."""
|
||||
|
||||
@track_tool_call
|
||||
async def sample_tool(param1: str, param2: int) -> str:
|
||||
"""A sample tool for testing."""
|
||||
await asyncio.sleep(0.01) # Simulate some work
|
||||
return f"Processed {param1} with {param2}"
|
||||
|
||||
# Clear history first
|
||||
tracker = get_tracker()
|
||||
await tracker.clear_history()
|
||||
|
||||
# Call the decorated function
|
||||
result = await sample_tool(param1="test", param2=123)
|
||||
assert result == "Processed test with 123"
|
||||
|
||||
# Check that it was tracked
|
||||
calls = await tracker.get_history(limit=1)
|
||||
assert len(calls) == 1
|
||||
assert calls[0].tool_name == "sample_tool"
|
||||
assert calls[0].status == "success"
|
||||
assert calls[0].input == {"param1": "test", "param2": 123}
|
||||
assert "Processed test with 123" in str(calls[0].output)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_track_tool_call_with_error():
|
||||
"""Test the track_tool_call decorator with an error."""
|
||||
|
||||
@track_tool_call
|
||||
async def failing_tool(param1: str) -> str:
|
||||
"""A tool that fails."""
|
||||
raise ValueError("Test error")
|
||||
|
||||
tracker = get_tracker()
|
||||
await tracker.clear_history()
|
||||
|
||||
# Call the decorated function and expect error
|
||||
with pytest.raises(ValueError, match="Test error"):
|
||||
await failing_tool(param1="test")
|
||||
|
||||
# Check that error was tracked
|
||||
calls = await tracker.get_history(limit=1)
|
||||
assert len(calls) == 1
|
||||
assert calls[0].tool_name == "failing_tool"
|
||||
assert calls[0].status == "error"
|
||||
assert calls[0].error == "Test error"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_filter_by_tool_name(tracker):
|
||||
"""Test filtering history by tool name."""
|
||||
# Add calls for different tools
|
||||
await tracker.add_call("tool_a", {"param": 1})
|
||||
await tracker.add_call("tool_b", {"param": 2})
|
||||
await tracker.add_call("tool_a", {"param": 3})
|
||||
|
||||
# Filter by tool name
|
||||
calls_a = await tracker.get_history(tool_name="tool_a")
|
||||
assert len(calls_a) == 2
|
||||
assert all(call.tool_name == "tool_a" for call in calls_a)
|
||||
|
||||
calls_b = await tracker.get_history(tool_name="tool_b")
|
||||
assert len(calls_b) == 1
|
||||
assert calls_b[0].tool_name == "tool_b"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_filter_by_time(tracker):
|
||||
"""Test filtering history by time."""
|
||||
# Add an old call
|
||||
old_call_id = await tracker.add_call("old_tool", {"param": "old"})
|
||||
|
||||
# Update timestamp to be 2 hours ago
|
||||
for call in tracker.history:
|
||||
if call.id == old_call_id:
|
||||
call.timestamp = time.time() - 7200 # 2 hours ago
|
||||
break
|
||||
|
||||
# Add a recent call
|
||||
await tracker.add_call("recent_tool", {"param": "recent"})
|
||||
|
||||
# Filter by time
|
||||
recent_calls = await tracker.get_history(since="1h ago")
|
||||
assert len(recent_calls) == 1
|
||||
assert recent_calls[0].tool_name == "recent_tool"
|
||||
|
||||
all_calls = await tracker.get_history(since="3h ago")
|
||||
assert len(all_calls) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_history_limit(tracker):
|
||||
"""Test that history respects max size limit."""
|
||||
# Create a tracker with small limit
|
||||
small_tracker = ToolHistoryTracker()
|
||||
small_tracker.max_history = 3
|
||||
await small_tracker.clear_history()
|
||||
|
||||
# Add more calls than the limit
|
||||
for i in range(5):
|
||||
await small_tracker.add_call(f"tool_{i}", {"index": i})
|
||||
|
||||
# Check that only the last 3 are kept
|
||||
calls = await small_tracker.get_history(limit=10)
|
||||
assert len(calls) == 3
|
||||
# Most recent should be tool_4
|
||||
assert calls[0].tool_name == "tool_4"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_history_mcp_tool():
|
||||
"""Test the tool_history MCP tool."""
|
||||
tracker = get_tracker()
|
||||
await tracker.clear_history()
|
||||
|
||||
# Add some test calls
|
||||
call_id1 = await tracker.add_call("write_note", {"title": "Test", "content": "Hello"})
|
||||
await tracker.update_call(call_id1, "success", 100.5, "Note created")
|
||||
|
||||
call_id2 = await tracker.add_call("search", {"query": "test"})
|
||||
await tracker.update_call(call_id2, "error", 50.2, error="Search failed")
|
||||
|
||||
# Test tool_history function
|
||||
result = await tool_history(limit=5, include_inputs=True, include_outputs=True)
|
||||
|
||||
assert "Tool Call History" in result
|
||||
assert "write_note" in result
|
||||
assert "search" in result
|
||||
assert "Test" in result # Input parameter
|
||||
assert "Note created" in result # Output
|
||||
assert "Search failed" in result # Error
|
||||
assert "2 calls (1 successful, 1 errors)" in result # Summary
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_tool_call_mcp_tool():
|
||||
"""Test the get_tool_call MCP tool."""
|
||||
tracker = get_tracker()
|
||||
await tracker.clear_history()
|
||||
|
||||
# Add a test call
|
||||
call_id = await tracker.add_call("test_tool", {"param": "value"})
|
||||
await tracker.update_call(call_id, "success", 75.3, "Test output")
|
||||
|
||||
# Get the specific call
|
||||
result = await get_tool_call(call_id)
|
||||
|
||||
assert f"Tool Call Details: {call_id}" in result
|
||||
assert "test_tool" in result
|
||||
assert "75.30ms" in result
|
||||
assert '"param": "value"' in result
|
||||
assert "Test output" in result
|
||||
|
||||
# Test non-existent call
|
||||
result = await get_tool_call("non_existent_id")
|
||||
assert "not found" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clear_tool_history_mcp_tool():
|
||||
"""Test the clear_tool_history MCP tool."""
|
||||
tracker = get_tracker()
|
||||
|
||||
# Add some calls
|
||||
await tracker.add_call("tool1", {})
|
||||
await tracker.add_call("tool2", {})
|
||||
|
||||
# Verify calls exist
|
||||
calls = await tracker.get_history()
|
||||
assert len(calls) > 0
|
||||
|
||||
# Clear history
|
||||
result = await clear_tool_history()
|
||||
assert "cleared" in result.lower()
|
||||
|
||||
# Verify history is empty
|
||||
calls = await tracker.get_history()
|
||||
assert len(calls) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_time_filter():
|
||||
"""Test various time filter formats."""
|
||||
tracker = ToolHistoryTracker()
|
||||
|
||||
# Test relative time formats
|
||||
now = time.time()
|
||||
|
||||
# Hours ago
|
||||
result = tracker._parse_time_filter("1h ago")
|
||||
assert abs(result - (now - 3600)) < 1
|
||||
|
||||
result = tracker._parse_time_filter("2h ago")
|
||||
assert abs(result - (now - 7200)) < 1
|
||||
|
||||
# Days ago
|
||||
result = tracker._parse_time_filter("1d ago")
|
||||
assert abs(result - (now - 86400)) < 1
|
||||
|
||||
# Minutes ago
|
||||
result = tracker._parse_time_filter("30m ago")
|
||||
assert abs(result - (now - 1800)) < 1
|
||||
|
||||
# Weeks ago
|
||||
result = tracker._parse_time_filter("1w ago")
|
||||
assert abs(result - (now - 604800)) < 1
|
||||
|
||||
# Invalid format should return current time
|
||||
result = tracker._parse_time_filter("invalid")
|
||||
assert abs(result - now) < 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_call_to_dict():
|
||||
"""Test ToolCall.to_dict method."""
|
||||
tool_call = ToolCall(
|
||||
id="test_id",
|
||||
timestamp=1700000000.0,
|
||||
tool_name="test_tool",
|
||||
status="success",
|
||||
execution_time_ms=123.45,
|
||||
input={"param": "value"},
|
||||
output="result",
|
||||
error=None,
|
||||
)
|
||||
|
||||
result = tool_call.to_dict()
|
||||
|
||||
assert result["id"] == "test_id"
|
||||
assert result["tool_name"] == "test_tool"
|
||||
assert result["status"] == "success"
|
||||
assert result["execution_time_ms"] == 123.45
|
||||
assert result["input"] == {"param": "value"}
|
||||
assert result["output"] == "result"
|
||||
assert result["error"] is None
|
||||
# Check timestamp is converted to ISO format
|
||||
assert "T" in result["timestamp"]
|
||||
assert result["timestamp"].endswith("Z")
|
||||
Reference in New Issue
Block a user