### Test Proxies with Python

Source: https://arshiacomplus.github.io/docs/python-v2ray/getting-started/quick-start

This Python script automates the process of testing proxy URIs. It ensures required binaries are downloaded, parses a list of URIs, and tests their connectivity concurrently. The results are then displayed, sorted by latency, indicating success or failure for each proxy.

```python
from pathlib import Path
from python_v2ray.downloader import BinaryDownloader
from python_v2ray.tester import ConnectionTester
from python_v2ray.config_parser import parse_uri

def main():
    """
    Ensures binaries are present, parses URIs, and tests their connectivity.
    """
    project_root = Path("./")

    # --- 1. Ensure all required binaries are available ---
    print("--- Verifying binaries ---")
    try:
        downloader = BinaryDownloader(project_root)
        downloader.ensure_all()
    except Exception as e:
        print(f"Fatal Error: {e}")
        return

    # --- 2. Add your proxy URIs here ---
    test_uris = [
        "vless://...",
        "vmess://...",
        "hysteria2://...",
        # ... add as many as you want
    ]

    # --- 3. Parse all URIs into a unified format ---
    print("\n* Parsing URIs...")
    parsed_configs = [p for p in (parse_uri(uri) for uri in test_uris) if p]

    if not parsed_configs:
        print("No valid configurations found to test.")
        return

    print(f"* Preparing to test {len(parsed_configs)} configurations concurrently...")

    # --- 4. Initialize and run the tester ---
    tester = ConnectionTester(
        vendor_path=str(project_root / "vendor"),
        core_engine_path=str(project_root / "core_engine")
    )
    results = tester.test_uris(parsed_configs)

    # --- 5. Display the results, sorted by latency ---
    print("\n" + "="*20 + " Test Results " + "="*20)
    if results:
        sorted_results = sorted(results, key=lambda x: x.get('ping_ms', 9999))
        for result in sorted_results:
            tag = result.get('tag', 'N/A')
            ping = result.get('ping_ms', -1)
            status = result.get('status', 'error')

            if status == 'success':
                print(f"✅ Tag: {tag:<35} | Latency: {ping:>4} ms | Status: {status}")
            else:
                # Clean up error messages for better readability
                clean_status = status.split('|').strip()
                print(f"❌ Tag: {tag:<35} | Latency: {ping:>4} ms | Status: {clean_status}")
    else:
        print("No results were received from the tester.")
    print("="*56)

if __name__ == "__main__":
    main()

```

--------------------------------

### Proxy Chaining with WARP ('WARP on Any')

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/proxy-chaining

This section explains the powerful proxy chaining feature, allowing traffic from any Xray-compatible configuration to be routed through a final exit-point proxy like Cloudflare WARP. It details the traffic flow and provides a practical example.

```APIDOC
## Proxy Chaining ('WARP on Any')

### Description
This feature enables routing traffic from any Xray-compatible configuration through a final exit-point proxy, commonly used for privacy enhancement or bypassing network restrictions by creating a 'WARP on Any' setup. The `XrayConfigBuilder` automatically configures outbounds to use the provided WARP configuration as their `dialerProxy`.

### Traffic Flow
`Your App` -> `Local SOCKS Inbound` -> `Primary Outbound (e.g., VLESS)` -> `WARP Outbound (WireGuard)` -> `Internet`

### Practical Example: Testing Connectivity through WARP
This example demonstrates testing connectivity of multiple configurations routed through a single WARP profile.

```python
# examples/07_warp_tester.py

import os
import sys
from pathlib import Path

# Add project root to path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from python_v2ray.downloader import BinaryDownloader
from python_v2ray.tester import ConnectionTester
from python_v2ray.config_parser import parse_uri

