FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

aivorynet/agent-python: AIVory Monitor Python Agent - Runtime exception monitoring via sys.settrace · GitHub

/ agent-python Public
BranchesTags

Repository files navigation

AIVory Monitor Python Agent

Python agent using sys.settrace for breakpoint support and sys.excepthook for exception capture.

Requirements

  • Python 3.8+
  • pip or pip3

Installation

pip install aivory-monitor

Usage

Option 1: Import and Initialize

import aivory_monitor

# Initialize with environment variables
aivory_monitor.init()

# Or pass configuration directly
aivory_monitor.init(
    api_key='your-api-key',
    environment='production'
)

# Your application code

Option 2: Automatic Initialization via Environment

Set environment variables before importing your application:

export AIVORY_API_KEY=your_api_key
python app.py

Then initialize in your application entry point:

import aivory_monitor
aivory_monitor.init()

# Rest of your application

Option 3: Framework Middleware

Django:

Add to your settings.py:

MIDDLEWARE = [
    'aivory_monitor.integrations.django.DjangoIntegration',
    # ... other middleware
]

# Initialize agent
import aivory_monitor
aivory_monitor.init(api_key='your-api-key')

Flask:

from flask import Flask
from aivory_monitor.integrations.flask import init_app
import aivory_monitor

app = Flask(__name__)

# Initialize agent
aivory_monitor.init(api_key='your-api-key')

# Add Flask integration
init_app(app)

FastAPI:

from fastapi import FastAPI
from aivory_monitor.integrations.fastapi import init_app
import aivory_monitor

app = FastAPI()

# Initialize agent
aivory_monitor.init(api_key='your-api-key')

# Add FastAPI integration
init_app(app)

Manual Exception Capture

import aivory_monitor

try:
    risky_operation()
except Exception as e:
    # Capture exception with additional context
    aivory_monitor.capture_exception(e, context={
        'user_id': '12345',
        'operation': 'payment_processing'
    })
    raise

Setting Context and User Information

import aivory_monitor

# Set custom context (included in all captures)
aivory_monitor.set_context({
    'feature_flags': {'new_ui': True},
    'tenant_id': 'acme-corp'
})

# Set user information
aivory_monitor.set_user(
    user_id='user-123',
    email='user@example.com',
    username='john_doe'
)

Configuration

All configuration options can be set via environment variables or passed to init():

Parameter Environment Variable Default Description
api_key AIVORY_API_KEY Required AIVory API key for authentication
backend_url AIVORY_BACKEND_URL wss://api.aivory.net/ws/agent Backend WebSocket URL
environment AIVORY_ENVIRONMENT production Environment name (production, staging, etc.)
sampling_rate AIVORY_SAMPLING_RATE 1.0 Exception sampling rate (0.0 - 1.0)
max_capture_depth AIVORY_MAX_DEPTH 10 Maximum depth for variable capture
max_string_length AIVORY_MAX_STRING_LENGTH 1000 Maximum string length to capture
max_collection_size AIVORY_MAX_COLLECTION_SIZE 100 Maximum collection elements to capture
enable_breakpoints AIVORY_ENABLE_BREAKPOINTS true Enable non-breaking breakpoint support
debug AIVORY_DEBUG false Enable debug logging

Configuration Examples

Environment Variables:

export AIVORY_API_KEY=your_api_key
export AIVORY_BACKEND_URL=wss://api.aivory.net/ws/agent
export AIVORY_ENVIRONMENT=production
export AIVORY_SAMPLING_RATE=0.5
export AIVORY_MAX_DEPTH=5
export AIVORY_DEBUG=false

Programmatic:

import aivory_monitor

aivory_monitor.init(
    api_key='your-api-key',
    backend_url='wss://api.aivory.net/ws/agent',
    environment='production',
    sampling_rate=0.5,
    max_capture_depth=5,
    max_string_length=500,
    max_collection_size=50,
    enable_breakpoints=True,
    debug=False
)

Building from Source

cd monitor-agents/agent-python
pip install -e .

# Or build distribution packages
pip install build
python -m build

How It Works

  1. sys.excepthook: Automatically captures uncaught exceptions with full stack traces and local variables
  2. sys.settrace: Implements non-breaking breakpoints by hooking into Python's trace mechanism
  3. Asyncio Integration: Uses WebSocket client with asyncio for real-time communication with backend
  4. Context Preservation: Captures thread-local and request context at the time of exception

Key Features:

  • Non-breaking breakpoints that don't pause execution
  • Full stack trace with local variables at each frame
  • Request context correlation for web frameworks
  • Configurable variable capture depth and sampling
  • Minimal performance overhead (uses sampling and conditional capture)

Framework Support

Django

The Django integration provides:

  • Automatic request context capture (method, path, headers, user)
  • Exception handling in views and middleware
  • Optional logging handler integration
