Agent and Request Inspection

class NgrokAgent:
    """
    Container for ngrok agent information.
    """
    data: dict          # Original agent data from ngrok
    status: str         # Status of the agent
    agent_version: str  # Version of the ngrok agent
    session: dict       # Session details for the agent
    uri: str           # URI of the agent

class CapturedRequest:
    """
    Container for captured HTTP request from ngrok tunnel.
    """
    data: dict          # Original request data from ngrok
    id: str            # ID of the captured request
    uri: str           # URI of the captured request
    tunnel_name: str   # Name of tunnel that captured the request
    remote_addr: str   # Remote address of the request
    start: str         # Start time of the request
    duration: int      # Duration of the request in milliseconds
    request: dict      # Request details (method, headers, body, etc.)
    response: dict     # Response details (status, headers, body, etc.)

def get_agent_status(pyngrok_config=None):
    """
    Get the ngrok agent status.

    Parameters:
    - pyngrok_config (PyngrokConfig, optional): Configuration override

    Returns:
    NgrokAgent: The agent status information
    """

from pyngrok import agent, ngrok

# Start a tunnel first to have an active agent
tunnel = ngrok.connect("8000")

# Get agent status
agent_status = agent.get_agent_status()
print(f"Agent status: {agent_status.status}")
print(f"Agent version: {agent_status.agent_version}")
print(f"Agent URI: {agent_status.uri}")

# Access raw agent data
print(f"Session info: {agent_status.session}")

def get_requests(tunnel_name=None, pyngrok_config=None):
    """
    Get list of requests made to tunnels.

    Parameters:
    - tunnel_name (str, optional): Filter by specific tunnel name
    - pyngrok_config (PyngrokConfig, optional): Configuration override

    Returns:
    list[CapturedRequest]: List of captured request objects
    """

from pyngrok import agent, ngrok
import requests
import time

# Create a tunnel
tunnel = ngrok.connect("8000")
print(f"Send requests to: {tunnel.public_url}")

# Make some HTTP requests to generate captured data
# (In real usage, external services would make these requests)
time.sleep(1)  # Allow tunnel to be ready

# Get all captured requests
all_requests = agent.get_requests()
print(f"Total requests captured: {len(all_requests)}")

for req in all_requests[-5:]:  # Last 5 requests
    print(f"Request ID: {req.id}")
    print(f"From: {req.remote_addr}")
    print(f"Method: {req.request.get('method', 'Unknown')}")
    print(f"Path: {req.request.get('uri', 'Unknown')}")
    print(f"Duration: {req.duration}ms")
    print("---")

# Get requests for specific tunnel
tunnel_requests = agent.get_requests(tunnel_name=tunnel.name)
print(f"Requests for {tunnel.name}: {len(tunnel_requests)}")

def get_request(request_id, pyngrok_config=None):
    """
    Get details for a specific captured request.

    Parameters:
    - request_id (str): ID of the request to fetch
    - pyngrok_config (PyngrokConfig, optional): Configuration override

    Returns:
    CapturedRequest: The detailed request information
    """

from pyngrok import agent, ngrok

# Create tunnel and get some requests
tunnel = ngrok.connect("8000")
requests_list = agent.get_requests()

if requests_list:
    # Get detailed info for first request
    request_id = requests_list[0].id
    detailed_request = agent.get_request(request_id)
    
    print(f"Request ID: {detailed_request.id}")
    print(f"Tunnel: {detailed_request.tunnel_name}")
    print(f"Start time: {detailed_request.start}")
    print(f"Duration: {detailed_request.duration}ms")
    
    # Access detailed request information
    request_info = detailed_request.request
    print(f"Method: {request_info.get('method')}")
    print(f"URI: {request_info.get('uri')}")
    print(f"Headers: {request_info.get('headers', {})}")
    
    # Access response information
    response_info = detailed_request.response
    if response_info:
        print(f"Status: {response_info.get('status')}")
        print(f"Response headers: {response_info.get('headers', {})}")

def replay_request(request_id, tunnel_name=None, pyngrok_config=None):
    """
    Replay a captured request through its original or different tunnel.

    Parameters:
    - request_id (str): ID of the request to replay
    - tunnel_name (str, optional): Name of tunnel to replay through
    - pyngrok_config (PyngrokConfig, optional): Configuration override
    """

from pyngrok import agent, ngrok