def main():
    project_root = Path(__file__).parent.parent

    # Ensure all necessary binaries are ready
    downloader = BinaryDownloader(project_root)
    downloader.ensure_all()

    # --- 1. Define your WARP config (must be a WireGuard URI) ---
    warp_uri = "wireguard://YOUR_PRIVATE_KEY@engage.cloudflareclient.com:2408?address=172.16.0.2/32&publicKey=...#WARP-Profile"

    # --- 2. Define the primary configs to test THROUGH WARP ---
    test_uris = [
        "vless://YOUR_UUID@your.domain.com:443?type=ws#VLESS-through-WARP",
        "trojan://YOUR_PASSWORD@another.domain.com:443#Trojan-through-WARP",
        # You can even chain another WireGuard config through the main WARP config
        # "wireguard://ANOTHER_PRIVATE_KEY@...#WG-on-WARP"
    ]

    # --- 3. Parse all configurations ---
    parsed_warp_config = parse_uri(warp_uri)
    parsed_test_configs = [p for p in (parse_uri(uri) for uri in test_uris) if p]

    # --- Input Validation ---
    if not parsed_warp_config or "YOUR_PRIVATE_KEY" in warp_uri:
        print("! Please provide a valid WARP WireGuard URI.")
        return
    if not parsed_test_configs:
        print("! No valid primary configurations found to test.")
        return

    tester = ConnectionTester(
        vendor_path=str(project_root / "vendor"), 
        core_engine_path=str(project_root / "core_engine")
    )

    # --- 4. Run the test, passing the WARP config to the `warp_config` parameter ---
    print(f"\n--- Running Connectivity Test with WARP-on-Any enabled ---")
    results = tester.test_uris(
        parsed_params=parsed_test_configs,
        warp_config=parsed_warp_config,
        timeout=30  # Allow a longer timeout for chained connections
    )

    # --- 5. Print the results ---
    if results:
        print("\n" + "="*20 + " WARP ON ANY PING TEST RESULTS " + "="*20)
        sorted_results = sorted(results, key=lambda x: x.get('ping_ms', 9999))
        for result in sorted_results:
            tag = result.get('tag', 'N/A')
            ping = result.get('ping_ms', -1)
            status = result.get('status', 'error')

            if status == 'success':
                print(f"* Tag: {tag:<30} | Ping: {ping:>4} ms | Status: SUCCESS")
            else:
                print(f"! Tag: {tag:<30} | Ping: ---- ms | Status: FAILED ({status})")
    else:
        print("! No results were received from the tester.")

if __name__ == "__main__":
    main()

```

### Important Considerations
*   The `warp_config` **must** be a valid `ConfigParams` object generated from a WireGuard URI.
*   This chaining feature is handled by Xray's `dialerProxy` and applies only to protocols managed within the merged Xray instance (VLESS, VMess, Trojan, etc.). It does **not** apply to protocols like Hysteria launched in a separate process by the Go test engine.
```

--------------------------------

### Load Configurations from Local File with Subscription Link (Python)

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/loading-from-sources

Loads configurations from a local file that contains a single subscription link. By setting the `is_subscription` flag to `True`, the function will read the URL from the file and then fetch and decode its content. This is useful for managing subscription links locally.

```python
# --- contents of my_sub.txt ---
# https://another.subscription.link/path

from pathlib import Path
from python_v2ray.config_parser import load_configs

file_path = Path("my_sub.txt")

# The flag tells the function to read the URL from the file and then fetch it
parsed_configs = load_configs(source=file_path, is_subscription=True)
```

--------------------------------

### Load Configurations from Subscription URL (Python)

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/loading-from-sources

Fetches and decodes configurations from a remote subscription URL. The `is_subscription` flag must be set to `True` to indicate that the source is a subscription link that needs to be fetched and decoded. This is the most common use case for loading multiple configurations at once.

```python
from python_v2ray.config_parser import load_configs

sub_url = "https://your.subscription.link/path"

# The flag tells the function to fetch the URL and decode its content
parsed_configs = load_configs(source=sub_url, is_subscription=True)

print(f"Loaded {len(parsed_configs)} configs from the subscription.")
```

--------------------------------

### Build Xray Config JSON using Python Builder

Source: https://arshiacomplus.github.io/docs/python-v2ray/api/config-parser

Demonstrates using `XrayConfigBuilder` to programmatically construct a complete Xray `config.json`. This fluent API allows chaining methods to add inbound and outbound configurations, including building outbound configurations directly from parsed `ConfigParams`.

```python
from python_v2ray.config_parser import XrayConfigBuilder, parse_uri

# Start with a parsed params object
params = parse_uri("vless://...")

# Initialize the builder
builder = XrayConfigBuilder()

# Chain methods to build the config
builder.add_inbound({
    "protocol": "socks",
    "port": 10808,
    "listen": "127.0.0.1"
})

outbound_config = builder.build_outbound_from_params(params)
builder.add_outbound(outbound_config)

# Add default routing rules
builder.add_outbound({"protocol": "freedom", "tag": "direct"})
builder.add_outbound({"protocol": "blackhole", "tag": "block"})

# Get the final JSON string
final_json = builder.to_json()

print(final_json)

```

