Skip to content
Vysion API Documentation

Python Library Guide

The Vysion Python library provides a convenient wrapper around the Vysion API, making it easy to integrate threat intelligence data into your Python applications. This guide covers installation, configuration, and usage patterns.

Installation

Section titled “Installation”

Install the Vysion Python library using pip:

Terminal window
pip install vysion

For development or the latest features, you can install from the source repository:

Terminal window
pip install git+https://github.com/vysion-ai/vysion-python.git

Quick Start

Section titled “Quick Start”

Basic Configuration

Section titled “Basic Configuration”
import os
from vysion import client


# Set your API key as an environment variable
os.environ['VYSION_API_KEY'] = 'your_api_key_here'


# Initialize the client
c = client.Client()

First API Call

Section titled “First API Call”
from vysion import client


c = client.Client(api_key='your_api_key_here')


# Search for documents containing a specific term
result = c.search_documents(query='lockbit')
print(f"Found {result['data']['total']} documents")


# Display first result
if result['data']['hits']:
    first_doc = result['data']['hits'][0]
    print(f"Title: {first_doc['page']['title']}")
    print(f"URL: {first_doc['page']['url']}")

Core Methods

Section titled “Core Methods”
Section titled “Document Search”

The search_documents method is the most versatile way to find documents:

# Simple text search
result = c.search_documents(query='ransomware')


# Search with pagination
result = c.search_documents(
    query='alphv',
    page=1,
    page_size=50
)

Document Retrieval

Section titled “Document Retrieval”

Get specific documents by various identifiers:

# Get document by ID
doc = c.get_document_by_id('document_id_here')


# Get document by URL
doc = c.get_document_by_url('https://example.onion/page')


# Get document by tag
docs = c.get_documents_by_tag('ransomware')


# Get documents by email
docs = c.get_documents_by_email('admin@company.com')


# Get documents by wallet address
docs = c.get_documents_by_wallet('BTC', 'bc1q026rl6hjkdywnsrtva26mq2w0avs9k850ew2d6')


# Get documents by phone number
docs = c.get_documents_by_phone('+1234567890')

HTML Content

Section titled “HTML Content”

Retrieve the full HTML content of documents:

# Get HTML content by document ID
html_content = c.get_document_html('document_id_here')


# Get HTML content by URL
html_content = c.get_document_html_by_url('https://example.onion/page')

Ransomware Intelligence

Section titled “Ransomware Intelligence”

Access ransomware-specific intelligence:

# Search ransomware victims
victims = c.search_ransomware_victims(
    query='healthcare',
    countries='US',
    groups='lockbit,alphv'
)


# Get specific victim by ID
victim = c.get_ransomware_victim('victim_id_here')

Instant Messaging

Section titled “Instant Messaging”

Access Telegram and Discord data:

# Get chat messages from a channel
chat = c.get_im_chat(
    platform='Telegram',
    channel_id='-1002018436281',
    gte='2024-05-16T10:57:58.632466',
    lte='2024-05-16T23:00:00'
)


# Get specific message
message = c.get_im_message('Telegram', '-1002018336281_85')

Feeds

Section titled “Feeds”

Access daily threat intelligence feeds:

from datetime import datetime, timedelta


# Get yesterday's feeds
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')


# Get ransomware feed
ransomware_feed = c.get_ransomware_feed(batch=yesterday)


# Get Telegram feed
telegram_feed = c.get_telegram_feed(batch=yesterday)


# Get wallets feed
wallets_feed = c.get_wallets_feed(batch=yesterday)

Advanced Usage Patterns

Section titled “Advanced Usage Patterns”

Bulk Data Processing

Section titled “Bulk Data Processing”
from vysion import client
import time


c = client.Client(api_key='your_api_key_here')


def process_all_victims(query, page_size=100):
    """Process all ransomware victims matching a query."""
    page = 1
    all_victims = []


    while True:
        try:
            result = c.search_ransomware_victims(
                query=query,
                page=page,
                page_size=page_size
            )


            hits = result['data']['hits']
            all_victims.extend(hits)


            print(f"Processed page {page}, got {len(hits)} victims")


            # Check if we've reached the end
            if len(hits) < page_size:
                break


            page += 1


            # Rate limiting - be respectful to the API
            time.sleep(0.1)


        except Exception as e:
            print(f"Error on page {page}: {e}")
            break


    return all_victims


# Usage
healthcare_victims = process_all_victims('healthcare')
print(f"Total healthcare victims found: {len(healthcare_victims)}")

Error Handling

Section titled “Error Handling”
from vysion import client
from requests.exceptions import RequestException


c = client.Client(api_key='your_api_key_here')


def safe_api_call(func, *args, **kwargs):
    """Wrapper for safe API calls with error handling."""
    try:
        result = func(*args, **kwargs)


        # Check for API errors
        if result.get('error'):
            print(f"API Error: {result['error']['message']}")
            return None


        return result


    except RequestException as e:
        print(f"Network Error: {e}")
        return None
    except Exception as e:
        print(f"Unexpected Error: {e}")
        return None


# Usage
result = safe_api_call(c.search_documents, query='lockbit')
if result:
    print(f"Found {result['data']['total']} documents")