# Create tunnels
tunnel1 = ngrok.connect("8000")
tunnel2 = ngrok.connect("9000")

# Get captured requests
requests_list = agent.get_requests()

if requests_list:
    request_id = requests_list[0].id
    
    # Replay through original tunnel
    print(f"Replaying request {request_id} through original tunnel")
    agent.replay_request(request_id)
    
    # Replay through different tunnel
    print(f"Replaying request {request_id} through {tunnel2.name}")
    agent.replay_request(request_id, tunnel_name=tunnel2.name)

def delete_requests(pyngrok_config=None):
    """
    Delete all captured request history.

    Parameters:
    - pyngrok_config (PyngrokConfig, optional): Configuration override
    """

from pyngrok import agent, ngrok

# Create tunnel and generate some requests
tunnel = ngrok.connect("8000")

# Check current request count
requests_before = agent.get_requests()
print(f"Requests before cleanup: {len(requests_before)}")

# Clear all request history
agent.delete_requests()

# Verify cleanup
requests_after = agent.get_requests()
print(f"Requests after cleanup: {len(requests_after)}")  # Should be 0

from pyngrok import agent, ngrok
from collections import Counter
import json

# Create tunnel for analysis
tunnel = ngrok.connect("8000")
print(f"Analyze requests at: {tunnel.public_url}")

# Wait for some requests to be made...
# Then analyze the captured data

requests_list = agent.get_requests()

# Analyze request patterns
if requests_list:
    # Count requests by method
    methods = [req.request.get('method', 'Unknown') for req in requests_list]
    method_counts = Counter(methods)
    print(f"Request methods: {dict(method_counts)}")
    
    # Count requests by path
    paths = [req.request.get('uri', 'Unknown') for req in requests_list]
    path_counts = Counter(paths)
    print(f"Popular paths: {dict(path_counts)}")
    
    # Analyze response times
    durations = [req.duration for req in requests_list if req.duration]
    if durations:
        avg_duration = sum(durations) / len(durations)
        max_duration = max(durations)
        print(f"Average response time: {avg_duration:.2f}ms")
        print(f"Max response time: {max_duration}ms")
    
    # Find errors
    error_requests = []
    for req in requests_list:
        if req.response:
            status = req.response.get('status', 0)
            if status >= 400:
                error_requests.append(req)
    
    print(f"Error requests: {len(error_requests)}")
    for error_req in error_requests[-3:]:  # Last 3 errors
        status = error_req.response.get('status')
        path = error_req.request.get('uri')
        print(f"  {status} error on {path}")

# Export request data for external analysis
def export_requests_to_json(filename):
    requests_list = agent.get_requests()
    export_data = []
    for req in requests_list:
        export_data.append({
            'id': req.id,
            'tunnel_name': req.tunnel_name,
            'remote_addr': req.remote_addr,
            'start': req.start,
            'duration': req.duration,
            'request': req.request,
            'response': req.response
        })
    
    with open(filename, 'w') as f:
        json.dump(export_data, f, indent=2)
    print(f"Exported {len(export_data)} requests to {filename}")

# Export current requests
export_requests_to_json("ngrok_requests.json")

from pyngrok import agent, ngrok
import time
import threading

def monitor_requests(tunnel_name, alert_threshold=10):
    """Monitor request rate and alert on high traffic"""
    last_count = 0
    
    while True:
        current_requests = agent.get_requests(tunnel_name=tunnel_name)
        current_count = len(current_requests)
        
        requests_per_minute = current_count - last_count
        if requests_per_minute > alert_threshold:
            print(f"ALERT: High traffic detected! {requests_per_minute} requests/minute")
            
            # Analyze recent requests
            recent_requests = current_requests[-requests_per_minute:]
            error_count = sum(1 for req in recent_requests 
                            if req.response and req.response.get('status', 0) >= 400)
            print(f"  Error rate: {error_count}/{requests_per_minute}")
        
        last_count = current_count
        time.sleep(60)  # Check every minute

# Start tunnel and monitoring
tunnel = ngrok.connect("8000")
monitor_thread = threading.Thread(
    target=monitor_requests, 
    args=(tunnel.name, 5),  # Alert if >5 requests/minute
    daemon=True
)
monitor_thread.start()

print(f"Monitoring traffic to {tunnel.public_url}")
print("Send requests to trigger monitoring...")