--------------------------------

### Load Configurations from Local File with URIs per Line (Python)

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/loading-from-sources

Loads configurations from a local text file where each line contains a single URI. The `load_configs` function automatically handles this format when a `Path` object is provided as the source. No special flags are required for this default file loading behavior.

```python
# --- contents of my_configs.txt ---
# vless://...
# trojan://...

from pathlib import Path
from python_v2ray.config_parser import load_configs

file_path = Path("my_configs.txt")

# No flag is needed; this is the default behavior for files
parsed_configs = load_configs(source=file_path)
```

--------------------------------

### API Reference: The `warp_config` Parameter

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/proxy-chaining

Details the `warp_config` parameter available in the `ConnectionTester` methods for enabling proxy chaining.

```APIDOC
## API Reference: The `warp_config` Parameter

### Description
To enable proxy chaining, a new optional parameter `warp_config` has been added to the primary testing methods of the `ConnectionTester` class.

### Methods Supporting `warp_config`
*   `test_uris(..., warp_config: Optional[ConfigParams] = None)`
*   `test_speed(..., warp_config: Optional[ConfigParams] = None)`
*   `test_upload(..., warp_config: Optional[ConfigParams] = None)`

### Usage
Pass a parsed `ConfigParams` object, generated from a valid WireGuard URI, to the `warp_config` parameter. The `ConnectionTester` will then automatically configure the proxy chain.
```

--------------------------------

### Deduplicate V2Ray Configurations with Python

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/deduplicating-configs

Demonstrates how to use the `deduplicate_configs` function to remove duplicate V2Ray configurations from a list. It parses raw URIs, applies deduplication, and prints the unique configurations. Dependencies include the `python_v2ray` library.

```python
# examples/09_deduplicate_configs.py

import os
import sys

# Add project root to path to find our library
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from python_v2ray.config_parser import load_configs, deduplicate_configs

def main():
    """
    * A minimal example to demonstrate how to remove duplicate configurations
    * from a list, ignoring their tags (# remarks).
    """

    # --- 1. Define a list with duplicate configurations ---
    # Note: The first two VLESS configs are identical except for their tag.
    raw_uris = [
        "vless://abcdef@example.com:443?type=ws#VLESS-Config-1",
        "vless://abcdef@example.com:443?type=ws#VLESS-Config-2-DUPLICATE",
        "trojan://password@anotherexample.com:443#Trojan-Config-UNIQUE",
    ]

    print("--- Initial List of Raw URIs ---")
    for uri in raw_uris:
        print(f"- {uri}")

    # --- 2. Load the configs into ConfigParams objects ---
    parsed_configs = load_configs(source=raw_uris)
    print(f"\n* Initially parsed {len(parsed_configs)} configurations.")

    # --- 3. Apply the deduplication function ---
    print("\n--- Applying Deduplication (Ignoring Tags) ---")
    unique_configs = deduplicate_configs(parsed_configs)

    # --- 4. Display the final, clean list ---
    print(f"\n* Found {len(unique_configs)} unique configurations.")
    print("Tags of final unique configs:", [p.tag for p in unique_configs])

if __name__ == "__main__":
    main()

```

--------------------------------

### Load Configurations from Python List (Python)

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/loading-from-sources

Loads configurations directly from a Python list of URI strings. This method is the most straightforward when you already have your configuration URIs available in memory. The `load_configs` function parses each URI in the list.

```python
from python_v2ray.config_parser import load_configs

my_uri_list = [
    "vless://abcdef@example.com:443?type=ws#Config-1",
    "trojan://password@anotherexample.com:443#Config-2",
]

# This is the most direct way to parse a list of configs
parsed_configs = load_configs(source=my_uri_list)
```

--------------------------------

### Test Connectivity Through WARP with Python V2Ray

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/proxy-chaining

This Python script demonstrates testing the connectivity of multiple proxy configurations by routing them through a single WARP (WireGuard) profile. It requires the `python-v2ray` library and ensures necessary binaries are downloaded. The script parses WARP and test URIs, then uses `ConnectionTester` to perform ping tests, outputting success or failure status.

