5a94737516
This major refactoring addresses critical bugs in conversation history management and significantly improves token efficiency through intelligent file embedding: **Key Improvements:** • Fixed conversation history duplication bug by centralizing reconstruction in server.py • Added intelligent file filtering to prevent re-embedding files already in conversation history • Centralized file processing logic in BaseTool._prepare_file_content_for_prompt() • Enhanced log monitoring with better categorization and file embedding visibility • Updated comprehensive test suite to verify new architecture and edge cases **Architecture Changes:** • Removed duplicate conversation history reconstruction from tools/base.py • Conversation history now handled exclusively by server.py:reconstruct_thread_context • All tools now use centralized file processing with automatic deduplication • Improved token efficiency by embedding unique files only once per conversation **Performance Benefits:** • Reduced token usage through smart file filtering • Eliminated redundant file embeddings in continued conversations • Better observability with detailed debug logging for file operations 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
111 lines
4.9 KiB
Python
111 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Log monitor for MCP server - monitors and displays tool activity
|
||
"""
|
||
|
||
import os
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
|
||
def monitor_mcp_activity():
|
||
"""Monitor MCP server activity by watching the log file"""
|
||
log_file = "/tmp/mcp_server.log"
|
||
activity_file = "/tmp/mcp_activity.log"
|
||
debug_file = "/tmp/gemini_debug.log"
|
||
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] MCP Log Monitor started")
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Monitoring: {log_file}")
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Activity file: {activity_file}")
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Debug file: {debug_file}")
|
||
print("-" * 60)
|
||
|
||
# Track file positions
|
||
log_pos = 0
|
||
activity_pos = 0
|
||
debug_pos = 0
|
||
|
||
# Ensure files exist
|
||
Path(log_file).touch()
|
||
Path(activity_file).touch()
|
||
Path(debug_file).touch()
|
||
|
||
while True:
|
||
try:
|
||
# Check activity file (most important for tool calls)
|
||
if os.path.exists(activity_file):
|
||
with open(activity_file) as f:
|
||
f.seek(activity_pos)
|
||
new_lines = f.readlines()
|
||
activity_pos = f.tell()
|
||
|
||
for line in new_lines:
|
||
line = line.strip()
|
||
if line:
|
||
if "TOOL_CALL:" in line:
|
||
tool_info = line.split("TOOL_CALL:")[-1].strip()
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Tool called: {tool_info}")
|
||
elif "TOOL_COMPLETED:" in line:
|
||
tool_name = line.split("TOOL_COMPLETED:")[-1].strip()
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] ✓ Tool completed: {tool_name}")
|
||
elif "CONVERSATION_RESUME:" in line:
|
||
resume_info = line.split("CONVERSATION_RESUME:")[-1].strip()
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Resume: {resume_info}")
|
||
elif "CONVERSATION_CONTEXT:" in line:
|
||
context_info = line.split("CONVERSATION_CONTEXT:")[-1].strip()
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Context: {context_info}")
|
||
elif "CONVERSATION_ERROR:" in line:
|
||
error_info = line.split("CONVERSATION_ERROR:")[-1].strip()
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ Conversation error: {error_info}")
|
||
|
||
# Check main log file for errors and warnings
|
||
if os.path.exists(log_file):
|
||
with open(log_file) as f:
|
||
f.seek(log_pos)
|
||
new_lines = f.readlines()
|
||
log_pos = f.tell()
|
||
|
||
for line in new_lines:
|
||
line = line.strip()
|
||
if line:
|
||
if "ERROR" in line:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ {line}")
|
||
elif "WARNING" in line:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] ⚠️ {line}")
|
||
elif "DEBUG" in line:
|
||
# Highlight file embedding debug logs
|
||
if "📄" in line or "📁" in line:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] 📂 FILE: {line}")
|
||
else:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] 🔍 {line}")
|
||
elif "INFO" in line and ("Gemini API" in line or "Tool" in line or "Conversation" in line):
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] ℹ️ {line}")
|
||
elif "Gemini API" in line and ("Sending" in line or "Received" in line):
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] API: {line}")
|
||
|
||
# Check debug file
|
||
if os.path.exists(debug_file):
|
||
with open(debug_file) as f:
|
||
f.seek(debug_pos)
|
||
new_lines = f.readlines()
|
||
debug_pos = f.tell()
|
||
|
||
for line in new_lines:
|
||
line = line.strip()
|
||
if line:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] DEBUG: {line}")
|
||
|
||
time.sleep(0.5) # Check every 500ms
|
||
|
||
except KeyboardInterrupt:
|
||
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Log monitor stopped")
|
||
break
|
||
except Exception as e:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] Monitor error: {e}")
|
||
time.sleep(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
monitor_mcp_activity()
|