Data Export

Section titled “Data Export”
import json
import csv
from vysion import client


c = client.Client(api_key='your_api_key_here')


def export_victims_to_csv(query, filename):
    """Export ransomware victims to CSV file."""
    victims = c.search_ransomware_victims(query=query, page_size=1000)


    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        fieldnames = ['companyName', 'country', 'industry', 'ransomwareGroup', 'detectionDate']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)


        writer.writeheader()
        for victim in victims['data']['hits']:
            writer.writerow({
                'companyName': victim.get('companyName', ''),
                'country': victim.get('country', ''),
                'industry': victim.get('industry', ''),
                'ransomwareGroup': victim.get('ransomwareGroup', ''),
                'detectionDate': victim.get('detectionDate', '')
            })


def export_to_json(data, filename):
    """Export data to JSON file."""
    with open(filename, 'w', encoding='utf-8') as jsonfile:
        json.dump(data, jsonfile, indent=2, ensure_ascii=False)


# Usage
export_victims_to_csv('healthcare', 'healthcare_victims.csv')

Monitoring and Alerting

Section titled “Monitoring and Alerting”
import time
from datetime import datetime, timedelta
from vysion import client


c = client.Client(api_key='your_api_key_here')


def monitor_new_victims(keywords, check_interval=3600):
    """Monitor for new ransomware victims containing keywords."""
    last_check = datetime.now() - timedelta(hours=24)


    while True:
        try:
            # Check for new victims since last check
            result = c.search_ransomware_victims(
                query=' OR '.join(keywords),
                gte=last_check.isoformat()
            )


            new_victims = result['data']['hits']


            if new_victims:
                print(f"ALERT: {len(new_victims)} new victims found!")
                for victim in new_victims:
                    print(f"- {victim['companyName']} ({victim['ransomwareGroup']})")


                # Send alerts (implement your notification logic here)
                send_alert(new_victims)


            last_check = datetime.now()
            time.sleep(check_interval)


        except Exception as e:
            print(f"Monitoring error: {e}")
            time.sleep(60)  # Wait before retrying


def send_alert(victims):
    """Implement your alerting logic here."""
    # Email, Slack, webhook, etc.
    pass


# Usage
keywords = ['healthcare', 'hospital', 'medical', 'clinic']
monitor_new_victims(keywords)

Configuration Options

Section titled “Configuration Options”

Client Configuration

Section titled “Client Configuration”
from vysion import client


# Full configuration
c = client.Client(
    api_key='your_api_key_here',
    base_url='https://api.vysion.ai',  # Custom base URL if needed
    timeout=30,  # Request timeout in seconds
    retries=3,   # Number of retry attempts
    backoff_factor=0.3  # Backoff factor for retries
)

Environment Variables

Section titled “Environment Variables”

The library supports these environment variables:

VariableDescriptionDefault
VYSION_API_KEYYour API keyNone
VYSION_BASE_URLAPI base URLhttps://api.vysion.ai
VYSION_TIMEOUTRequest timeout30

Performance Tips

Section titled “Performance Tips”

1. Use Appropriate Page Sizes

Section titled “1. Use Appropriate Page Sizes”
# For bulk processing, use larger page sizes
result = c.search_documents(query='ransomware', page_size=1000)


# For real-time applications, use smaller page sizes
result = c.search_documents(query='ransomware', page_size=10)

2. Implement Caching

Section titled “2. Implement Caching”
import functools
import time


@functools.lru_cache(maxsize=128)
def cached_country_stats(countries, timestamp):
    """Cache country stats for 1 hour."""
    return c.ransomware_countries_stats(countries)


# Usage with hourly cache
current_hour = int(time.time() // 3600)
stats = cached_country_stats('US,UK,DE', current_hour)

3. Use Specific Queries

Section titled “3. Use Specific Queries”
# Less efficient - broad search
result = c.search_documents(query='data')


# More efficient - specific search
result = c.search_documents(query='healthcare AND lockbit')

Troubleshooting

Section titled “Troubleshooting”

Common Issues

Section titled “Common Issues”

  1. Authentication Errors

    # Check if your API key is valid
    try:
        result = c.search_documents(query='test', page_size=1)
        print("Authentication successful")
    except Exception as e:
        print(f"Authentication failed: {e}")

  2. Rate Limiting

    import time
    

    def rate_limited_call(func, *args, **kwargs):
        while True:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if 'rate limit' in str(e).lower():
                    print("Rate limited, waiting...")
                    time.sleep(60)
                else:
                    raise e

  3. Network Issues

    import requests
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry
    

    # Configure retry strategy
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

Debug Mode

Section titled “Debug Mode”

Enable debug mode to see detailed request/response information:

import logging
from vysion import client


# Enable debug logging
logging.basicConfig(level=logging.DEBUG)


c = client.Client(api_key='your_api_key_here')
result = c.search_documents(query='test')

Next Steps

Section titled “Next Steps”
  • Explore the API Reference for detailed endpoint documentation
  • Check out MISP Integration for threat intelligence platform integration
  • Review Rate Limiting best practices
  • Join our community forums for support and discussions