```python
# examples/07_warp_tester.py

import os
import sys
from pathlib import Path

# Add project root to path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from python_v2ray.downloader import BinaryDownloader
from python_v2ray.tester import ConnectionTester
from python_v2ray.config_parser import parse_uri

def main():
    project_root = Path(__file__).parent.parent

    # Ensure all necessary binaries are ready
    downloader = BinaryDownloader(project_root)
    downloader.ensure_all()

    # --- 1. Define your WARP config (must be a WireGuard URI) ---
    warp_uri = "wireguard://YOUR_PRIVATE_KEY@engage.cloudflareclient.com:2408?address=172.16.0.2/32&publicKey=...#WARP-Profile"

    # --- 2. Define the primary configs to test THROUGH WARP ---
    test_uris = [
        "vless://YOUR_UUID@your.domain.com:443?type=ws#VLESS-through-WARP",
        "trojan://YOUR_PASSWORD@another.domain.com:443#Trojan-through-WARP",
        # You can even chain another WireGuard config through the main WARP config
        # "wireguard://ANOTHER_PRIVATE_KEY@...#WG-on-WARP"
    ]

    # --- 3. Parse all configurations ---
    parsed_warp_config = parse_uri(warp_uri)
    parsed_test_configs = [p for p in (parse_uri(uri) for uri in test_uris) if p]

    # --- Input Validation ---
    if not parsed_warp_config or "YOUR_PRIVATE_KEY" in warp_uri:
        print("! Please provide a valid WARP WireGuard URI.")
        return
    if not parsed_test_configs:
        print("! No valid primary configurations found to test.")
        return

    tester = ConnectionTester(
        vendor_path=str(project_root / "vendor"), 
        core_engine_path=str(project_root / "core_engine")
    )

    # --- 4. Run the test, passing the WARP config to the `warp_config` parameter ---
    print(f"\n--- Running Connectivity Test with WARP-on-Any enabled ---")
    results = tester.test_uris(
        parsed_params=parsed_test_configs,
        warp_config=parsed_warp_config,
        timeout=30  # Allow a longer timeout for chained connections
    )

    # --- 5. Print the results ---
    if results:
        print("\n" + "="*20 + " WARP ON ANY PING TEST RESULTS " + "="*20)
        sorted_results = sorted(results, key=lambda x: x.get('ping_ms', 9999))
        for result in sorted_results:
            tag = result.get('tag', 'N/A')
            ping = result.get('ping_ms', -1)
            status = result.get('status', 'error')

            if status == 'success':
                print(f"* Tag: {tag:<30} | Ping: {ping:>4} ms | Status: SUCCESS")
            else:
                print(f"! Tag: {tag:<30} | Ping: ---- ms | Status: FAILED ({status})")
    else:
        print("! No results were received from the tester.")

if __name__ == "__main__":
    main()

```

--------------------------------

### Enable Global Mux Setting in python-v2ray

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/multiplexing

This code snippet shows how to apply a default Mux configuration to all Xray-compatible proxies that don't already have Mux enabled in their URI. It uses the `ConnectionTester` class to test URIs and applies the global Mux settings.

```python
from python_v2ray.tester import ConnectionTester
from python_v2ray.config_parser import parse_uri

# Define your global Mux settings
global_mux_settings = {"enabled": True, "concurrency": 8}

# This standard vless config does NOT have Mux defined in its URL
uris = ["vless://UUID@example.com:443?type=ws#StandardVless"]
parsed_configs = [parse_uri(uri) for uri in uris]

# Initialize the tester
tester = ConnectionTester(vendor_path="./vendor", core_engine_path="./core_engine")

# Pass the global config to the tester
# The tester will now apply these Mux settings to the 'StandardVless' config
results = tester.test_uris(
    parsed_params=parsed_configs,
    mux_config=global_mux_settings
)
```

--------------------------------

### Enable Mux via URI in python-v2ray

Source: https://arshiacomplus.github.io/docs/python-v2ray/core-concepts/multiplexing

This code snippet demonstrates how to enable Mux and set its concurrency directly within the configuration link using the `mvless` protocol scheme. It parses a URI containing Mux parameters and prints the Mux status and concurrency.

```python
from python_v2ray.config_parser import parse_uri

# Example mvless URI with Mux enabled
uri = "mvless://UUID@example.com:443?mux=ON&muxConcurrency=8&type=ws#MyMuxConfig"

params = parse_uri(uri)

if params:
    print(f"Mux Enabled: {params.mux_enabled}")
    print(f"Mux Concurrency: {params.mux_concurrency}")
```

--------------------------------

### Monitor Live Proxy Traffic with Xray API

Source: https://arshiacomplus.github.io/docs/python-v2ray/api/monitoring-traffic