# settings.py
MIDDLEWARE = [
    'aivory_monitor.integrations.django.DjangoIntegration',
    # ... other middleware
]

import aivory_monitor
aivory_monitor.init(api_key='your-api-key')

# Optional: Configure logging integration
from aivory_monitor.integrations.django import configure_django_logging
LOGGING = configure_django_logging()

Flask

The Flask integration provides:

  • Before-request context setup
  • Automatic exception handling
  • Request timing information
from flask import Flask
from aivory_monitor.integrations.flask import init_app
import aivory_monitor

app = Flask(__name__)
aivory_monitor.init(api_key='your-api-key')
init_app(app)

FastAPI

The FastAPI integration provides:

  • ASGI middleware for request/response tracking
  • Async exception handling
  • Path and query parameter capture
from fastapi import FastAPI
from aivory_monitor.integrations.fastapi import init_app
import aivory_monitor

app = FastAPI()
aivory_monitor.init(api_key='your-api-key')
init_app(app)

For generic ASGI applications, use AIVoryMiddleware:

from aivory_monitor.integrations.fastapi import AIVoryMiddleware

app = AIVoryMiddleware(app)

Local Development Testing

Quick Test Script

Create a test script to trigger exceptions:

# test-app.py
import aivory_monitor
import time

aivory_monitor.init(
    api_key='ilscipio-dev-2024',
    backend_url='ws://localhost:19999/ws/monitor/agent',
    environment='development',
    debug=True
)

print("Agent initialized, waiting 2s for connection...")
time.sleep(2)

# Test 1: Simple exception
try:
    raise ValueError("Test exception from Python agent")
except Exception as e:
    aivory_monitor.capture_exception(e)

# Test 2: Null reference
try:
    x = None
    x.some_method()
except Exception as e:
    aivory_monitor.capture_exception(e)

# Test 3: With context
try:
    result = 10 / 0
except Exception as e:
    aivory_monitor.capture_exception(e, context={
        'operation': 'divide',
        'user_id': 'test-user'
    })

print("Test exceptions sent. Check backend logs.")
time.sleep(2)
aivory_monitor.shutdown()

Run with:

python test-app.py

Flask Test Server

# test-server.py
from flask import Flask
from aivory_monitor.integrations.flask import init_app
import aivory_monitor

app = Flask(__name__)

aivory_monitor.init(
    api_key='ilscipio-dev-2024',
    backend_url='ws://localhost:19999/ws/monitor/agent',
    environment='development',
    debug=True
)

init_app(app)

@app.route('/error')
def trigger_error():
    raise ValueError("Test sync error")

@app.route('/null')
def trigger_null():
    x = None
    return x.some_attribute

@app.route('/divide')
def trigger_divide():
    return 10 / 0

@app.route('/')
def index():
    return 'Endpoints: /error, /null, /divide'

if __name__ == '__main__':
    print("Test server: http://localhost:5000")
    app.run(port=5000, debug=False)

Test URLs:

Prerequisites for Local Testing

  1. Backend running on localhost:19999
  2. Dev token bypass enabled (uses ilscipio-dev-2024)
  3. Org schema org_test_20 exists in database

Troubleshooting

Breakpoints not working:

  • Ensure enable_breakpoints=True in configuration
  • Check that sys.settrace is not being overridden by debuggers or profilers
  • Only one trace function can be active at a time

High memory usage:

  • Reduce max_capture_depth (default: 10)
  • Reduce max_string_length (default: 1000)
  • Reduce max_collection_size (default: 100)
  • Use sampling_rate to capture only a percentage of exceptions

Agent not connecting:

  • Check backend is running: curl http://localhost:19999/health
  • Check WebSocket endpoint: ws://localhost:19999/ws/monitor/agent
  • Verify API key is set correctly
  • Enable debug mode: debug=True or AIVORY_DEBUG=true

Exceptions not captured in Django:

  • Ensure middleware is added to MIDDLEWARE in settings.py
  • Ensure aivory_monitor.init() is called before Django starts
  • Check that DEBUG=False in production (Django only logs exceptions when DEBUG=False)

Threading issues:

  • The agent uses thread-safe mechanisms for context storage
  • Each thread maintains its own trace function for breakpoints
  • WebSocket connection runs in a separate background thread

Import errors:

  • Ensure websockets>=11.0 is installed: pip install websockets
  • For framework integrations, install optional dependencies:
    • Django: pip install aivory-monitor[django]
    • Flask: pip install aivory-monitor[flask]
    • FastAPI: pip install aivory-monitor[fastapi]

About

AIVory Monitor Python Agent - Runtime exception monitoring via sys.settrace

aivory.net/monitor/

Topics

apmdebuggingexception-handlingmonitoringobservabilitypython

Resources

Readme
MIT license

Contributing

Contributing

Security policy

Security policy
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages

Back | FazBrowse Home | New Git URL