Spaces:
Sleeping
Sleeping
File size: 2,281 Bytes
626b033 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
"""
Automatic Usage Tracking - Wraps @mcp.tool() decorator to track all tool calls
"""
SIMPLE_TRACKING_CODE = '''
# ============================================================================
# AUTOMATIC USAGE TRACKING
# ============================================================================
import os
import requests
import time
from datetime import datetime
from functools import wraps
WEBHOOK_URL = os.getenv('MCP_WEBHOOK_URL', '')
DEPLOYMENT_ID = os.getenv('MCP_DEPLOYMENT_ID', 'unknown')
def _send_tracking(tool_name, duration_ms, success, error=None):
"""Send tracking data to webhook endpoint"""
if not WEBHOOK_URL:
return
try:
requests.post(WEBHOOK_URL, json={
'deployment_id': DEPLOYMENT_ID,
'tool_name': tool_name,
'timestamp': datetime.utcnow().isoformat() + 'Z',
'duration_ms': duration_ms,
'success': success,
'error': error
}, timeout=2)
except:
pass # Silent failure (fix Later)
# Save the original mcp.tool decorator
_original_tool_decorator = mcp.tool
def _tracking_tool_decorator(*dec_args, **dec_kwargs):
"""Wraps @mcp.tool() to automatically track all tool calls"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
success = True
error_msg = None
try:
result = func(*args, **kwargs)
return result
except Exception as e:
success = False
error_msg = str(e)
raise
finally:
duration_ms = int((time.time() - start_time) * 1000)
_send_tracking(func.__name__, duration_ms, success, error_msg)
# Apply the original FastMCP decorator to our tracking wrapper
return _original_tool_decorator(*dec_args, **dec_kwargs)(wrapper)
return decorator
# Replace mcp.tool with our tracking version
mcp.tool = _tracking_tool_decorator
if WEBHOOK_URL:
print(f"✅ Tracking enabled: {WEBHOOK_URL}")
print(f"📍 Deployment ID: {DEPLOYMENT_ID}")
else:
print("⚠️ No webhook URL - tracking disabled")
'''
def get_tracking_code():
return SIMPLE_TRACKING_CODE
|