This Python script enables monitoring of live uplink and downlink traffic for Xray-compatible protocols. It requires the `python-v2ray` library and a running Xray instance. The script configures Xray to expose its API, then continuously fetches and displays traffic statistics for a specified outbound tag.

```python
import time
from pathlib import Path
from python_v2ray.core import XrayCore
from python_v2ray.config_parser import parse_uri, XrayConfigBuilder

def monitor_proxy_traffic():
    project_root = Path("./")

    # --- Step 1: Configuration ---

    # A working Xray-compatible URI
    proxy_uri = "vless://YOUR_UUID@your.domain.com:443?security=tls&sni=your.domain.com#MyProxy"

    # Define a tag for the outbound we want to monitor
    outbound_tag = "proxy_main"

    # Choose a free port for the gRPC API service
    api_port = 62789

    # --- Step 2: Build the Xray Config with API Enabled ---

    params = parse_uri(proxy_uri)
    if not params:
        print("Invalid URI.")
        return

    # Assign our custom tag to the parsed config
    params.tag = outbound_tag

    print(f"* Building config to monitor outbound with tag: '{outbound_tag}'")
    builder = XrayConfigBuilder()

    # !! CRITICAL: Enable the API service on the specified port
    builder.enable_api(port=api_port)

    builder.add_inbound({
        "port": 10808, "listen": "127.0.0.1", "protocol": "socks"
    })

    outbound_config = builder.build_outbound_from_params(params)
    builder.add_outbound(outbound_config)

    # --- Step 3: Run XrayCore and Start Monitoring ---

    print("* Starting Xray with API enabled...")

    # !! CRITICAL: Pass the api_port to the XrayCore constructor
    with XrayCore(vendor_dir=str(project_root/"vendor"), config_builder=builder, api_port=api_port) as xray:
        if not xray.is_running():
            print("! Xray failed to start.")
            return

        print(f"\n* Xray is running. SOCKS on 10808. API on {api_port}.")
        print("* Now, generate some traffic through the SOCKS proxy (e.g., watch a video).")
        print("* Press Ctrl+C to stop.\n")

        try:
            while True:
                time.sleep(2)
                stats = xray.get_stats(tag=outbound_tag)

                if stats:
                    # Convert bytes to Megabytes (MB)
                    uplink_mb = stats['uplink'] / (1024 * 1024)
                    downlink_mb = stats['downlink'] / (1024 * 1024)
                    print(
                        f"\r* Live Stats for '{outbound_tag}': "
                        f"Uplink: {uplink_mb:.2f} MB | "
                        f"Downlink: {downlink_mb:.2f} MB   ",
                        end=""
                    )
                else:
                    # This message appears until the first byte of traffic is sent
                    print(f"\r* Waiting for traffic on tag '{outbound_tag}'...", end="")

        except KeyboardInterrupt:
            print("\n* User stopped the stats viewer.")

    print("\n* Monitoring finished. Xray has been stopped.")
if __name__ == "__main__":
    monitor_proxy_traffic()

```

--------------------------------

### Parse Proxy URI String in Python

Source: https://arshiacomplus.github.io/docs/python-v2ray/api/config-parser

The `parse_uri` function parses various supported proxy URI strings, automatically detecting the protocol and using the appropriate internal parser. It returns a `ConfigParams` object on success or `None` if the URI is invalid or the protocol is unsupported.

```python
from python_v2ray.config_parser import parse_uri

uri = "trojan://YOUR_PASSWORD@your.domain.com:443?sni=your.domain.com#MyTrojan"
params = parse_uri(uri)

if params:
    assert params.protocol == "trojan"
    assert params.password == "YOUR_PASSWORD"
    assert params.sni == "your.domain.com"
    assert params.tag == "MyTrojan"

```

--------------------------------

### Define ConfigParams Dataclass in Python

Source: https://arshiacomplus.github.io/docs/python-v2ray/api/config-parser

Defines the `ConfigParams` dataclass, a standardized, protocol-agnostic data structure for proxy configurations. It includes fields for protocol, address, port, and various optional protocol-specific parameters.

```python
from dataclasses import dataclass
from typing import Optional

@dataclass
class ConfigParams:
    protocol: str
    address: str
    port: int
    tag: Optional[str] = "proxy"
    id: Optional[str] = ""
    security: Optional[str] = ""
    network: Optional[str] = "tcp"
    # ... and many other protocol-specific fields

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.