Back to skills
SkillHub ClubShip Full StackFull Stack
aria2-json-rpc
Interact with aria2 download manager via JSON-RPC 2.0. Manage downloads, query status, and control tasks through natural language commands. Use when working with aria2, download management, or torrent operations.
Packaged view
This page reorganizes the original catalog entry around fit, installability, and workflow context first. The original raw source lives below.
Stars
3,131
Hot score
99
Updated
March 20, 2026
Overall rating
C0.0
Composite score
0.0
Best-practice grade
B73.6
Install command
npx @skill-hub/cli install openclaw-skills-aria2-json-rpc
Repository
openclaw/skills
Skill path: skills/azzgo/aria2-json-rpc
Interact with aria2 download manager via JSON-RPC 2.0. Manage downloads, query status, and control tasks through natural language commands. Use when working with aria2, download management, or torrent operations.
Open repositoryBest for
Primary workflow: Ship Full Stack.
Technical facets: Full Stack.
Target audience: everyone.
License: MIT.
Original source
Catalog source: SkillHub Club.
Repository owner: openclaw.
This is still a mirrored public skill entry. Review the repository before installing into production workflows.
What it helps with
- Install aria2-json-rpc into Claude Code, Codex CLI, Gemini CLI, or OpenCode workflows
- Review https://github.com/openclaw/skills before adding aria2-json-rpc to shared team environments
- Use aria2-json-rpc for development workflows
Works across
Claude CodeCodex CLIGemini CLIOpenCode
Favorites: 0.
Sub-skills: 0.
Aggregator: No.
Original source / Raw SKILL.md
---
name: aria2-json-rpc
description: Interact with aria2 download manager via JSON-RPC 2.0. Manage downloads, query status, and control tasks through natural language commands. Use when working with aria2, download management, or torrent operations.
license: MIT
compatibility: Requires Python 3.6+. WebSocket support requires websockets package (pip install websockets) and Python version must match dependency requirements.
metadata:
author: ISON
version: "1.1.0"
---
## What This Skill Does
This skill enables you to control aria2 download manager through natural language commands:
- Download files (HTTP/HTTPS/FTP/Magnet/Torrent/Metalink)
- Monitor download progress and status
- Control downloads (pause, resume, remove)
- Manage batch operations (pause all, resume all)
- View statistics and configure options
## How to Use (For AI Agents)
**⚠️ CRITICAL: DO NOT manually construct JSON-RPC requests.**
**✅ ALWAYS use the Python scripts in the `scripts/` directory.**
**⚠️ IMPORTANT: Use `python3` command, NOT `python`** (especially on macOS where `python` symlink doesn't exist)
### Workflow (MUST FOLLOW)
**Step 1: Check Configuration Status**
Before executing any aria2 commands, ALWAYS check if configuration is ready:
```bash
python3 scripts/config_loader.py test
```
- If **successful**: Proceed to execute user's command
- If **failed**: Guide user to initialize configuration (see Step 2)
**Step 2: Initialize Configuration (if needed)**
If connection test fails, guide user to set up configuration:
```bash
# Recommended: User config (survives skill updates)
python3 scripts/config_loader.py init --user
# Alternative: Local config (project-specific)
python3 scripts/config_loader.py init --local
```
Then instruct user to edit the generated config file with their aria2 server details.
**Step 3: Execute User Commands**
Once configuration is ready, execute the requested aria2 operations.
### Example Workflow
**User:** "download http://example.com/file.zip"
**You execute:**
```bash
# 1. Check configuration
python3 scripts/config_loader.py test
```
If test passes:
```bash
# 2. Execute download command
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
```
**You respond:** "✓ Download started! GID: 2089b05ecca3d829"
If test fails:
```
Configuration not ready. Please initialize:
1. Run: python3 scripts/config_loader.py init --user
2. Edit ~/.config/aria2-skill/config.json with your aria2 server details
3. Run: python3 scripts/config_loader.py test (to verify)
```
## Documentation Structure
**For detailed execution instructions, see:**
- **[references/execution-guide.md](references/execution-guide.md)** - Complete guide for AI agents with:
- Command mapping table (user intent → script call)
- Parameter formatting rules
- Step-by-step examples
- Common mistakes to avoid
- Response formatting guidelines
**For aria2 method reference, see:**
- **[references/aria2-methods.md](references/aria2-methods.md)** - Detailed aria2 RPC method documentation
## Common Commands Quick Reference
| User Intent | Command Example |
|-------------|----------------|
| Download a file | `python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'` |
| Check status | `python3 scripts/rpc_client.py aria2.tellStatus <GID>` |
| List active downloads | `python3 scripts/rpc_client.py aria2.tellActive` |
| List stopped downloads | `python3 scripts/rpc_client.py aria2.tellStopped 0 100` |
| Pause download | `python3 scripts/rpc_client.py aria2.pause <GID>` |
| Resume download | `python3 scripts/rpc_client.py aria2.unpause <GID>` |
| Show statistics | `python3 scripts/rpc_client.py aria2.getGlobalStat` |
| Show version | `python3 scripts/rpc_client.py aria2.getVersion` |
| Purge results | `python3 scripts/rpc_client.py aria2.purgeDownloadResult` |
For detailed usage and more commands, see [execution-guide.md](references/execution-guide.md).
## Available Scripts
- `scripts/rpc_client.py` - Main interface for RPC calls
- `scripts/examples/list-downloads.py` - Formatted download list
- `scripts/examples/pause-all.py` - Pause all downloads
- `scripts/examples/add-torrent.py` - Add torrent downloads
- `scripts/examples/monitor-downloads.py` - Real-time monitoring
- `scripts/examples/set-options.py` - Modify options
## Configuration
Scripts automatically load configuration from multiple sources with the following priority (highest to lowest):
### Configuration Priority
1. **Environment Variables** (highest priority - temporary override)
- `ARIA2_RPC_HOST`, `ARIA2_RPC_PORT`, `ARIA2_RPC_PATH`, etc.
- Best for: CI/CD pipelines, temporary overrides, testing
- **Note**: For reference only. Agents should use config files instead.
2. **Skill Directory Config** (project-specific configuration)
- Location: `skills/aria2-json-rpc/config.json`
- Best for: Project-specific settings, local testing, development
- ⚠️ **Warning**: Lost when running `npx skills add` to update the skill
3. **User Config Directory** (global fallback, update-safe) 🆕
- Location: `~/.config/aria2-skill/config.json`
- Best for: Personal default settings across all projects
- ✅ **Safe**: Survives skill updates via `npx skills add`
4. **Defaults** (localhost:6800)
- Zero-configuration fallback for local development
### Configuration Options
- **host**: Hostname or IP address (default: `localhost`)
- **port**: Port number (default: `6800`)
- **path**: URL path (default: `null`). Set to `/jsonrpc` for standard aria2, or custom path for reverse proxy
- **secret**: RPC secret token (default: `null`)
- **secure**: Use HTTPS instead of HTTP (default: `false`)
- **timeout**: Request timeout in milliseconds (default: `30000`)
### Quick Setup (For AI Agents)
**IMPORTANT**: Always use Python scripts to manage configuration. Do NOT use shell commands directly.
**Step 1: Check current configuration status**
```bash
python3 scripts/config_loader.py show
```
**Step 2: Initialize configuration if needed**
User config (recommended - survives updates):
```bash
python3 scripts/config_loader.py init --user
```
Local config (project-specific):
```bash
python3 scripts/config_loader.py init --local
```
**Step 3: Guide user to edit the config file**
After initialization, the tool will display the config file path. Instruct user to edit it with their aria2 server details (host, port, secret, etc.).
**Step 4: Verify configuration**
```bash
python3 scripts/config_loader.py test
```
Example config file content:
```json
{
"host": "localhost",
"port": 6800,
"secret": "your-secret-token",
"secure": false,
"timeout": 30000
}
```
### Configuration Management (For AI Agents)
**Available Python scripts for configuration management:**
```bash
# Check current configuration and source
python3 scripts/config_loader.py show
# Initialize user config (recommended - update-safe)
python3 scripts/config_loader.py init --user
# Initialize local config (project-specific)
python3 scripts/config_loader.py init --local
# Test connection to aria2 server
python3 scripts/config_loader.py test
```
**Agent Workflow for Configuration Setup:**
1. **Check if config exists**: Run `python3 scripts/config_loader.py show`
2. **If config missing or invalid**: Guide user to run `python3 scripts/config_loader.py init --user`
3. **User edits config**: Tell user the file path and required fields (host, port, secret)
4. **Verify setup**: Run `python3 scripts/config_loader.py test`
5. **Proceed with operations**: Once test passes, execute user's aria2 commands
### Advanced Configuration
**Reverse Proxy Setup:**
For reverse proxy setups like `https://example.com:443/jsonrpc`, the config file should contain:
```json
{
"host": "example.com",
"port": 443,
"path": "/jsonrpc",
"secret": "your-secret-token",
"secure": true
}
```
**Environment Variables (for reference only):**
Configuration can also be overridden via environment variables:
- `ARIA2_RPC_HOST`: Hostname
- `ARIA2_RPC_PORT`: Port number
- `ARIA2_RPC_PATH`: URL path
- `ARIA2_RPC_SECRET`: Secret token
- `ARIA2_RPC_SECURE`: "true" or "false"
Note: Use Python scripts for configuration management. Environment variables are documented here for reference only.
## Key Principles (For AI Agents)
1. **Never** construct JSON-RPC requests manually
2. **Always** call Python scripts via Bash tool using `python3` (not `python`)
3. **Always** check configuration before executing commands:
- Run `python3 scripts/config_loader.py test` first
- If test fails, guide user through initialization
4. **Never** run raw shell commands (mkdir, cat, export, etc.) directly
- Use Python scripts: `config_loader.py init`, `config_loader.py show`, etc.
5. **Parse** script output and format for users
6. **Refer to** execution-guide.md when unsure
## Supported Operations
### Download Management
- Add downloads (HTTP/FTP/Magnet/Torrent/Metalink)
- Pause/resume (individual or all)
- Remove downloads
- Add with custom options
### Monitoring
- Check download status
- List active/waiting/stopped downloads
- Get global statistics
- Real-time monitoring
### Configuration
- Get/change download options
- Get/change global options
- Query aria2 version
- List available methods
### Maintenance
- Purge download results
- Remove specific results
## Need Help?
- **Execution details:** [references/execution-guide.md](references/execution-guide.md)
- **Method reference:** [references/aria2-methods.md](references/aria2-methods.md)
- **Troubleshooting:** [references/troubleshooting.md](references/troubleshooting.md)
- **aria2 official docs:** https://aria2.github.io/
---
## Referenced Files
> The following files are referenced in this skill and included for context.
### references/execution-guide.md
```markdown
# Execution Guide for AI Agents
This guide provides detailed instructions for AI agents on how to execute aria2-json-rpc skill commands.
## Core Principles
**NEVER manually construct JSON-RPC requests. ALWAYS use the provided Python scripts.**
**⚠️ CRITICAL: Use `python3` command, NOT `python`**
- On macOS, the `python` symlink doesn't exist by default
- Always use `python3` for cross-platform compatibility
- All examples in this guide use `python3`
## Execution Workflow
1. **Parse** the user's natural language command to identify intent
2. **Map** the intent to the appropriate script and method
3. **Execute** the script using the Bash tool with proper parameters
4. **Format** the output in a user-friendly way
## Available Scripts
### Primary Script
- `scripts/rpc_client.py` - Direct RPC method calls (main interface)
### Helper Scripts
- `scripts/command_mapper.py` - Parse natural language to RPC methods
- `scripts/examples/list-downloads.py` - List all downloads with formatting
- `scripts/examples/pause-all.py` - Pause all active downloads
- `scripts/examples/monitor-downloads.py` - Real-time monitoring
- `scripts/examples/add-torrent.py` - Add torrent downloads
- `scripts/examples/set-options.py` - Modify download options
## Command Mapping Table
| User Intent | Script Call |
|-------------|-------------|
| Download a file | `python3 scripts/rpc_client.py aria2.addUri '["{URL}"]'` |
| Download multiple files separately | Call `python3 scripts/rpc_client.py aria2.addUri '["URL"]'` once per file |
| Check download status | `python3 scripts/rpc_client.py aria2.tellStatus {GID}` |
| Check if download completed | `python3 scripts/rpc_client.py aria2.tellStatus {GID}` (check status="complete") |
| Pause download | `python3 scripts/rpc_client.py aria2.pause {GID}` |
| Pause all downloads | `python3 scripts/rpc_client.py aria2.pauseAll` |
| Resume download | `python3 scripts/rpc_client.py aria2.unpause {GID}` |
| Resume all downloads | `python3 scripts/rpc_client.py aria2.unpauseAll` |
| Remove active download | `python3 scripts/rpc_client.py aria2.remove {GID}` |
| Remove completed download | `python3 scripts/rpc_client.py aria2.removeDownloadResult {GID}` |
| List active downloads | `python3 scripts/rpc_client.py aria2.tellActive` |
| List waiting downloads | `python3 scripts/rpc_client.py aria2.tellWaiting 0 100` |
| List stopped downloads | `python3 scripts/rpc_client.py aria2.tellStopped 0 100` |
| Get last 10 completed downloads | `python3 scripts/rpc_client.py aria2.tellStopped -1 10` |
| Show global stats | `python3 scripts/rpc_client.py aria2.getGlobalStat` |
| Show aria2 version | `python3 scripts/rpc_client.py aria2.getVersion` |
| List all available RPC methods | `python3 scripts/rpc_client.py system.listMethods` |
| Get download options | `python3 scripts/rpc_client.py aria2.getOption {GID}` |
| Get global options | `python3 scripts/rpc_client.py aria2.getGlobalOption` |
| Change download speed limit | `python3 scripts/rpc_client.py aria2.changeOption {GID} '{"max-download-limit":"1048576"}'` |
| Purge download results | `python3 scripts/rpc_client.py aria2.purgeDownloadResult` |
## Parameter Formatting
### Pattern 1: No Parameters
```bash
python3 scripts/rpc_client.py aria2.getGlobalStat
python3 scripts/rpc_client.py aria2.pauseAll
python3 scripts/rpc_client.py aria2.getVersion
```
### Pattern 2: Single String (GID)
```bash
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829
python3 scripts/rpc_client.py aria2.pause 2089b05ecca3d829
python3 scripts/rpc_client.py aria2.remove 2089b05ecca3d829
```
### Pattern 3: Array of Strings (URLs)
```bash
# Single URL
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
# Multiple URLs
python3 scripts/rpc_client.py aria2.addUri '["http://url1.com", "http://url2.com"]'
```
### Pattern 4: Multiple Parameters (Numbers)
```bash
python3 scripts/rpc_client.py aria2.tellWaiting 0 100
python3 scripts/rpc_client.py aria2.tellStopped 0 50
```
### Pattern 5: Helper Scripts
```bash
python3 scripts/examples/list-downloads.py
python3 scripts/examples/pause-all.py
python3 scripts/examples/add-torrent.py /path/to/file.torrent
```
## Step-by-Step Execution Examples
### Example 1: Download a File
**User Command:** "Please download http://example.com/file.zip"
**Thought Process:**
1. User wants to download → use `aria2.addUri`
2. Need to pass URL as array parameter
3. Call rpc_client.py with proper formatting using `python3`
**Execute:**
```bash
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
```
**Parse Output:** Extract GID from script output (e.g., "2089b05ecca3d829")
**Response:**
```
✓ Download started successfully!
GID: 2089b05ecca3d829
You can check progress with: "show status for GID 2089b05ecca3d829"
```
### Example 2: Check Download Status
**User Command:** "What's the status of GID 2089b05ecca3d829?"
**Thought Process:**
1. User wants status → use `aria2.tellStatus`
2. GID is the parameter
3. Parse JSON output and format nicely
**Execute:**
```bash
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829
```
**Parse Output:** Script returns JSON with fields like:
- `status`: "active", "paused", "complete", etc.
- `completedLength`: bytes downloaded
- `totalLength`: total file size
- `downloadSpeed`: current speed
**Response:**
```
Download Status:
- Status: active
- Progress: 45.2 MB / 100 MB (45%)
- Speed: 2.3 MB/s
- ETA: ~2 minutes
```
### Example 3: List All Downloads
**User Command:** "Show me what's downloading"
**Thought Process:**
1. User wants overview → use helper script for nice formatting
2. `list-downloads.py` shows active, waiting, and stopped
**Execute:**
```bash
python3 scripts/examples/list-downloads.py
```
**Response:** Summarize the output, for example:
```
Current Downloads:
Active (2):
- ubuntu-20.04.iso: 45% complete, 2.3 MB/s
- archive.zip: 78% complete, 1.5 MB/s
Waiting (1):
- movie.mp4: queued
Stopped (3):
- file1.zip: completed
- file2.tar.gz: completed
- file3.pdf: error
```
## Common Mistakes to Avoid
### ❌ WRONG: Manually construct JSON-RPC
```bash
# DON'T do this!
curl -X POST http://localhost:6800/jsonrpc \
-d '{"jsonrpc":"2.0","method":"aria2.addUri",...}'
# DON'T do this!
echo '{"jsonrpc": "2.0", "method": "aria2.addUri", ...}'
```
### ✅ CORRECT: Use Python scripts
```bash
# DO this!
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
```
### ❌ WRONG: Try to import aria2
```python
# DON'T do this!
import aria2 # aria2 is not a Python library!
```
### ✅ CORRECT: Call scripts via subprocess
```python
# DO this if needed!
import subprocess
result = subprocess.run(
["python3", "scripts/rpc_client.py", "aria2.getGlobalStat"],
capture_output=True, text=True
)
```
## Response Formatting Guidelines
### For addUri (download started)
```
✓ Download started successfully!
GID: {gid}
```
### For tellStatus (download progress)
```
Status: {status}
Progress: {completed}/{total} ({percentage}%)
Speed: {speed}
```
### For pause/unpause operations
```
✓ Download {paused/resumed}
GID: {gid}
```
### For batch operations (pauseAll, unpauseAll)
```
✓ All downloads {paused/resumed}
```
---
## Data Formatting for Agents
### Converting aria2 Data to Human-Readable Format
aria2 returns numbers as strings. Agents should convert these for better user experience.
#### Byte Conversion
```python
def format_bytes(bytes_str):
"""Convert byte string to human-readable format.
Examples:
"1024" -> "1.0 KB"
"1048576" -> "1.0 MB"
"22434" -> "21.9 KB"
"""
bytes_val = int(bytes_str)
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if bytes_val < 1024.0:
return f"{bytes_val:.1f} {unit}"
bytes_val /= 1024.0
return f"{bytes_val:.1f} PB"
# Usage with aria2 response
status = client.tell_status(gid)
total = format_bytes(status['totalLength'])
completed = format_bytes(status['completedLength'])
print(f"Progress: {completed} / {total}")
```
#### Percentage Calculation
```python
def calculate_progress(status):
"""Calculate download progress percentage.
Returns:
Float percentage (0-100) or 0 if total is unknown
"""
completed = int(status.get('completedLength', 0))
total = int(status.get('totalLength', 0))
if total == 0:
return 0.0
return (completed / total) * 100
# Usage
status = client.tell_status(gid)
progress = calculate_progress(status)
print(f"Progress: {progress:.1f}%")
```
#### Speed Formatting
```python
def format_speed(speed_str):
"""Convert speed string (bytes/sec) to human-readable format.
Examples:
"0" -> "0 B/s"
"102400" -> "100.0 KB/s"
"2097152" -> "2.0 MB/s"
"""
speed = int(speed_str)
if speed == 0:
return "0 B/s"
for unit in ['B/s', 'KB/s', 'MB/s', 'GB/s']:
if speed < 1024.0:
return f"{speed:.1f} {unit}"
speed /= 1024.0
return f"{speed:.1f} TB/s"
# Usage
status = client.tell_status(gid)
dl_speed = format_speed(status['downloadSpeed'])
ul_speed = format_speed(status['uploadSpeed'])
print(f"Download: {dl_speed}, Upload: {ul_speed}")
```
#### ETA Calculation
```python
def calculate_eta(status):
"""Calculate estimated time to completion.
Returns:
String like "2m 34s" or "Unknown" if can't calculate
"""
completed = int(status.get('completedLength', 0))
total = int(status.get('totalLength', 0))
speed = int(status.get('downloadSpeed', 0))
if speed == 0 or total == 0 or completed >= total:
return "Unknown"
remaining_bytes = total - completed
eta_seconds = remaining_bytes / speed
if eta_seconds < 60:
return f"{int(eta_seconds)}s"
elif eta_seconds < 3600:
minutes = int(eta_seconds / 60)
seconds = int(eta_seconds % 60)
return f"{minutes}m {seconds}s"
else:
hours = int(eta_seconds / 3600)
minutes = int((eta_seconds % 3600) / 60)
return f"{hours}h {minutes}m"
# Usage
status = client.tell_status(gid)
eta = calculate_eta(status)
print(f"ETA: {eta}")
```
#### Complete Status Formatting Example
```python
def format_download_status(status):
"""Format complete download status for user display."""
gid = status['gid']
state = status['status']
# Basic info
result = [
f"GID: {gid}",
f"Status: {state}"
]
# Progress info (if applicable)
if state in ['active', 'paused', 'waiting']:
total = format_bytes(status['totalLength'])
completed = format_bytes(status['completedLength'])
progress = calculate_progress(status)
result.append(f"Progress: {completed} / {total} ({progress:.1f}%)")
# Speed info (if active)
if state == 'active':
dl_speed = format_speed(status['downloadSpeed'])
result.append(f"Speed: {dl_speed}")
eta = calculate_eta(status)
if eta != "Unknown":
result.append(f"ETA: {eta}")
# Error info (if error)
if state == 'error':
error_code = status.get('errorCode', 'Unknown')
error_msg = status.get('errorMessage', 'No message')
result.append(f"Error: [{error_code}] {error_msg}")
# Files info
files = status.get('files', [])
if files:
result.append(f"Files: {len(files)}")
for file in files[:3]: # Show first 3 files
path = file.get('path', 'Unknown')
result.append(f" - {path}")
if len(files) > 3:
result.append(f" ... and {len(files) - 3} more")
return "\n".join(result)
# Usage in agent response
status = client.tell_status(gid)
formatted = format_download_status(status)
print(formatted)
```
### Quick Reference: Common Field Conversions
| aria2 Field | Type | Conversion Needed |
|-------------|------|-------------------|
| `totalLength` | string (bytes) | → Human-readable size |
| `completedLength` | string (bytes) | → Human-readable size |
| `downloadSpeed` | string (bytes/sec) | → Speed with unit |
| `uploadSpeed` | string (bytes/sec) | → Speed with unit |
| `numActive` | string (number) | → Integer display |
| `numWaiting` | string (number) | → Integer display |
| `status` | string | → Capitalize or use icons |
| `errorCode` | string | → Error message lookup |
---
## Troubleshooting
For detailed troubleshooting information, see **[troubleshooting.md](troubleshooting.md)**.
### Quick Troubleshooting
**Script not found:** Change to skill directory or use absolute path
**Connection refused:** Check if aria2 is running with `--enable-rpc`
**Parameter error:** Use single quotes around JSON: `'["url"]'`
**GID not found:** Check stopped downloads with `aria2.tellStopped 0 100`
## Configuration
Scripts automatically load configuration from:
1. Environment variables (highest priority)
2. `config.json` in skill directory
3. Defaults (localhost:6800)
You don't need to set configuration manually - scripts handle it automatically.
## See Also
- [aria2-methods.md](aria2-methods.md) - Detailed aria2 RPC method reference
- [Official aria2 documentation](https://aria2.github.io/) - aria2 daemon documentation
```
### references/aria2-methods.md
```markdown
# Aria2 RPC Methods Reference
Pure technical reference for all aria2 RPC methods. For execution examples, see [execution-guide.md](execution-guide.md).
## Method Index
### Download Operations
- [aria2.addUri](#aria2adduri) - Add download from URLs
- [aria2.addTorrent](#aria2addtorrent) - Add download from torrent file
- [aria2.addMetalink](#aria2addmetalink) - Add download from metalink file
- [aria2.remove](#aria2remove) - Stop and remove download
### Control Operations
- [aria2.pause](#aria2pause) - Pause a download
- [aria2.pauseAll](#aria2pauseall) - Pause all downloads
- [aria2.unpause](#aria2unpause) - Resume a paused download
- [aria2.unpauseAll](#aria2unpauseall) - Resume all paused downloads
### Monitoring Operations
- [aria2.tellStatus](#aria2tellstatus) - Get download status
- [aria2.tellActive](#aria2tellactive) - List active downloads
- [aria2.tellWaiting](#aria2tellwaiting) - List waiting downloads
- [aria2.tellStopped](#aria2tellstopped) - List stopped downloads
- [aria2.getGlobalStat](#aria2getglobalstat) - Get global statistics
### Configuration Operations
- [aria2.getOption](#aria2getoption) - Get download options
- [aria2.changeOption](#aria2changeoption) - Change download options
- [aria2.getGlobalOption](#aria2getglobaloption) - Get global options
- [aria2.changeGlobalOption](#aria2changeglobaloption) - Change global options
### Maintenance Operations
- [aria2.purgeDownloadResult](#aria2purgedownloadresult) - Remove all stopped results
- [aria2.removeDownloadResult](#aria2removedownloadresult) - Remove specific result
### System Operations
- [aria2.getVersion](#aria2getversion) - Get aria2 version info
- [system.listMethods](#systemlistmethods) - List all available methods
- [system.multicall](#systemmulticall) - Execute multiple methods
---
## Download Operations
### aria2.addUri
Add a new download from HTTP/HTTPS/FTP/SFTP/Magnet URIs.
**Parameters:**
- `uris` (array of strings, required): URIs pointing to the same resource
- **Important:** Multiple URIs are treated as backup/mirror sources for the SAME download
- aria2 will try each URI in order if one fails (fallback mechanism)
- To download multiple separate files, call `aria2.addUri` multiple times with one URL each
- Example: `["http://mirror1.com/file.zip", "http://mirror2.com/file.zip"]` creates ONE download with 2 sources
- `options` (object, optional): Download options (see Options section)
- `position` (integer, optional): Position in download queue (0-based)
**Returns:** `string` - GID of the newly created download (16-character hex)
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
python3 scripts/rpc_client.py aria2.addUri '["http://mirror1.com/file.zip", "http://mirror2.com/file.zip"]'
```
### aria2.addTorrent
Add a BitTorrent download.
**Parameters:**
- `torrent` (string, required): Base64-encoded torrent file content
- `uris` (array of strings, optional): Web seed URIs
- `options` (object, optional): Download options
- `position` (integer, optional): Position in download queue
**Returns:** `string` - GID of the newly created download
**Script call:**
```bash
python3 scripts/examples/add-torrent.py /path/to/file.torrent
```
### aria2.addMetalink
Add downloads from a Metalink file.
**Parameters:**
- `metalink` (string, required): Base64-encoded metalink file content
- `options` (object, optional): Download options
- `position` (integer, optional): Position in download queue
**Returns:** `array of strings` - GIDs of newly created downloads
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.addMetalink '<base64-encoded-metalink>'
```
### aria2.remove
Remove a download. If the download is active, it will be stopped first.
**Parameters:**
- `gid` (string, required): GID of the download to remove
**Returns:** `string` - GID of the removed download
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.remove 2089b05ecca3d829
```
**When to use:**
- Download is in `active`, `waiting`, or `paused` state
- You want to stop AND remove the download
- Will fail if download is already `complete`, `error`, or `removed`
**See also:** Use [aria2.removeDownloadResult](#aria2removedownloadresult) for completed/error downloads.
---
## Control Operations
### aria2.pause
Pause an active download.
**Parameters:**
- `gid` (string, required): GID of the download to pause
**Returns:** `string` - GID of the paused download
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.pause 2089b05ecca3d829
```
**Notes:**
- Download must be in `active` state
- Returns error if download is already paused or completed
### aria2.pauseAll
Pause all active downloads.
**Parameters:** None
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.pauseAll
```
### aria2.unpause
Resume a paused download.
**Parameters:**
- `gid` (string, required): GID of the download to resume
**Returns:** `string` - GID of the resumed download
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.unpause 2089b05ecca3d829
```
### aria2.unpauseAll
Resume all paused downloads.
**Parameters:** None
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.unpauseAll
```
---
## Monitoring Operations
### aria2.tellStatus
Get detailed status of a download.
**Parameters:**
- `gid` (string, required): GID of the download
- `keys` (array of strings, optional): Specific keys to retrieve (returns all if omitted)
**Returns:** `object` - Download status with fields:
- `gid` - Download GID
- `status` - "active", "waiting", "paused", "error", "complete", "removed"
- `totalLength` - Total size in bytes (string)
- `completedLength` - Downloaded size in bytes (string)
- `downloadSpeed` - Download speed in bytes/sec (string)
- `uploadSpeed` - Upload speed in bytes/sec (string)
- `files` - Array of file information
- `errorCode` - Error code if status is "error"
- `errorMessage` - Error message if status is "error"
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829 '["status", "totalLength", "completedLength"]'
```
### aria2.tellActive
Get list of all active downloads.
**Parameters:**
- `keys` (array of strings, optional): Specific keys to retrieve for each download
**Returns:** `array of objects` - Array of download status objects (same structure as tellStatus)
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.tellActive
python3 scripts/rpc_client.py aria2.tellActive '["gid", "status", "downloadSpeed"]'
```
### aria2.tellWaiting
Get list of downloads in the waiting queue.
**Parameters:**
- `offset` (integer, required): Starting position in queue (0-based)
- `num` (integer, required): Number of downloads to retrieve
- `keys` (array of strings, optional): Specific keys to retrieve
**Returns:** `array of objects` - Array of download status objects
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.tellWaiting 0 100
python3 scripts/rpc_client.py aria2.tellWaiting 0 10 '["gid", "status"]'
```
### aria2.tellStopped
Get list of stopped downloads (completed, error, or removed).
**Parameters:**
- `offset` (integer, required): Starting position (0-based)
- `num` (integer, required): Number of downloads to retrieve
- `keys` (array of strings, optional): Specific keys to retrieve
**Returns:** `array of objects` - Array of download status objects
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.tellStopped 0 100
python3 scripts/rpc_client.py aria2.tellStopped -1 10 # Get last 10
```
---
### Pagination Best Practices
Both `aria2.tellWaiting` and `aria2.tellStopped` use pagination parameters:
**Common Use Cases:**
| Scenario | offset | num | Example |
|----------|--------|-----|---------|
| Get first page | `0` | `100` | `python3 scripts/rpc_client.py aria2.tellStopped 0 100` |
| Get all items (small queue) | `0` | `1000` | Get up to 1000 items at once |
| Get recent items only | `-1` | `10` | Last 10 stopped downloads |
| Paginate large results | `0`, `100`, `200`, ... | `100` | Loop with increasing offset |
| Quick status check | `0` | `10` | First 10 items for fast response |
**Performance Tips:**
- **Small `num` (10-20)**: Faster response, good for quick checks or UI updates
- **Large `num` (100-1000)**: Fewer requests, good for batch processing
- **Negative offset**: `-1` means start from the end (most recent)
**Example: Get All Stopped Downloads**
```bash
# Start with offset 0
python3 scripts/rpc_client.py aria2.tellStopped 0 100
# If 100 items returned, get next batch
python3 scripts/rpc_client.py aria2.tellStopped 100 100
# Continue until empty array returned
```
---
## Configuration Operations
### aria2.getOption
Get options for a specific download.
**Parameters:**
- `gid` (string, required): GID of the download
**Returns:** `object` - Download options as key-value pairs
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.getOption 2089b05ecca3d829
```
### aria2.changeOption
Change options for an active download.
**Parameters:**
- `gid` (string, required): GID of the download
- `options` (object, required): Options to change
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.changeOption 2089b05ecca3d829 '{"max-download-limit":"1048576"}'
```
**Common options:**
- `max-download-limit` - Max download speed in bytes/sec
- `max-upload-limit` - Max upload speed in bytes/sec (for BitTorrent)
**Note:** Not all options can be changed after download starts. See aria2 documentation for changeable options.
### aria2.getGlobalOption
Get global aria2 options.
**Parameters:** None
**Returns:** `object` - Global options as key-value pairs
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.getGlobalOption
```
### aria2.changeGlobalOption
Change global aria2 options.
**Parameters:**
- `options` (object, required): Options to change
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.changeGlobalOption '{"max-concurrent-downloads":"5"}'
```
**Common options:**
- `max-concurrent-downloads` - Maximum number of parallel downloads
- `max-overall-download-limit` - Overall download speed limit
- `max-overall-upload-limit` - Overall upload speed limit
---
## Maintenance Operations
### aria2.purgeDownloadResult
Remove completed/error/removed downloads from memory to free up resources.
**Parameters:** None
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.purgeDownloadResult
```
**Note:** This only removes download results from memory, not the downloaded files.
### aria2.removeDownloadResult
Remove a specific download result from memory.
**Parameters:**
- `gid` (string, required): GID of the download result to remove
**Returns:** `string` - "OK" on success
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.removeDownloadResult 2089b05ecca3d829
```
**When to use:**
- Download is in `complete`, `error`, or `removed` state
- You want to clear the download record from memory
- Does not affect the downloaded files on disk
**See also:** Use [aria2.remove](#aria2remove) for active/waiting/paused downloads.
---
## Download Removal Guide
### Choosing the Right Removal Method
| Download Status | Use Method | What it Does |
|----------------|------------|--------------|
| `active` | `aria2.remove` | Stops download, then removes from list |
| `waiting` | `aria2.remove` | Cancels queued download, removes from list |
| `paused` | `aria2.remove` | Removes paused download from list |
| `complete` | `aria2.removeDownloadResult` | Clears completed record from memory |
| `error` | `aria2.removeDownloadResult` | Clears failed record from memory |
| `removed` | `aria2.removeDownloadResult` | Clears already-removed record from memory |
**Tip:** If you're unsure, try `aria2.remove` first. If it returns an error like "GID is not active", use `aria2.removeDownloadResult` instead.
**Note:** Neither method deletes downloaded files from disk. They only affect aria2's internal download list.
---
## System Operations
### aria2.getVersion
Get aria2 version and enabled feature information.
**Parameters:** None
**Returns:** `object` with fields:
- `version` - aria2 version string
- `enabledFeatures` - Array of enabled features (e.g., "BitTorrent", "Metalink", "Async DNS")
**Script call:**
```bash
python3 scripts/rpc_client.py aria2.getVersion
```
### system.listMethods
List all available RPC methods.
**Parameters:** None
**Returns:** `array of strings` - Method names
**Script call:**
```bash
python3 scripts/rpc_client.py system.listMethods
```
**Note:** This method does not require authentication.
### system.multicall
Execute multiple RPC methods in a single request (batch operation).
**Parameters:**
- `calls` (array of objects, required): Each object has:
- `methodName` (string): Method to call
- `params` (array): Parameters for the method
**Returns:** `array` - Results for each method call in order
**Example (Python):**
```python3
from scripts.rpc_client import Aria2RpcClient
from scripts.config_loader import Aria2Config
config = Aria2Config().load()
client = Aria2RpcClient(config)
calls = [
{"methodName": "aria2.tellStatus", "params": ["2089b05ecca3d829"]},
{"methodName": "aria2.getGlobalStat", "params": []},
]
results = client.multicall(calls)
```
---
## Common Option Keys
### Download Options
Options that can be set when adding or changing downloads:
- `dir` - Download directory
- `out` - Output filename
- `max-download-limit` - Speed limit in bytes/sec (0 = unlimited)
- `max-upload-limit` - Upload speed limit for torrents
- `split` - Number of connections per server (1-16)
- `max-connection-per-server` - Max connections per server (1-16)
- `min-split-size` - Minimum size for split downloading (1M-1024M)
- `lowest-speed-limit` - Minimum speed threshold (bytes/sec)
- `referer` - HTTP Referer header
- `user-agent` - HTTP User-Agent header
- `header` - Additional HTTP headers (array of "Header: Value")
### Global Options
Options affecting all downloads:
- `max-concurrent-downloads` - Max parallel downloads (1-unlimited)
- `max-overall-download-limit` - Total download speed limit
- `max-overall-upload-limit` - Total upload speed limit
- `download-result` - How long to keep completed download info ("default", "full", "hide")
---
## Status Values
### Download Status
- `active` - Currently downloading
- `waiting` - In queue, waiting to start
- `paused` - Paused by user
- `error` - Download failed (see errorCode/errorMessage)
- `complete` - Download finished successfully
- `removed` - Removed by user
### Error Codes
- `0` - All downloads successful
- `1` - Unknown error
- `2` - Timeout
- `3` - Resource not found
- `4` - Too many redirects
- `5` - Not enough disk space
- `7` - Duplicated file or duplicate GID
- `8` - Resume failed (cannot resume)
- `9` - No such file or directory
- `19` - File I/O error
- `24` - HTTP/FTP protocol error
For complete error code reference, see aria2 official documentation.
---
## GID Format
GID (Global Identifier) is a 16-character hexadecimal string that uniquely identifies a download.
**Format:** `[0-9a-f]{16}`
**Example:** `2089b05ecca3d829`
**Usage:** GID is returned when adding downloads and used to reference downloads in all other operations.
---
## See Also
- [execution-guide.md](execution-guide.md) - Detailed execution guide for AI agents
- [troubleshooting.md](troubleshooting.md) - Common issues and solutions
- [Official aria2 documentation](https://aria2.github.io/manual/en/html/aria2c.html) - Complete aria2 reference
```
### references/troubleshooting.md
```markdown
# Troubleshooting Guide
Common issues and solutions when using the aria2-json-rpc skill.
**⚠️ NOTE: All commands use `python3`, not `python`** (especially important on macOS where `python` symlink doesn't exist)
## Script Execution Errors
### Python Command Not Found
**Error:** `command not found: python`
**Cause:** On macOS and some Linux systems, only `python3` is available
**Solution:**
```bash
# Always use python3
python3 scripts/rpc_client.py aria2.getVersion
# NOT python (this will fail on macOS)
```
### File Not Found
**Error:** `File not found: scripts/rpc_client.py`
**Causes:**
- Not in the correct directory
- Script path is incorrect
- Skill not properly installed
**Solutions:**
1. Change to the skill directory first:
```bash
cd /path/to/skills/aria2-json-rpc
python3 scripts/rpc_client.py aria2.getVersion
```
2. Or use absolute path:
```bash
python3 /full/path/to/skills/aria2-json-rpc/scripts/rpc_client.py aria2.getVersion
```
3. Verify the script exists:
```bash
ls -l skills/aria2-json-rpc/scripts/rpc_client.py
```
### Permission Denied
**Error:** `Permission denied: scripts/rpc_client.py`
**Solution:**
```bash
chmod +x scripts/rpc_client.py
# Or run with python3 explicitly
python3 scripts/rpc_client.py aria2.getVersion
```
## Connection Errors
### Cannot Connect to aria2 RPC Server
**Error:** `Cannot connect to aria2 RPC server` or `Connection refused`
**Causes:**
- aria2 daemon is not running
- Wrong host/port configuration
- Firewall blocking connection
- aria2 not started with RPC enabled
**Solutions:**
1. **Check if aria2 is running:**
```bash
ps aux | grep aria2c
# Or on macOS
pgrep -fl aria2c
```
2. **Start aria2 with RPC enabled:**
```bash
aria2c --enable-rpc --rpc-listen-port=6800
```
3. **Test connection:**
```bash
python3 scripts/rpc_client.py aria2.getVersion
```
4. **Verify configuration:**
```bash
# Check what host/port the scripts are using
python3 scripts/config_loader.py
```
5. **Try with curl:**
```bash
curl -X POST http://localhost:6800/jsonrpc \
-d '{"jsonrpc":"2.0","id":"test","method":"aria2.getVersion","params":[]}'
```
### Connection Timeout
**Error:** `Connection timeout` or script hangs
**Causes:**
- aria2 server is slow or overloaded
- Network issues
- Firewall dropping packets
**Solutions:**
- Increase timeout in config.json:
```json
{
"timeout": 60000
}
```
- Check network connectivity
- Try a local connection first (localhost)
## Authentication Errors
### Authentication Failed
**Error:** `Authentication failed` or `Unauthorized`
**Causes:**
- Wrong or missing secret token
- Token mismatch between skill and aria2
**Solutions:**
1. **Check aria2's secret:**
```bash
# If you started aria2 with:
aria2c --enable-rpc --rpc-secret=your-secret-here
```
2. **Set the matching secret in environment:**
```bash
export ARIA2_RPC_SECRET=your-secret-here
```
3. **Or in config.json:**
```json
{
"secret": "your-secret-here"
}
```
4. **Verify configuration loaded correctly:**
```bash
python3 scripts/config_loader.py
# Should show: secret: ****** (hidden)
```
5. **Note:** `system.listMethods` doesn't require authentication - use it to test:
```bash
python3 scripts/rpc_client.py system.listMethods
```
## Parameter Errors
### Invalid JSON Parameter
**Error:** `Invalid JSON parameter` or `Parse error`
**Causes:**
- Incorrect JSON formatting
- Missing quotes around JSON in bash
- Wrong parameter structure
**Solutions:**
1. **Use single quotes around JSON arrays:**
```bash
# Correct
python3 scripts/rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
# Wrong - bash will interpret the quotes
python3 scripts/rpc_client.py aria2.addUri ["http://example.com/file.zip"]
```
2. **Don't forget the array brackets:**
```bash
# Correct
'["http://example.com/file.zip"]'
# Wrong - not an array
'"http://example.com/file.zip"'
```
3. **Escape quotes in complex JSON:**
```bash
python3 scripts/rpc_client.py aria2.changeOption 2089b05ecca3d829 '{"max-download-limit":"1M"}'
```
### Wrong Number of Parameters
**Error:** `Missing required parameter` or `Too many parameters`
**Solution:** Check the method signature in [aria2-methods.md](aria2-methods.md)
**Examples:**
```bash
# tellWaiting needs offset and num
python3 scripts/rpc_client.py aria2.tellWaiting 0 100
# tellStatus needs only GID
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829
# getGlobalStat needs no parameters
python3 scripts/rpc_client.py aria2.getGlobalStat
```
## Download Errors
### GID Not Found
**Error:** `GID not found` or code `1`
**Causes:**
- Download already completed and removed from memory
- Invalid GID format
- Download was purged
- Typo in GID
**Solutions:**
1. **Check GID format (16 hex characters):**
```bash
# Valid: 2089b05ecca3d829
# Invalid: 2089b05 (too short)
# Invalid: 2089b05ecca3d82g (contains 'g' - not hex)
```
2. **Search in stopped downloads:**
```bash
python3 scripts/rpc_client.py aria2.tellStopped 0 100
```
3. **List all current downloads:**
```bash
python3 scripts/examples/list-downloads.py
```
4. **Note:** aria2 automatically purges old results based on its configuration
### Download Not Active
**Error:** `Download not active` when trying to pause
**Causes:**
- Download is already paused
- Download is completed
- Download is in error state
**Solutions:**
1. **Check current status:**
```bash
python3 scripts/rpc_client.py aria2.tellStatus 2089b05ecca3d829
```
2. **Status field shows the state:**
- `active` - can be paused
- `paused` - can be resumed
- `complete` - finished, cannot pause
- `error` - failed, cannot pause
- `removed` - deleted, cannot pause
3. **Use appropriate command:**
- If paused: use `unpause` instead
- If complete: no action needed
- If error: check error message in status
### Invalid URI
**Error:** `Invalid URI` or `Unsupported scheme`
**Causes:**
- Malformed URL
- Unsupported protocol
- Server unreachable
**Solutions:**
1. **Check URL format:**
```bash
# Valid URLs
http://example.com/file.zip
https://example.com/file.zip
ftp://ftp.example.com/file.tar.gz
magnet:?xt=urn:btih:...
# Invalid
example.com/file.zip # Missing protocol
htp://example.com # Typo in protocol
```
2. **Verify aria2 supports the protocol:**
```bash
python3 scripts/rpc_client.py aria2.getVersion
# Check "enabledFeatures" array
```
3. **Test URL accessibility:**
```bash
curl -I http://example.com/file.zip
```
## Performance Issues
### Slow Downloads
**Symptoms:**
- Download speed slower than expected
- Frequent pauses/resumes
- High CPU usage
**Solutions:**
1. **Increase connections per server:**
```bash
python3 scripts/rpc_client.py aria2.changeOption 2089b05ecca3d829 \
'{"max-connection-per-server":"16"}'
```
2. **Enable split downloading:**
```bash
python3 scripts/rpc_client.py aria2.changeOption 2089b05ecca3d829 \
'{"split":"10"}'
```
3. **Adjust concurrent downloads:**
```bash
python3 scripts/rpc_client.py aria2.changeGlobalOption \
'{"max-concurrent-downloads":"3"}'
```
4. **Check disk I/O:**
- Slow disk can bottleneck downloads
- Change download directory to faster disk
### Script Runs Slowly
**Symptoms:**
- Scripts take long time to execute
- No error but delayed response
**Causes:**
- High network latency
- aria2 server is busy
- Default timeout too high
**Solutions:**
1. **Check aria2 server load:**
```bash
python3 scripts/rpc_client.py aria2.getGlobalStat
# Check numActive - too many concurrent downloads?
```
2. **Reduce timeout if on localhost:**
```json
{
"timeout": 5000
}
```
3. **Use helper scripts instead of multiple calls:**
```bash
# Instead of multiple tellStatus calls
python3 scripts/examples/list-downloads.py
```
## Configuration Issues
### Configuration Not Loading
**Error:** Config file exists but settings not applied
**Solutions:**
1. **Check JSON syntax:**
```bash
python3 -m json.tool config.json
```
2. **Verify file location:**
```bash
ls -l skills/aria2-json-rpc/config.json
```
3. **Check environment variables override:**
```bash
env | grep ARIA2_RPC
# Environment variables take priority over config.json
```
4. **Test configuration loading:**
```bash
python3 scripts/config_loader.py
```
### Secret Token Not Working
**Error:** Authentication fails despite correct token
**Solutions:**
1. **Check for whitespace in token:**
```bash
# Bad
export ARIA2_RPC_SECRET=" your-token " # Has spaces
# Good
export ARIA2_RPC_SECRET="your-token"
```
2. **Verify token in config.json has no special characters issue:**
```json
{
"secret": "plain-token-here"
}
```
3. **Restart aria2 after changing its secret:**
```bash
killall aria2c
aria2c --enable-rpc --rpc-secret=new-token
```
## Helper Script Issues
### Script Output Not as Expected
**Issue:** Helper scripts produce unexpected output
**Solutions:**
1. **Check script dependencies:**
```bash
python3 --version # Should be 3.6+
```
2. **Run with verbose mode if available:**
```bash
python3 scripts/examples/list-downloads.py --verbose
```
3. **Check for errors in stderr:**
```bash
python3 scripts/examples/list-downloads.py 2>&1 | grep -i error
```
### Import Errors
**Error:** `ModuleNotFoundError` or `ImportError`
**Solutions:**
1. **Verify Python path:**
```bash
# Scripts should be run from skill directory or with proper paths
cd skills/aria2-json-rpc
python3 scripts/examples/list-downloads.py
```
2. **Check script imports:**
```python
# Scripts use relative imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
```
## Getting Debug Information
### Enable Verbose Logging
Add debug output to help diagnose issues:
```python
# In rpc_client.py, temporarily add:
import logging
logging.basicConfig(level=logging.DEBUG)
```
### Capture Full Error Details
```bash
# Get full traceback
python3 scripts/rpc_client.py aria2.getGlobalStat 2>&1 | tee error.log
```
### Test aria2 Directly
```bash
# Bypass skill scripts to test aria2 itself
curl -X POST http://localhost:6800/jsonrpc \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": "test",
"method": "aria2.getVersion",
"params": []
}' | python3 -m json.tool
```
## Common Workflow Issues
### Downloads Don't Start
**Possible causes:**
1. aria2's max-concurrent-downloads reached
2. Download directory doesn't exist or no write permission
3. Not enough disk space
**Check:**
```bash
# Global stats
python3 scripts/rpc_client.py aria2.getGlobalStat
# Download options
python3 scripts/rpc_client.py aria2.getOption <GID>
```
### Cannot Resume Paused Download
**Possible causes:**
1. Download file was deleted
2. Original server no longer available
3. Session expired
**Solutions:**
- Check file still exists in download directory
- Try restarting the download with same URL
- Check aria2 error message in tellStatus
## Need More Help?
1. **Check aria2 official documentation:** https://aria2.github.io/
2. **Review method reference:** [aria2-methods.md](aria2-methods.md)
3. **Check execution guide:** [execution-guide.md](execution-guide.md)
4. **Enable aria2 logging:** Start aria2 with `--log=/path/to/aria2.log --log-level=debug`
```
### scripts/config_loader.py
```python
#!/usr/bin/env python3
"""
Configuration loader for aria2-json-rpc skill.
Supports three-tier configuration priority:
1. Environment variables (highest priority)
2. config.json file in skill directory
3. Interactive defaults (fallback)
Configuration schema:
{
"host": "localhost",
"port": 6800,
"secret": null,
"secure": false,
"timeout": 30000
}
"""
import json
import os
import sys
import urllib.request
import urllib.error
class ConfigurationError(Exception):
"""Raised when configuration is invalid or cannot be loaded."""
pass
class Aria2Config:
"""
Manages aria2 RPC configuration with multi-source loading.
Configuration priority (highest to lowest):
1. Environment variables (ARIA2_RPC_*)
2. Skill directory config (project-specific)
3. User config directory (global fallback, update-safe)
4. Defaults
"""
DEFAULT_CONFIG = {
"host": "localhost",
"port": 6800,
"path": None, # Optional: specify path like "/jsonrpc". If null, no path is appended.
"secret": None,
"secure": False,
"timeout": 30000,
}
# Environment variable names
ENV_PREFIX = "ARIA2_RPC_"
ENV_VARS = {
"host": "ARIA2_RPC_HOST",
"port": "ARIA2_RPC_PORT",
"path": "ARIA2_RPC_PATH",
"secret": "ARIA2_RPC_SECRET",
"secure": "ARIA2_RPC_SECURE",
"timeout": "ARIA2_RPC_TIMEOUT",
}
# User config directory (XDG standard)
USER_CONFIG_DIR = os.path.expanduser("~/.config/aria2-skill")
USER_CONFIG_FILE = os.path.join(USER_CONFIG_DIR, "config.json")
def __init__(self, config_path=None):
"""
Initialize configuration loader.
Args:
config_path (str, optional): Explicit path to config.json file.
If provided, only this path will be used.
Otherwise, searches multiple locations.
"""
self.explicit_config_path = config_path
self.config = self.DEFAULT_CONFIG.copy()
self._loaded = False
self._loaded_from = None # Track where config was loaded from
self._loaded_from_env = False # Track if env vars were used
def _get_skill_config_path(self):
"""Get the config.json path in the skill directory."""
script_dir = os.path.dirname(os.path.abspath(__file__))
skill_dir = os.path.dirname(script_dir)
return os.path.join(skill_dir, "config.json")
def _get_config_search_paths(self):
"""
Get list of config file paths to search (in priority order).
Priority (high to low):
1. Environment variables (handled separately in load())
2. Skill directory config (project-specific, may be lost on update)
3. User config directory (global fallback, update-safe)
4. Defaults (handled separately in load())
Returns:
list: List of (path, description) tuples
"""
if self.explicit_config_path:
return [(self.explicit_config_path, "explicit path")]
paths = [
(self._get_skill_config_path(), "skill directory"),
(self.USER_CONFIG_FILE, "user config directory"),
]
return paths
def load(self):
"""
Load configuration from all sources with priority resolution.
Priority: Environment Variables > Skill Directory > User Directory > Defaults
Returns:
dict: Loaded configuration
Raises:
ConfigurationError: If configuration is invalid
"""
# 1. Start with defaults
config = self.DEFAULT_CONFIG.copy()
self._loaded_from = "defaults"
# 2. Search config files (skill dir → user dir)
for config_path, description in self._get_config_search_paths():
if os.path.exists(config_path):
try:
file_config = self._load_from_file(config_path)
config.update(file_config)
self._loaded_from = config_path
break # Use first found config
except Exception as e:
# If explicit path was specified, raise the error
if self.explicit_config_path:
raise
# Otherwise, log warning and continue to next path
print(f"Warning: Failed to load config from {config_path}: {e}")
continue
# 3. Override with environment variables (highest priority)
env_config = self._load_from_env()
if env_config:
config.update(env_config)
self._loaded_from_env = True
# 4. Validate configuration
self._validate_config(config)
self.config = config
self._loaded = True
return config
def _load_from_file(self, path):
"""
Load configuration from JSON file.
Args:
path (str): Path to config.json
Returns:
dict: Configuration from file
Raises:
ConfigurationError: If file is invalid JSON or unreadable
"""
try:
with open(path, "r") as f:
file_config = json.load(f)
except json.JSONDecodeError as e:
raise ConfigurationError(
f"Invalid JSON in config file: {path}\n"
f"Error at line {e.lineno}, column {e.colno}: {e.msg}\n"
f"Example valid structure:\n"
f"{json.dumps(self.DEFAULT_CONFIG, indent=2)}"
)
except PermissionError:
raise ConfigurationError(
f"Permission denied reading config file: {path}\n"
f"Please check file permissions: chmod 644 {path}"
)
except Exception as e:
raise ConfigurationError(f"Error reading config file: {e}")
return file_config
def _load_from_env(self):
"""
Load configuration from environment variables.
Returns:
dict: Configuration from environment variables
"""
env_config = {}
for key, env_var in self.ENV_VARS.items():
value = os.environ.get(env_var)
if value is not None:
# Type conversion based on expected type
if key == "port" or key == "timeout":
try:
env_config[key] = int(value)
except ValueError:
print(
f"WARNING: Invalid {env_var}={value}, expected integer. Ignoring."
)
elif key == "secure":
env_config[key] = value.lower() in ("true", "1", "yes")
elif key == "secret":
# Empty string means no secret
env_config[key] = value if value else None
elif key == "path":
# Empty string means no path (will be None)
# Ensure non-empty path starts with /
if value:
if not value.startswith("/"):
value = "/" + value
env_config[key] = value
else:
env_config[key] = None
else:
env_config[key] = value
return env_config
def _validate_config(self, config):
"""
Validate configuration values.
Args:
config (dict): Configuration to validate
Raises:
ConfigurationError: If configuration is invalid
"""
# Validate required fields
if not config.get("host"):
raise ConfigurationError(
"Missing required field: 'host'\n"
f"Example configuration:\n{json.dumps(self.DEFAULT_CONFIG, indent=2)}"
)
if (
not isinstance(config.get("port"), int)
or config["port"] <= 0
or config["port"] > 65535
):
raise ConfigurationError(
f"Invalid port: {config.get('port')}. Must be integer between 1-65535."
)
if not isinstance(config.get("timeout"), int) or config["timeout"] <= 0:
raise ConfigurationError(
f"Invalid timeout: {config.get('timeout')}. Must be positive integer (milliseconds)."
)
# Validate optional fields
if config.get("secret") is not None and not isinstance(config["secret"], str):
raise ConfigurationError("Secret must be a string or null")
if not isinstance(config.get("secure"), bool):
raise ConfigurationError("Secure must be a boolean (true/false)")
# Validate path (optional field)
path = config.get("path")
if path is not None:
if not isinstance(path, str):
raise ConfigurationError("Path must be a string or null")
if not path.startswith("/"):
raise ConfigurationError(
f"Invalid path: {path}. Must start with '/' (e.g., '/jsonrpc') or be null"
)
def test_connection(self):
"""
Test connection to aria2 RPC endpoint.
Returns:
bool: True if connection successful, False otherwise
"""
if not self._loaded:
self.load()
# Build URL
protocol = "https" if self.config["secure"] else "http"
path = self.config.get("path") or ""
url = f"{protocol}://{self.config['host']}:{self.config['port']}{path}"
# Create a simple test request (aria2.getVersion)
request_data = {
"jsonrpc": "2.0",
"id": "test-connection",
"method": "aria2.getVersion",
"params": [],
}
# Add token if configured
if self.config.get("secret"):
request_data["params"].insert(0, f"token:{self.config['secret']}")
try:
req = urllib.request.Request(
url,
data=json.dumps(request_data).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
timeout_sec = self.config["timeout"] / 1000.0
response = urllib.request.urlopen(req, timeout=timeout_sec)
result = json.loads(response.read().decode("utf-8"))
# Check for valid JSON-RPC response
if "result" in result or "error" in result:
return True
else:
return False
except urllib.error.URLError as e:
print(f"Connection test failed: {e}")
return False
except Exception as e:
print(f"Connection test error: {e}")
return False
def get(self, key, default=None):
"""Get configuration value."""
if not self._loaded:
self.load()
return self.config.get(key, default)
def get_all(self):
"""Get all configuration values."""
if not self._loaded:
self.load()
return self.config.copy()
def get_loaded_from(self):
"""
Get information about where configuration was loaded from.
Returns:
dict: Dictionary with keys:
- 'path': Config file path or 'defaults'
- 'has_env_override': Whether environment variables were used
"""
if not self._loaded:
self.load()
return {
"path": self._loaded_from,
"has_env_override": self._loaded_from_env,
}
def reload(self):
"""
Reload configuration from all sources.
Returns:
dict: Reloaded configuration
Note: If reload fails, previous configuration is preserved.
"""
old_config = self.config.copy()
old_loaded_from = self._loaded_from
old_loaded_from_env = self._loaded_from_env
try:
self.config = self.DEFAULT_CONFIG.copy()
self._loaded = False
self._loaded_from = None
self._loaded_from_env = False
return self.load()
except Exception as e:
# Restore previous configuration on error
self.config = old_config
self._loaded = True
self._loaded_from = old_loaded_from
self._loaded_from_env = old_loaded_from_env
raise ConfigurationError(
f"Configuration reload failed: {e}\nPrevious configuration preserved."
)
def get_endpoint_url(self):
"""Get the full RPC endpoint URL."""
if not self._loaded:
self.load()
protocol = "https" if self.config["secure"] else "http"
path = self.config.get("path") or ""
return f"{protocol}://{self.config['host']}:{self.config['port']}{path}"
if __name__ == "__main__":
import argparse
import shutil
parser = argparse.ArgumentParser(
description="aria2 RPC Configuration Manager",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Show current configuration and source
python3 config_loader.py show
# Initialize user config (update-safe)
python3 config_loader.py init --user
# Initialize local config (project-specific)
python3 config_loader.py init --local
# Test connection
python3 config_loader.py test
""",
)
subparsers = parser.add_subparsers(dest="command", help="Command to execute")
# Show command
show_parser = subparsers.add_parser(
"show", help="Show current configuration and source"
)
# Init command
init_parser = subparsers.add_parser("init", help="Initialize configuration file")
init_group = init_parser.add_mutually_exclusive_group(required=True)
init_group.add_argument(
"--user",
action="store_true",
help="Initialize user config (~/.config/aria2-skill/)",
)
init_group.add_argument(
"--local", action="store_true", help="Initialize skill directory config"
)
# Test command
test_parser = subparsers.add_parser("test", help="Test connection to aria2 RPC")
args = parser.parse_args()
if not args.command:
parser.print_help()
sys.exit(0)
try:
if args.command == "show":
# Show current configuration
config = Aria2Config()
config.load()
loaded_info = config.get_loaded_from()
print("Configuration Status:")
print(f" Loaded from: {loaded_info['path']}")
if loaded_info["has_env_override"]:
print(" Environment overrides: Yes")
print()
print("Active Configuration:")
for key, value in config.get_all().items():
if key == "secret" and value:
print(f" {key}: ****** (hidden)")
else:
print(f" {key}: {value}")
print()
print(f"Endpoint URL: {config.get_endpoint_url()}")
elif args.command == "init":
# Initialize configuration file
script_dir = os.path.dirname(os.path.abspath(__file__))
skill_dir = os.path.dirname(script_dir)
example_file = os.path.join(skill_dir, "config.example.json")
if not os.path.exists(example_file):
print(f"✗ Error: config.example.json not found at {example_file}")
sys.exit(1)
if args.user:
target_dir = Aria2Config.USER_CONFIG_DIR
target_file = Aria2Config.USER_CONFIG_FILE
location_desc = "user config directory (update-safe)"
else: # args.local
target_dir = skill_dir
target_file = os.path.join(skill_dir, "config.json")
location_desc = "skill directory (project-specific)"
# Create directory if needed
if not os.path.exists(target_dir):
os.makedirs(target_dir, exist_ok=True)
print(f"✓ Created directory: {target_dir}")
# Check if config already exists
if os.path.exists(target_file):
response = input(
f"Config file already exists at {target_file}\nOverwrite? [y/N]: "
)
if response.lower() not in ["y", "yes"]:
print("Cancelled.")
sys.exit(0)
# Copy example to target
shutil.copy2(example_file, target_file)
print(f"✓ Configuration initialized at: {target_file}")
print(f" Location: {location_desc}")
print()
print("Next steps:")
print(f" 1. Edit the file: {target_file}")
print(" 2. Update host, port, secret as needed")
print(" 3. Test connection: python3 config_loader.py test")
elif args.command == "test":
# Test connection
print("Loading configuration...")
config = Aria2Config()
config.load()
loaded_info = config.get_loaded_from()
print(f"✓ Configuration loaded from: {loaded_info['path']}")
print()
print(f"Testing connection to {config.get_endpoint_url()}...")
if config.test_connection():
print("✓ Connection successful")
print()
print("aria2 RPC is accessible and responding correctly.")
else:
print("✗ Connection failed")
print()
print("Possible reasons:")
print(" 1. aria2 daemon is not running")
print(" 2. Wrong host/port configuration")
print(" 3. Network/firewall issues")
print(" 4. Wrong RPC secret token")
print()
print("Current configuration:")
for key, value in config.get_all().items():
if key == "secret" and value:
print(f" {key}: ****** (hidden)")
else:
print(f" {key}: {value}")
sys.exit(1)
except ConfigurationError as e:
print(f"✗ Configuration error: {e}")
sys.exit(1)
except KeyboardInterrupt:
print("\nCancelled by user.")
sys.exit(0)
except Exception as e:
print(f"✗ Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
```
### scripts/rpc_client.py
```python
#!/usr/bin/env python3
"""
JSON-RPC 2.0 client for aria2.
Implements the core RPC client with:
- JSON-RPC 2.0 request formatting
- Token authentication injection
- HTTP POST transport using urllib.request
- Response parsing and error handling
"""
import json
import urllib.request
import urllib.error
import sys
import time
import base64
import os
from typing import Any, Dict, List, Optional, Union
class Aria2RpcError(Exception):
"""Raised when aria2 returns an error response."""
def __init__(
self, code: int, message: str, data: Any = None, request_id: str = None
):
self.code = code
self.message = message
self.data = data
self.request_id = request_id
super().__init__(f"aria2 RPC error [{code}]: {message}")
class Aria2RpcClient:
"""
JSON-RPC 2.0 client for aria2.
Handles request formatting, authentication, HTTP transport,
and response parsing according to JSON-RPC 2.0 specification.
"""
def __init__(self, config: Dict[str, Any]):
"""
Initialize the RPC client with configuration.
Args:
config: Dictionary with keys: host, port, secret, secure, timeout
"""
self.config = config
self.request_counter = 0
self.endpoint_url = self._build_endpoint_url()
def _build_endpoint_url(self) -> str:
"""Build the full RPC endpoint URL."""
protocol = "https" if self.config.get("secure", False) else "http"
host = self.config["host"]
port = self.config["port"]
path = self.config.get("path") or ""
return f"{protocol}://{host}:{port}{path}"
def _generate_request_id(self) -> str:
"""Generate a unique request ID."""
self.request_counter += 1
return f"aria2-rpc-{self.request_counter}"
def _inject_token(self, params: List[Any]) -> List[Any]:
"""
Inject authentication token as first parameter if configured.
Args:
params: Original parameters array
Returns:
Parameters array with token prepended if secret is configured
"""
secret = self.config.get("secret")
if secret:
return [f"token:{secret}"] + params
return params
def _format_request(self, method: str, params: List[Any] = None) -> Dict[str, Any]:
"""
Format a JSON-RPC 2.0 request.
Args:
method: RPC method name (e.g., "aria2.addUri")
params: Method parameters (list)
Returns:
JSON-RPC 2.0 request dictionary
"""
if params is None:
params = []
# Inject token for aria2.* methods (not system.* methods)
if method.startswith("aria2."):
params = self._inject_token(params)
request = {
"jsonrpc": "2.0",
"id": self._generate_request_id(),
"method": method,
"params": params,
}
return request
def _send_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
Send HTTP POST request to aria2 RPC endpoint.
Args:
request: JSON-RPC request dictionary
Returns:
JSON-RPC response dictionary
Raises:
urllib.error.URLError: On network errors
json.JSONDecodeError: On response parse errors
"""
request_data = json.dumps(request).encode("utf-8")
req = urllib.request.Request(
self.endpoint_url,
data=request_data,
headers={
"Content-Type": "application/json",
"User-Agent": "aria2-json-rpc-client/1.0",
},
)
timeout_sec = self.config.get("timeout", 30000) / 1000.0
try:
response = urllib.request.urlopen(req, timeout=timeout_sec)
response_data = response.read().decode("utf-8")
return json.loads(response_data)
except urllib.error.HTTPError as e:
# Try to parse error response
try:
error_data = e.read().decode("utf-8")
return json.loads(error_data)
except:
raise Exception(f"HTTP error {e.code}: {e.reason}")
except urllib.error.URLError as e:
raise Exception(f"Network error: {e.reason}")
except json.JSONDecodeError as e:
raise Exception(
f"Invalid JSON response from aria2\n"
f"Parse error at line {e.lineno}, column {e.colno}: {e.msg}"
)
def _parse_response(self, response: Dict[str, Any], request_id: str) -> Any:
"""
Parse JSON-RPC 2.0 response and extract result or error.
Args:
response: JSON-RPC response dictionary
request_id: Request ID for correlation
Returns:
Result value from response
Raises:
Aria2RpcError: If response contains an error
Exception: If response is invalid
"""
# Validate JSON-RPC 2.0 response structure
if not isinstance(response, dict):
raise Exception("Invalid JSON-RPC response: not a dictionary")
if response.get("jsonrpc") != "2.0":
raise Exception(
"Invalid JSON-RPC response: missing or invalid jsonrpc field"
)
if response.get("id") != request_id:
raise Exception(
f"Invalid JSON-RPC response: ID mismatch "
f"(expected {request_id}, got {response.get('id')})"
)
# Check for error response
if "error" in response:
error = response["error"]
raise Aria2RpcError(
code=error.get("code", -1),
message=error.get("message", "Unknown error"),
data=error.get("data"),
request_id=request_id,
)
# Extract result
if "result" not in response:
raise Exception(
"Invalid JSON-RPC response: missing both result and error fields"
)
return response["result"]
def call(self, method: str, params: List[Any] = None) -> Any:
"""
Call an aria2 RPC method.
Args:
method: RPC method name (e.g., "aria2.addUri")
params: Method parameters (list)
Returns:
Result from aria2
Raises:
Aria2RpcError: If aria2 returns an error
Exception: On network or parse errors
"""
request = self._format_request(method, params)
request_id = request["id"]
try:
response = self._send_request(request)
return self._parse_response(response, request_id)
except Aria2RpcError:
# Re-raise aria2 errors as-is
raise
except Exception as e:
# Wrap other exceptions with context
raise Exception(
f"Failed to call {method}: {e}\n"
f"Request ID: {request_id}\n"
f"Endpoint: {self.endpoint_url}"
)
# Milestone 1 methods
def add_uri(
self,
uris: Union[str, List[str]],
options: Dict[str, Any] = None,
position: int = None,
) -> str:
"""
Add a new download task from URIs.
Args:
uris: Single URI string or list of URIs
options: Download options (e.g., {"dir": "/path", "out": "filename"})
position: Position in download queue (optional)
Returns:
GID (Global ID) of the new download task
"""
# Convert single URI to list
if isinstance(uris, str):
uris = [uris]
params = [uris]
if options:
params.append(options)
if position is not None:
params.append(position)
return self.call("aria2.addUri", params)
def tell_status(self, gid: str, keys: List[str] = None) -> Dict[str, Any]:
"""
Query status of a download task.
Args:
gid: GID of the download task
keys: Specific keys to retrieve (optional, returns all if None)
Returns:
Status dictionary with download information
"""
params = [gid]
if keys:
params.append(keys)
return self.call("aria2.tellStatus", params)
def remove(self, gid: str) -> str:
"""
Remove a download task.
Args:
gid: GID of the download task to remove
Returns:
GID of the removed task
"""
return self.call("aria2.remove", [gid])
def get_global_stat(self) -> Dict[str, Any]:
"""
Get global statistics.
Returns:
Dictionary with global stats (numActive, numWaiting, downloadSpeed, uploadSpeed, etc.)
"""
return self.call("aria2.getGlobalStat", [])
# Milestone 2 methods
def pause(self, gid: str) -> str:
"""
Pause a download task.
Args:
gid: GID of the download task to pause
Returns:
GID of the paused task
"""
return self.call("aria2.pause", [gid])
def pause_all(self) -> str:
"""
Pause all active downloads.
Returns:
"OK" on success
"""
return self.call("aria2.pauseAll", [])
def unpause(self, gid: str) -> str:
"""
Resume a paused download task.
Args:
gid: GID of the download task to resume
Returns:
GID of the resumed task
"""
return self.call("aria2.unpause", [gid])
def unpause_all(self) -> str:
"""
Resume all paused downloads.
Returns:
"OK" on success
"""
return self.call("aria2.unpauseAll", [])
def tell_active(self, keys: List[str] = None) -> List[Dict[str, Any]]:
"""
Get list of all active downloads.
Args:
keys: Specific keys to retrieve for each download (optional)
Returns:
List of active download status dictionaries
"""
params = []
if keys:
params.append(keys)
return self.call("aria2.tellActive", params)
def tell_waiting(
self, offset: int = 0, num: int = 100, keys: List[str] = None
) -> List[Dict[str, Any]]:
"""
Get list of waiting downloads with pagination.
Args:
offset: Starting offset in the queue (default: 0)
num: Number of downloads to retrieve (default: 100)
keys: Specific keys to retrieve for each download (optional)
Returns:
List of waiting download status dictionaries
"""
params = [offset, num]
if keys:
params.append(keys)
return self.call("aria2.tellWaiting", params)
def tell_stopped(
self, offset: int = 0, num: int = 100, keys: List[str] = None
) -> List[Dict[str, Any]]:
"""
Get list of stopped downloads with pagination.
Args:
offset: Starting offset in the queue (default: 0)
num: Number of downloads to retrieve (default: 100)
keys: Specific keys to retrieve for each download (optional)
Returns:
List of stopped download status dictionaries
"""
params = [offset, num]
if keys:
params.append(keys)
return self.call("aria2.tellStopped", params)
def get_option(self, gid: str) -> Dict[str, Any]:
"""
Get options for a specific download.
Args:
gid: GID of the download task
Returns:
Dictionary of download options
"""
return self.call("aria2.getOption", [gid])
def change_option(self, gid: str, options: Dict[str, Any]) -> str:
"""
Change options for a specific download.
Args:
gid: GID of the download task
options: Dictionary of options to change
Returns:
"OK" on success
"""
return self.call("aria2.changeOption", [gid, options])
def get_global_option(self) -> Dict[str, Any]:
"""
Get global aria2 options.
Returns:
Dictionary of global options
"""
return self.call("aria2.getGlobalOption", [])
def change_global_option(self, options: Dict[str, Any]) -> str:
"""
Change global aria2 options.
Args:
options: Dictionary of global options to change
Returns:
"OK" on success
"""
return self.call("aria2.changeGlobalOption", [options])
def purge_download_result(self) -> str:
"""
Remove completed/error/removed downloads from memory.
Returns:
"OK" on success
"""
return self.call("aria2.purgeDownloadResult", [])
def remove_download_result(self, gid: str) -> str:
"""
Remove a specific download result from memory.
Args:
gid: GID of the download result to remove
Returns:
"OK" on success
"""
return self.call("aria2.removeDownloadResult", [gid])
def get_version(self) -> Dict[str, Any]:
"""
Get aria2 version and enabled features.
Returns:
Dictionary with version information
"""
return self.call("aria2.getVersion", [])
def list_methods(self) -> List[str]:
"""
List all available RPC methods.
Returns:
List of method names
"""
return self.call("system.listMethods", [])
def multicall(self, calls: List[Dict[str, Any]]) -> List[Any]:
"""
Execute multiple RPC calls in a single request.
Args:
calls: List of method call dictionaries with keys:
- methodName: str (e.g., "aria2.tellStatus")
- params: List[Any]
Returns:
List of results corresponding to each call
"""
return self.call("system.multicall", [calls])
# Milestone 3 methods
def add_torrent(
self,
torrent: Union[str, bytes],
uris: List[str] = None,
options: Dict[str, Any] = None,
position: int = None,
) -> str:
"""
Add a new download from a torrent file.
Args:
torrent: Torrent file path, bytes content, or base64-encoded string
uris: Web seed URIs (optional)
options: Download options (optional)
position: Position in download queue (optional)
Returns:
GID (Global ID) of the new torrent download task
"""
# Convert torrent to base64 if needed
if isinstance(torrent, str):
# Check if it's already base64 or a file path
if os.path.isfile(torrent):
# Read file and encode to base64
with open(torrent, "rb") as f:
torrent_bytes = f.read()
torrent_base64 = base64.b64encode(torrent_bytes).decode("utf-8")
else:
# Assume it's already base64
torrent_base64 = torrent
elif isinstance(torrent, bytes):
# Encode bytes to base64
torrent_base64 = base64.b64encode(torrent).decode("utf-8")
else:
raise ValueError(
"torrent must be a file path (str), bytes content, or base64 string"
)
params = [torrent_base64]
if uris:
params.append(uris)
elif options or position is not None:
# If uris is not provided but options/position is, add empty list
params.append([])
if options:
params.append(options)
elif position is not None:
# If options is not provided but position is, add empty dict
params.append({})
if position is not None:
params.append(position)
return self.call("aria2.addTorrent", params)
def add_metalink(
self,
metalink: Union[str, bytes],
options: Dict[str, Any] = None,
position: int = None,
) -> List[str]:
"""
Add new downloads from a metalink file.
Args:
metalink: Metalink file path, bytes content, or base64-encoded string
options: Download options (optional)
position: Position in download queue (optional)
Returns:
List of GIDs for each download defined in the metalink
"""
# Convert metalink to base64 if needed
if isinstance(metalink, str):
# Check if it's already base64 or a file path
if os.path.isfile(metalink):
# Read file and encode to base64
with open(metalink, "rb") as f:
metalink_bytes = f.read()
metalink_base64 = base64.b64encode(metalink_bytes).decode("utf-8")
else:
# Assume it's already base64
metalink_base64 = metalink
elif isinstance(metalink, bytes):
# Encode bytes to base64
metalink_base64 = base64.b64encode(metalink).decode("utf-8")
else:
raise ValueError(
"metalink must be a file path (str), bytes content, or base64 string"
)
params = [metalink_base64]
if options:
params.append(options)
elif position is not None:
# If options is not provided but position is, add empty dict
params.append({})
if position is not None:
params.append(position)
return self.call("aria2.addMetalink", params)
def main():
"""
Command-line interface for aria2 RPC client.
Usage:
python rpc_client.py # Test connection
python rpc_client.py <method> # Call method with no params
python rpc_client.py <method> <param1> ... # Call method with params
Examples:
python rpc_client.py aria2.getGlobalStat
python rpc_client.py aria2.tellStatus 2089b05ecca3d829
python rpc_client.py aria2.addUri '["http://example.com/file.zip"]'
python rpc_client.py aria2.tellWaiting 0 100
"""
import sys
import os
# Add parent directory to path to import config_loader
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config_loader import Aria2Config
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
# Create RPC client
client = Aria2RpcClient(config)
# Parse command-line arguments
if len(sys.argv) == 1:
# No arguments: test connection with getGlobalStat
print("Testing aria2 JSON-RPC client...")
print()
print("✓ Configuration loaded")
print(f" Endpoint: {config_loader.get_endpoint_url()}")
print()
print("Testing connection with aria2.getGlobalStat...")
stats = client.get_global_stat()
print("✓ Connection successful")
print()
print("Global Statistics:")
for key, value in stats.items():
print(f" {key}: {value}")
else:
# Arguments provided: call specified method
method = sys.argv[1]
# Parse parameters
params = []
for arg in sys.argv[2:]:
# Try to parse as JSON first (for arrays/objects)
try:
params.append(json.loads(arg))
except json.JSONDecodeError:
# Not JSON, try as integer
try:
params.append(int(arg))
except ValueError:
# Not an integer, use as string
params.append(arg)
# Call the method
result = client.call(method, params)
# Print result as JSON
print(json.dumps(result, indent=2, ensure_ascii=False))
except Aria2RpcError as e:
print(f"✗ aria2 error: {e}", file=sys.stderr)
print(f" Code: {e.code}", file=sys.stderr)
print(f" Message: {e.message}", file=sys.stderr)
if e.data:
print(f" Data: {e.data}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"✗ Error: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/examples/list-downloads.py
```python
#!/usr/bin/env python3
"""
Example script: List all downloads (active, waiting, and stopped).
Usage:
python list-downloads.py [--limit NUM]
This script demonstrates how to query and display all download tasks
across different states: active, waiting, and stopped.
"""
import sys
import os
import argparse
# Add scripts directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config_loader import Aria2Config
from rpc_client import Aria2RpcClient, Aria2RpcError
def format_size(bytes_val):
"""Format bytes to human-readable size."""
if not bytes_val or bytes_val == "0":
return "0 B"
bytes_int = int(bytes_val)
units = ["B", "KB", "MB", "GB", "TB"]
unit_index = 0
while bytes_int >= 1024 and unit_index < len(units) - 1:
bytes_int /= 1024.0
unit_index += 1
return f"{bytes_int:.2f} {units[unit_index]}"
def format_speed(speed_val):
"""Format download speed."""
if not speed_val or speed_val == "0":
return "0 B/s"
return f"{format_size(speed_val)}/s"
def print_downloads(title, downloads):
"""Print formatted download list."""
print(f"\n{title}:")
print("=" * 80)
if not downloads:
print(" (none)")
return
for download in downloads:
gid = download.get("gid", "unknown")
status = download.get("status", "unknown")
# Get file info
files = download.get("files", [])
if files:
file_path = files[0].get("path", "unknown")
filename = (
os.path.basename(file_path) if file_path != "unknown" else "unknown"
)
else:
filename = "unknown"
# Get size info
total_length = download.get("totalLength", "0")
completed_length = download.get("completedLength", "0")
# Calculate progress
if total_length and total_length != "0":
total_int = int(total_length)
completed_int = int(completed_length)
progress = (completed_int / total_int) * 100 if total_int > 0 else 0
else:
progress = 0
# Get speed
download_speed = download.get("downloadSpeed", "0")
print(f"\nGID: {gid}")
print(f" Status: {status}")
print(f" File: {filename}")
print(
f" Progress: {format_size(completed_length)} / {format_size(total_length)} ({progress:.1f}%)"
)
if status == "active":
print(f" Speed: {format_speed(download_speed)}")
def main():
"""List all downloads."""
parser = argparse.ArgumentParser(description="List all aria2 downloads")
parser.add_argument(
"--limit",
type=int,
default=10,
help="Maximum number of stopped downloads to show (default: 10)",
)
args = parser.parse_args()
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
print(f"Connecting to aria2 at {config_loader.get_endpoint_url()}...")
# Create RPC client
client = Aria2RpcClient(config)
# Get active downloads
active = client.tell_active()
print_downloads(f"Active Downloads ({len(active)})", active)
# Get waiting downloads
waiting = client.tell_waiting(0, args.limit)
print_downloads(f"Waiting Downloads ({len(waiting)})", waiting)
# Get stopped downloads (completed, error, removed)
stopped = client.tell_stopped(0, args.limit)
print_downloads(
f"Stopped Downloads ({len(stopped)}, showing up to {args.limit})", stopped
)
# Summary
print("\n" + "=" * 80)
print(
f"Summary: {len(active)} active, {len(waiting)} waiting, {len(stopped)} stopped (limited)"
)
except Aria2RpcError as e:
print(f"✗ aria2 error: {e}")
print(f" Code: {e.code}")
print(f" Message: {e.message}")
sys.exit(1)
except Exception as e:
print(f"✗ Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/examples/pause-all.py
```python
#!/usr/bin/env python3
"""
Example script: Pause all active downloads.
Usage:
python pause-all.py
This script demonstrates how to pause all active downloads in aria2.
"""
import sys
import os
# Add scripts directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config_loader import Aria2Config
from rpc_client import Aria2RpcClient, Aria2RpcError
def main():
"""Pause all active downloads."""
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
print(f"Connecting to aria2 at {config_loader.get_endpoint_url()}...")
# Create RPC client
client = Aria2RpcClient(config)
# Get active downloads first to show what will be paused
print("\nGetting list of active downloads...")
active = client.tell_active()
if not active:
print("✓ No active downloads to pause")
return
print(f"Found {len(active)} active download(s):")
for download in active:
gid = download.get("gid", "unknown")
status = download.get("status", "unknown")
files = download.get("files", [])
filename = files[0].get("path", "unknown") if files else "unknown"
print(f" - GID {gid}: {filename} (status: {status})")
# Pause all downloads
print(f"\nPausing all {len(active)} download(s)...")
result = client.pause_all()
if result == "OK":
print("✓ All downloads paused successfully")
else:
print(f"✓ Pause result: {result}")
# Verify by checking active downloads again
print("\nVerifying...")
active_after = client.tell_active()
print(f"Active downloads after pause: {len(active_after)}")
except Aria2RpcError as e:
print(f"✗ aria2 error: {e}")
print(f" Code: {e.code}")
print(f" Message: {e.message}")
sys.exit(1)
except Exception as e:
print(f"✗ Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/examples/add-torrent.py
```python
#!/usr/bin/env python3
"""
Example script: Add a torrent file to aria2.
Usage:
python add-torrent.py <torrent-file> [--dir DIR] [--select-file INDEX]
This script demonstrates how to add a torrent file to aria2 with various options.
It shows:
- Loading a torrent file (automatically Base64 encoded)
- Setting download directory
- Selecting specific files from multi-file torrents
- Adding web seeds for additional download sources
- Monitoring the download progress
Requirements:
- aria2 must be running with RPC enabled
- Torrent file must exist at the specified path
"""
import sys
import os
import argparse
import time
# Add scripts directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config_loader import Aria2Config
from rpc_client import Aria2RpcClient, Aria2RpcError
def format_size(bytes_val):
"""Format bytes to human-readable size."""
if not bytes_val or bytes_val == "0":
return "0 B"
bytes_int = int(bytes_val)
units = ["B", "KB", "MB", "GB", "TB"]
unit_index = 0
while bytes_int >= 1024 and unit_index < len(units) - 1:
bytes_int /= 1024.0
unit_index += 1
return f"{bytes_int:.2f} {units[unit_index]}"
def format_speed(speed_val):
"""Format download speed."""
if not speed_val or speed_val == "0":
return "0 B/s"
return f"{format_size(speed_val)}/s"
def monitor_download(client, gid, duration=5):
"""Monitor download progress for a few seconds."""
print(f"\nMonitoring download for {duration} seconds...")
print("=" * 80)
start_time = time.time()
while time.time() - start_time < duration:
try:
status = client.tell_status(gid)
state = status.get("status", "unknown")
if state == "complete":
print("\n✓ Download completed!")
break
elif state == "error":
error_msg = status.get("errorMessage", "Unknown error")
print(f"\n✗ Download error: {error_msg}")
break
elif state == "removed":
print("\n✗ Download was removed")
break
# Get progress info
completed = int(status.get("completedLength", "0"))
total = int(status.get("totalLength", "0"))
speed = status.get("downloadSpeed", "0")
if total > 0:
progress = (completed / total) * 100
print(
f"\rProgress: {format_size(completed)} / {format_size(total)} "
f"({progress:.1f}%) - Speed: {format_speed(speed)}",
end="",
flush=True,
)
else:
print(
f"\rDownloading metadata... Speed: {format_speed(speed)}",
end="",
flush=True,
)
time.sleep(1)
except Aria2RpcError as e:
if e.code == 1: # GID not found - download might be complete
print("\n✓ Download completed (GID not found)")
break
raise
print() # New line after progress
def main():
"""Add a torrent file to aria2."""
parser = argparse.ArgumentParser(
description="Add a torrent file to aria2",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Add a torrent file
python add-torrent.py ubuntu.torrent
# Add torrent to specific directory
python add-torrent.py ubuntu.torrent --dir /downloads/ubuntu
# Add torrent and select only file at index 1
python add-torrent.py distro.torrent --select-file 1
# Add torrent with web seed mirror
python add-torrent.py file.torrent --web-seed http://mirror.example.com/file.iso
""",
)
parser.add_argument("torrent_file", help="Path to the torrent file")
parser.add_argument("--dir", help="Download directory (default: aria2's default)")
parser.add_argument(
"--select-file",
help="Select specific file index(es) from torrent (comma-separated)",
)
parser.add_argument(
"--web-seed",
action="append",
help="Add web seed URL (can be specified multiple times)",
)
parser.add_argument(
"--no-monitor", action="store_true", help="Don't monitor download progress"
)
args = parser.parse_args()
# Validate torrent file exists
if not os.path.isfile(args.torrent_file):
print(f"✗ Error: Torrent file not found: {args.torrent_file}")
sys.exit(1)
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
print(f"Connecting to aria2 at {config_loader.get_endpoint_url()}...")
# Create RPC client
client = Aria2RpcClient(config)
# Prepare options
options = {}
if args.dir:
options["dir"] = args.dir
print(f"Download directory: {args.dir}")
if args.select_file:
options["select-file"] = args.select_file
print(f"Selecting file(s): {args.select_file}")
# Prepare web seeds
web_seeds = args.web_seed if args.web_seed else []
if web_seeds:
print(f"Web seeds: {', '.join(web_seeds)}")
# Add the torrent
print(f"\nAdding torrent: {os.path.basename(args.torrent_file)}...")
gid = client.add_torrent(
torrent=args.torrent_file, uris=web_seeds, options=options
)
print(f"✓ Torrent added successfully!")
print(f" GID: {gid}")
# Get initial status
status = client.tell_status(gid)
# Show torrent info
files = status.get("files", [])
if files:
print(f"\nTorrent contains {len(files)} file(s):")
for i, file_info in enumerate(files):
file_path = file_info.get("path", "unknown")
file_size = file_info.get("length", "0")
selected = file_info.get("selected", "true")
marker = "✓" if selected == "true" else "✗"
print(
f" {marker} [{i + 1}] {os.path.basename(file_path)} ({format_size(file_size)})"
)
# Monitor download unless disabled
if not args.no_monitor:
monitor_download(client, gid)
else:
print("\nUse 'python list-downloads.py' to check download progress")
except Aria2RpcError as e:
print(f"✗ aria2 error: {e}")
print(f" Code: {e.code}")
print(f" Message: {e.message}")
sys.exit(1)
except Exception as e:
print(f"✗ Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/examples/monitor-downloads.py
```python
#!/usr/bin/env python3
"""
Example script: Monitor downloads in real-time using WebSocket.
Usage:
python monitor-downloads.py [--duration SECONDS]
This script demonstrates how to use aria2's WebSocket RPC interface
to monitor download progress in real-time. It shows:
- Connecting to aria2 via WebSocket
- Registering event handlers for download events
- Receiving real-time notifications about download progress
- Displaying download statistics as they update
Requirements:
- aria2 must be running with RPC enabled (preferably WebSocket)
- Python websockets library: `uv pip install websockets`
Note: If websockets is not available, the script will gracefully
fall back to suggesting HTTP polling alternatives.
"""
import sys
import os
import asyncio
import argparse
import time
# Add scripts directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config_loader import Aria2Config
from dependency_check import check_optional_websockets
# Check if websockets is available
if not check_optional_websockets():
print("✗ Error: This example requires the 'websockets' library")
print()
print("Install it with:")
print(" uv pip install websockets")
print()
print("Or use HTTP polling examples instead:")
print(" python list-downloads.py")
sys.exit(1)
from websocket_client import Aria2WebSocketClient
from rpc_client import Aria2RpcError
def format_size(bytes_val):
"""Format bytes to human-readable size."""
if not bytes_val or bytes_val == "0":
return "0 B"
try:
bytes_int = int(bytes_val)
except (ValueError, TypeError):
return "0 B"
units = ["B", "KB", "MB", "GB", "TB"]
unit_index = 0
while bytes_int >= 1024 and unit_index < len(units) - 1:
bytes_int /= 1024.0
unit_index += 1
return f"{bytes_int:.2f} {units[unit_index]}"
def format_speed(speed_val):
"""Format download speed."""
if not speed_val or speed_val == "0":
return "0 B/s"
return f"{format_size(speed_val)}/s"
class DownloadMonitor:
"""Monitor download events and display progress."""
def __init__(self):
self.downloads = {} # GID -> download info
self.start_time = time.time()
self.event_count = 0
async def on_download_start(self, event_data):
"""Handle download start event."""
gid = event_data.get("gid")
self.event_count += 1
print(f"\n[{self.event_count}] ▶ Download started: {gid}")
self.downloads[gid] = {"status": "active", "start_time": time.time()}
async def on_download_pause(self, event_data):
"""Handle download pause event."""
gid = event_data.get("gid")
self.event_count += 1
print(f"\n[{self.event_count}] ⏸ Download paused: {gid}")
if gid in self.downloads:
self.downloads[gid]["status"] = "paused"
async def on_download_stop(self, event_data):
"""Handle download stop event."""
gid = event_data.get("gid")
self.event_count += 1
print(f"\n[{self.event_count}] ⏹ Download stopped: {gid}")
if gid in self.downloads:
self.downloads[gid]["status"] = "stopped"
async def on_download_complete(self, event_data):
"""Handle download complete event."""
gid = event_data.get("gid")
self.event_count += 1
# Get final download info
try:
# We need to use HTTP client for this query since we're in an event handler
# In a real application, you might maintain this state differently
print(f"\n[{self.event_count}] ✓ Download completed: {gid}")
if gid in self.downloads:
elapsed = time.time() - self.downloads[gid].get(
"start_time", time.time()
)
print(f" Time elapsed: {elapsed:.1f} seconds")
self.downloads[gid]["status"] = "complete"
except Exception as e:
print(f" (Could not fetch final status: {e})")
async def on_download_error(self, event_data):
"""Handle download error event."""
gid = event_data.get("gid")
self.event_count += 1
print(f"\n[{self.event_count}] ✗ Download error: {gid}")
if gid in self.downloads:
self.downloads[gid]["status"] = "error"
async def on_bt_download_complete(self, event_data):
"""Handle BitTorrent download complete event."""
gid = event_data.get("gid")
self.event_count += 1
print(f"\n[{self.event_count}] ✓ BitTorrent download completed: {gid}")
if gid in self.downloads:
self.downloads[gid]["status"] = "complete"
def print_summary(self):
"""Print monitoring summary."""
elapsed = time.time() - self.start_time
print("\n" + "=" * 80)
print(f"Monitoring Summary:")
print(f" Duration: {elapsed:.1f} seconds")
print(f" Events received: {self.event_count}")
print(f" Downloads tracked: {len(self.downloads)}")
if self.downloads:
status_counts = {}
for dl in self.downloads.values():
status = dl.get("status", "unknown")
status_counts[status] = status_counts.get(status, 0) + 1
print(f" Status breakdown:")
for status, count in sorted(status_counts.items()):
print(f" {status}: {count}")
async def monitor_downloads(config, duration):
"""Monitor downloads using WebSocket connection."""
monitor = DownloadMonitor()
# Create WebSocket client
client = Aria2WebSocketClient(config)
# Register event handlers
client.on("onDownloadStart", monitor.on_download_start)
client.on("onDownloadPause", monitor.on_download_pause)
client.on("onDownloadStop", monitor.on_download_stop)
client.on("onDownloadComplete", monitor.on_download_complete)
client.on("onDownloadError", monitor.on_download_error)
client.on("onBtDownloadComplete", monitor.on_bt_download_complete)
print("Connecting to aria2 WebSocket...")
print(f"Will monitor for {duration} seconds (Ctrl+C to stop early)")
print("=" * 80)
try:
# Connect to WebSocket
await client.connect()
print("✓ Connected to aria2 WebSocket")
print("Listening for download events...\n")
# Wait for the specified duration
await asyncio.sleep(duration)
except KeyboardInterrupt:
print("\n\nInterrupted by user")
except Exception as e:
print(f"\n✗ Error: {e}")
import traceback
traceback.print_exc()
finally:
# Disconnect
await client.disconnect()
print("\n✓ Disconnected from aria2")
# Print summary
monitor.print_summary()
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Monitor aria2 downloads in real-time via WebSocket",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Monitor downloads for 30 seconds
python monitor-downloads.py
# Monitor downloads for 5 minutes
python monitor-downloads.py --duration 300
# Monitor indefinitely (until Ctrl+C)
python monitor-downloads.py --duration 999999
Events monitored:
- onDownloadStart: New download started
- onDownloadPause: Download paused
- onDownloadStop: Download stopped
- onDownloadComplete: Download completed
- onDownloadError: Download encountered error
- onBtDownloadComplete: BitTorrent download completed
Tips:
- Start a download in another terminal while monitoring
- Use 'python add-torrent.py' to test torrent download events
- Use aria2c CLI to start downloads and watch events here
""",
)
parser.add_argument(
"--duration",
type=int,
default=30,
help="How long to monitor in seconds (default: 30)",
)
args = parser.parse_args()
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
# Check if WebSocket is configured
if not config.get("websocket_url"):
print("⚠ Warning: No WebSocket URL configured in config.json")
print(
"Using HTTP endpoint instead, but WebSocket is recommended for monitoring"
)
print()
# Run the async monitor
asyncio.run(monitor_downloads(config, args.duration))
except KeyboardInterrupt:
print("\n\nInterrupted by user")
sys.exit(0)
except Exception as e:
print(f"✗ Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/examples/set-options.py
```python
#!/usr/bin/env python3
"""
Example script: Set options for a download or globally.
Usage:
python set-options.py --gid <GID> --max-download-limit <VALUE>
python set-options.py --global --max-concurrent-downloads <VALUE>
This script demonstrates how to modify download options for a specific
task or change global aria2 options.
"""
import sys
import os
import argparse
# Add scripts directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from config_loader import Aria2Config
from rpc_client import Aria2RpcClient, Aria2RpcError
def parse_speed_limit(value):
"""Parse speed limit value (e.g., '1M', '500K', '1024' bytes)."""
value = value.strip().upper()
if value.endswith("K"):
return str(int(value[:-1]) * 1024)
elif value.endswith("M"):
return str(int(value[:-1]) * 1024 * 1024)
elif value.endswith("G"):
return str(int(value[:-1]) * 1024 * 1024 * 1024)
else:
# Assume bytes
return value
def main():
"""Set download or global options."""
parser = argparse.ArgumentParser(
description="Set aria2 options for a download or globally"
)
parser.add_argument(
"--gid",
type=str,
help="GID of the download to modify (omit for global options)",
)
parser.add_argument(
"--global",
dest="is_global",
action="store_true",
help="Modify global options instead of a specific download",
)
parser.add_argument(
"--max-download-limit",
type=str,
help="Max download speed (e.g., '1M', '500K', '0' for unlimited)",
)
parser.add_argument(
"--max-upload-limit",
type=str,
help="Max upload speed (e.g., '100K', '0' for unlimited)",
)
parser.add_argument(
"--max-concurrent-downloads",
type=int,
help="Max concurrent downloads (global option)",
)
parser.add_argument(
"--max-connection-per-server", type=int, help="Max connections per server"
)
parser.add_argument(
"--split", type=int, help="Number of connections for a single download"
)
args = parser.parse_args()
# Build options dictionary
options = {}
if args.max_download_limit:
options["max-download-limit"] = parse_speed_limit(args.max_download_limit)
if args.max_upload_limit:
options["max-upload-limit"] = parse_speed_limit(args.max_upload_limit)
if args.max_concurrent_downloads:
options["max-concurrent-downloads"] = str(args.max_concurrent_downloads)
if args.max_connection_per_server:
options["max-connection-per-server"] = str(args.max_connection_per_server)
if args.split:
options["split"] = str(args.split)
if not options:
print("✗ No options specified. Use --help for usage.")
sys.exit(1)
if not args.is_global and not args.gid:
print("✗ Either --gid or --global must be specified")
sys.exit(1)
try:
# Load configuration
config_loader = Aria2Config()
config = config_loader.load()
print(f"Connecting to aria2 at {config_loader.get_endpoint_url()}...")
# Create RPC client
client = Aria2RpcClient(config)
if args.is_global:
# Set global options
print(f"\nSetting global options:")
for key, value in options.items():
print(f" {key}: {value}")
result = client.change_global_option(options)
if result == "OK":
print("✓ Global options updated successfully")
else:
print(f"✓ Result: {result}")
# Show updated global options
print("\nUpdated global options:")
global_opts = client.get_global_option()
for key in options.keys():
print(f" {key}: {global_opts.get(key, 'N/A')}")
else:
# Set options for specific download
print(f"\nSetting options for GID {args.gid}:")
for key, value in options.items():
print(f" {key}: {value}")
result = client.change_option(args.gid, options)
if result == "OK":
print(f"✓ Options updated successfully for GID {args.gid}")
else:
print(f"✓ Result: {result}")
# Show updated options
print(f"\nUpdated options for GID {args.gid}:")
download_opts = client.get_option(args.gid)
for key in options.keys():
print(f" {key}: {download_opts.get(key, 'N/A')}")
except Aria2RpcError as e:
print(f"✗ aria2 error: {e}")
print(f" Code: {e.code}")
print(f" Message: {e.message}")
sys.exit(1)
except Exception as e:
print(f"✗ Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
```
---
## Skill Companion Files
> Additional files collected from the skill directory layout.
### _meta.json
```json
{
"owner": "azzgo",
"slug": "aria2-json-rpc",
"displayName": "Aria2 Json Rpc",
"latest": {
"version": "0.1.0",
"publishedAt": 1770273051710,
"commit": "https://github.com/clawdbot/skills/commit/c47f330a0f858e795edb86ac137ea34f74d1cfce"
},
"history": []
}
```
### scripts/command_mapper.py
```python
#!/usr/bin/env python3
"""
Natural language command mapping for aria2-json-rpc skill.
Maps natural language commands to aria2 RPC methods with parameter extraction.
Enables AI agents to interpret commands like:
- "download http://example.com/file.zip"
- "show status for GID 2089b05ecca3d829"
- "remove download 2089b05ecca3d829"
- "show global stats"
"""
import re
from typing import Dict, List, Any, Optional, Tuple
class CommandMapper:
"""
Maps natural language commands to aria2 RPC method calls.
Uses keyword detection and pattern matching for intent classification.
"""
# Milestone 1 command patterns
PATTERNS = {
# addUri patterns - order matters, most specific first
"add_uri": [
r"^download\s+(https?://\S+|ftp://\S+|magnet:\S+|sftp://\S+)",
r"^add\s+(?:download\s+)?(?:uri\s+)?(https?://\S+|ftp://\S+|magnet:\S+)",
r"^fetch\s+(https?://\S+|ftp://\S+|sftp://\S+)",
],
# tellStatus patterns - require full GID
"tell_status": [
r"^(?:show|get|check)\s+status\s+(?:for\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^status\s+of\s+(?:gid\s+)?([a-f0-9]{16})",
],
# remove patterns - require full GID
"remove": [
r"^remove\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^delete\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^cancel\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
],
# getGlobalStat patterns - no parameters
"get_global_stat": [
r"^(?:show|get|display)\s+global\s+(?:stats?|statistics)",
r"^(?:show|get)\s+(?:overall|all)\s+(?:stats?|statistics)",
r"^what'?s\s+downloading\??$",
r"^how\s+many\s+(?:downloads?|tasks?)\??$",
],
# Milestone 2: pause patterns
"pause": [
r"^pause\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^stop\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
],
# Milestone 2: pauseAll patterns
"pause_all": [
r"^pause\s+all\s+(?:downloads?|tasks?)",
r"^stop\s+all\s+(?:downloads?|tasks?)",
r"^pause\s+everything",
],
# Milestone 2: unpause patterns
"unpause": [
r"^(?:unpause|resume)\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^continue\s+(?:download\s+)?(?:gid\s+)?([a-f0-9]{16})",
],
# Milestone 2: unpauseAll patterns
"unpause_all": [
r"^(?:unpause|resume)\s+all\s+(?:downloads?|tasks?)",
r"^continue\s+all\s+(?:downloads?|tasks?)",
r"^(?:unpause|resume)\s+everything",
],
# Milestone 2: tellActive patterns
"tell_active": [
r"^(?:show|list|get)\s+active\s+(?:downloads?|tasks?)",
r"^what'?s\s+(?:currently\s+)?(?:downloading|active)",
r"^active\s+(?:downloads?|tasks?)",
],
# Milestone 2: tellWaiting patterns
"tell_waiting": [
r"^(?:show|list|get)\s+waiting\s+(?:downloads?|tasks?)",
r"^(?:show|list|get)\s+queued?\s+(?:downloads?|tasks?)",
r"^waiting\s+(?:downloads?|tasks?)",
],
# Milestone 2: tellStopped patterns
"tell_stopped": [
r"^(?:show|list|get)\s+stopped\s+(?:downloads?|tasks?)",
r"^(?:show|list|get)\s+(?:completed?|finished)\s+(?:downloads?|tasks?)",
r"^stopped\s+(?:downloads?|tasks?)",
],
# Milestone 2: getOption patterns
"get_option": [
r"^(?:show|get)\s+options?\s+(?:for\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^options?\s+(?:of|for)\s+(?:gid\s+)?([a-f0-9]{16})",
],
# Milestone 2: changeOption patterns - complex, need GID + options
"change_option": [
r"^(?:change|set|modify)\s+options?\s+(?:for\s+)?(?:gid\s+)?([a-f0-9]{16})",
],
# Milestone 2: getGlobalOption patterns
"get_global_option": [
r"^(?:show|get)\s+global\s+options?",
r"^global\s+options?",
],
# Milestone 2: changeGlobalOption patterns
"change_global_option": [
r"^(?:change|set|modify)\s+global\s+options?",
],
# Milestone 2: purgeDownloadResult patterns
"purge_download_result": [
r"^purge\s+(?:download\s+)?(?:results?|history)",
r"^clear\s+(?:download\s+)?(?:results?|history)",
r"^clean\s+up\s+(?:completed?|finished)\s+downloads?",
],
# Milestone 2: removeDownloadResult patterns
"remove_download_result": [
r"^remove\s+(?:download\s+)?result\s+(?:for\s+)?(?:gid\s+)?([a-f0-9]{16})",
r"^clear\s+(?:download\s+)?result\s+(?:for\s+)?(?:gid\s+)?([a-f0-9]{16})",
],
# Milestone 2: getVersion patterns
"get_version": [
r"^(?:show|get)\s+(?:aria2\s+)?version",
r"^version\s+info(?:rmation)?",
r"^what\s+version\s+of\s+aria2",
],
# Milestone 2: listMethods patterns
"list_methods": [
r"^(?:show|list|get)\s+(?:available\s+)?(?:rpc\s+)?methods?",
r"^what\s+methods?\s+are\s+available",
r"^available\s+(?:rpc\s+)?methods?",
],
# Milestone 3: addTorrent patterns
"add_torrent": [
r"^(?:add|download)\s+torrent\s+(.+\.torrent)",
r"^download\s+from\s+torrent\s+(.+\.torrent)",
r"^(?:add|start)\s+bittorrent\s+(.+\.torrent)",
],
# Milestone 3: addMetalink patterns
"add_metalink": [
r"^(?:add|download)\s+metalink\s+(.+\.metalink)",
r"^download\s+from\s+metalink\s+(.+\.metalink)",
],
}
def __init__(self):
"""Initialize the command mapper."""
self.compiled_patterns = {}
self._compile_patterns()
def _compile_patterns(self):
"""Compile regex patterns for faster matching."""
for method, patterns in self.PATTERNS.items():
self.compiled_patterns[method] = [
re.compile(pattern, re.IGNORECASE) for pattern in patterns
]
def map_command(self, command: str) -> Optional[Tuple[str, List[Any]]]:
"""
Map a natural language command to an aria2 RPC method and parameters.
Args:
command: Natural language command string
Returns:
Tuple of (method_name, params) if matched, None otherwise
Example: ("aria2.addUri", [["http://example.com/file.zip"]])
"""
command = command.strip()
# Try each method's patterns
for method, compiled_patterns in self.compiled_patterns.items():
for pattern in compiled_patterns:
match = pattern.search(command)
if match:
# Extract parameters based on method
params = self._extract_params(method, match, command)
rpc_method = self._method_to_rpc(method)
return (rpc_method, params)
return None
def _method_to_rpc(self, method: str) -> str:
"""Convert internal method name to aria2 RPC method name."""
mapping = {
# Milestone 1
"add_uri": "aria2.addUri",
"tell_status": "aria2.tellStatus",
"remove": "aria2.remove",
"get_global_stat": "aria2.getGlobalStat",
# Milestone 2
"pause": "aria2.pause",
"pause_all": "aria2.pauseAll",
"unpause": "aria2.unpause",
"unpause_all": "aria2.unpauseAll",
"tell_active": "aria2.tellActive",
"tell_waiting": "aria2.tellWaiting",
"tell_stopped": "aria2.tellStopped",
"get_option": "aria2.getOption",
"change_option": "aria2.changeOption",
"get_global_option": "aria2.getGlobalOption",
"change_global_option": "aria2.changeGlobalOption",
"purge_download_result": "aria2.purgeDownloadResult",
"remove_download_result": "aria2.removeDownloadResult",
"get_version": "aria2.getVersion",
"list_methods": "system.listMethods",
# Milestone 3
"add_torrent": "aria2.addTorrent",
"add_metalink": "aria2.addMetalink",
}
return mapping.get(method, method)
def _extract_params(self, method: str, match: re.Match, command: str) -> List[Any]:
"""
Extract parameters from regex match based on method type.
Args:
method: Internal method name
match: Regex match object
command: Original command string
Returns:
List of parameters for RPC method
"""
# Milestone 1 methods
if method == "add_uri":
# Extract URIs (single URI from match group)
uri = match.group(1).strip()
# Return as array of URIs (aria2.addUri expects array of URIs)
return [[uri]]
elif method == "tell_status":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "remove":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "get_global_stat":
# No parameters needed
return []
# Milestone 2 methods
elif method == "pause":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "pause_all":
# No parameters needed
return []
elif method == "unpause":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "unpause_all":
# No parameters needed
return []
elif method == "tell_active":
# No parameters needed (can optionally add keys parameter)
return []
elif method == "tell_waiting":
# Default pagination: offset=0, num=100
return [0, 100]
elif method == "tell_stopped":
# Default pagination: offset=0, num=100
return [0, 100]
elif method == "get_option":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "change_option":
# Extract GID (options would need to be provided separately)
gid = match.group(1).strip()
return [gid, {}] # Empty options dict as placeholder
elif method == "get_global_option":
# No parameters needed
return []
elif method == "change_global_option":
# Empty options dict as placeholder
return [{}]
elif method == "purge_download_result":
# No parameters needed
return []
elif method == "remove_download_result":
# Extract GID
gid = match.group(1).strip()
return [gid]
elif method == "get_version":
# No parameters needed
return []
elif method == "list_methods":
# No parameters needed
return []
# Milestone 3 methods
elif method == "add_torrent":
# Extract torrent file path
torrent_path = match.group(1).strip()
return [torrent_path]
elif method == "add_metalink":
# Extract metalink file path
metalink_path = match.group(1).strip()
return [metalink_path]
return []
def _looks_like_uri(self, text: str) -> bool:
"""Check if text looks like a URI."""
# Check for common URI schemes
uri_schemes = ["http://", "https://", "ftp://", "sftp://", "magnet:", "file://"]
text_lower = text.lower()
for scheme in uri_schemes:
if text_lower.startswith(scheme):
return True
# Check for common file extensions (might be relative path or filename)
if any(
text.endswith(ext)
for ext in [".zip", ".tar", ".gz", ".iso", ".mp4", ".pdf", ".torrent"]
):
return True
return False
def get_supported_commands(self) -> Dict[str, List[str]]:
"""
Get documentation of supported commands.
Returns:
Dictionary mapping method names to example commands
"""
return {
# Milestone 1
"add_uri": [
"download http://example.com/file.zip",
"add download https://example.com/file.iso",
"fetch ftp://mirror.org/archive.tar.gz",
],
"tell_status": [
"show status for GID 2089b05ecca3d829",
"status of 2089b05ecca3d829",
"check status 2089b05ecca3d829",
],
"remove": [
"remove download 2089b05ecca3d829",
"delete 2089b05ecca3d829",
"cancel download 2089b05ecca3d829",
],
"get_global_stat": [
"show global stats",
"get global statistics",
"what's downloading",
"how many downloads",
],
# Milestone 2
"pause": [
"pause download 2089b05ecca3d829",
"pause GID 2089b05ecca3d829",
"stop download 2089b05ecca3d829",
],
"pause_all": [
"pause all downloads",
"stop all tasks",
"pause everything",
],
"unpause": [
"unpause download 2089b05ecca3d829",
"resume GID 2089b05ecca3d829",
"continue download 2089b05ecca3d829",
],
"unpause_all": [
"unpause all downloads",
"resume all tasks",
"continue everything",
],
"tell_active": [
"show active downloads",
"list active tasks",
"what's currently downloading",
],
"tell_waiting": [
"show waiting downloads",
"list queued tasks",
"waiting downloads",
],
"tell_stopped": [
"show stopped downloads",
"list completed tasks",
"stopped downloads",
],
"get_option": [
"show options for GID 2089b05ecca3d829",
"get options 2089b05ecca3d829",
],
"change_option": [
"change options for GID 2089b05ecca3d829",
"set options 2089b05ecca3d829",
],
"get_global_option": [
"show global options",
"get global options",
],
"change_global_option": [
"change global options",
"set global options",
],
"purge_download_result": [
"purge download results",
"clear download history",
"clean up completed downloads",
],
"remove_download_result": [
"remove download result 2089b05ecca3d829",
"clear result for GID 2089b05ecca3d829",
],
"get_version": [
"show aria2 version",
"get version",
"what version of aria2",
],
"list_methods": [
"show available methods",
"list RPC methods",
"what methods are available",
],
# Milestone 3
"add_torrent": [
"add torrent /path/to/file.torrent",
"download torrent ubuntu-20.04.torrent",
"download from torrent movie.torrent",
],
"add_metalink": [
"add metalink file.metalink",
"download metalink archive.metalink",
"download from metalink package.metalink",
],
}
def main():
"""Test command mapping."""
print("Testing natural language command mapper...")
print()
mapper = CommandMapper()
# Test commands
test_commands = [
"download http://example.com/file.zip",
"download http://example.com/file1.zip http://example.com/file2.zip",
"show status for GID 2089b05ecca3d829",
"status of abc123def456",
"remove download 2089b05ecca3d829",
"show global stats",
"what's downloading",
"this should not match anything",
]
for command in test_commands:
result = mapper.map_command(command)
if result:
method, params = result
print(f"✓ '{command}'")
print(f" → {method}")
print(f" → params: {params}")
else:
print(f"✗ '{command}'")
print(f" → No match")
print()
# Show supported commands
print("Supported Commands:")
print()
for method, examples in mapper.get_supported_commands().items():
print(f"{method}:")
for example in examples:
print(f" - {example}")
print()
if __name__ == "__main__":
main()
```
### scripts/dependency_check.py
```python
#!/usr/bin/env python3
"""
Dependency check module for aria2-json-rpc skill.
Checks Python version and required builtin modules at startup.
Provides clear error messages for missing or incompatible dependencies.
"""
import sys
def check_python_version():
"""
Check if Python version meets minimum requirement (3.6+).
Returns:
bool: True if version is compatible, False otherwise
"""
if sys.version_info < (3, 6):
print("ERROR: Python 3.6 or higher is required")
print(f"Current version: {sys.version}")
print("Please install Python 3.6+ from https://www.python.org/downloads/")
return False
return True
def check_builtin_modules():
"""
Check if required builtin modules are available.
These should always be present in Python 3.6+, but we check
to detect corrupted installations.
Returns:
tuple: (bool, list) - Success status and list of missing modules
"""
required_modules = ["urllib.request", "json", "base64", "os"]
missing_modules = []
for module in required_modules:
try:
__import__(module)
except ImportError:
missing_modules.append(module)
if missing_modules:
print(f"ERROR: Required builtin modules missing: {', '.join(missing_modules)}")
print("This indicates a corrupted Python installation.")
print("Please reinstall Python 3.6+ or report this as a Python bug.")
return False, missing_modules
return True, []
def check_optional_websockets():
"""
Check if optional websockets library is available (Milestone 3).
Returns:
bool: True if websockets is available, False otherwise
"""
try:
import websockets
return True
except ImportError:
return False
def check_all_dependencies(milestone=1, verbose=True):
"""
Check all dependencies for the specified milestone.
Args:
milestone (int): Milestone number (1, 2, or 3)
verbose (bool): Whether to print warnings for optional dependencies
Returns:
bool: True if all required dependencies are met, False otherwise
"""
# Check Python version (mandatory for all milestones)
if not check_python_version():
sys.exit(1)
# Check builtin modules (mandatory for all milestones)
success, missing = check_builtin_modules()
if not success:
sys.exit(1)
# Check optional websockets (Milestone 3 only)
if milestone >= 3 and verbose:
if not check_optional_websockets():
print("WARNING: 'websockets' library not found")
print(
"WebSocket features will be disabled. Install with: pip install websockets"
)
print()
return True
if __name__ == "__main__":
# Run dependency checks when module is executed directly
print("Checking aria2-json-rpc dependencies...")
print()
# Determine milestone from command line args if provided
milestone = 1
if len(sys.argv) > 1:
try:
milestone = int(sys.argv[1])
except ValueError:
print(f"Invalid milestone: {sys.argv[1]}, using default (1)")
if check_all_dependencies(milestone=milestone):
print("✓ All required dependencies are satisfied")
print(
f"✓ Python {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
)
# Show optional dependencies status
if milestone >= 3:
if check_optional_websockets():
print("✓ Optional: websockets library available")
else:
print(
"✗ Optional: websockets library not available (WebSocket features disabled)"
)
sys.exit(0)
else:
print("✗ Dependency check failed")
sys.exit(1)
```
### scripts/websocket_client.py
```python
#!/usr/bin/env python3
"""
WebSocket client for aria2 JSON-RPC 2.0.
Provides persistent WebSocket connection for:
- Sending JSON-RPC requests over WebSocket
- Receiving server notifications (events)
- Automatic reconnection on connection loss
This module requires the 'websockets' library (optional dependency).
If not available, the skill falls back to HTTP POST transport.
"""
import json
import asyncio
import sys
from typing import Any, Dict, List, Optional, Callable
from dependency_check import check_optional_websockets
# Check if websockets is available
WEBSOCKET_AVAILABLE = check_optional_websockets()
if WEBSOCKET_AVAILABLE:
import websockets
from websockets.client import WebSocketClientProtocol
class Aria2WebSocketError(Exception):
"""Raised when WebSocket operation fails."""
pass
class Aria2WebSocketClient:
"""
WebSocket client for aria2 JSON-RPC 2.0.
Manages persistent WebSocket connection with:
- Event notification handling
- Automatic reconnection
- Request/response correlation
"""
def __init__(self, config: Dict[str, Any]):
"""
Initialize WebSocket client.
Args:
config: Dictionary with keys: host, port, secret, secure, timeout
"""
if not WEBSOCKET_AVAILABLE:
raise ImportError(
"websockets library not available. Install with: pip install websockets"
)
self.config = config
self.ws_url = self._build_ws_url()
self.connection: Optional[WebSocketClientProtocol] = None
self.request_counter = 0
self.event_handlers: Dict[str, List[Callable]] = {}
self.reconnect_enabled = True
self.reconnect_delay = 5 # seconds
self._running = False
def _build_ws_url(self) -> str:
"""Build the WebSocket endpoint URL."""
protocol = "wss" if self.config.get("secure", False) else "ws"
host = self.config["host"]
port = self.config["port"]
return f"{protocol}://{host}:{port}/jsonrpc"
def _generate_request_id(self) -> str:
"""Generate a unique request ID."""
self.request_counter += 1
return f"ws-aria2-rpc-{self.request_counter}"
def _inject_token(self, params: List[Any]) -> List[Any]:
"""
Inject authentication token as first parameter if configured.
Args:
params: Original parameters array
Returns:
Parameters array with token prepended if secret is configured
"""
secret = self.config.get("secret")
if secret:
return [f"token:{secret}"] + params
return params
def register_event_handler(self, event: str, handler: Callable):
"""
Register a callback for a specific aria2 event.
Args:
event: Event name (e.g., "aria2.onDownloadStart")
handler: Callback function that accepts event data
"""
if event not in self.event_handlers:
self.event_handlers[event] = []
self.event_handlers[event].append(handler)
def unregister_event_handler(self, event: str, handler: Callable):
"""
Unregister a callback for a specific aria2 event.
Args:
event: Event name
handler: Callback function to remove
"""
if event in self.event_handlers:
self.event_handlers[event].remove(handler)
async def connect(self):
"""
Establish WebSocket connection to aria2.
Raises:
Aria2WebSocketError: If connection fails
"""
try:
timeout_sec = self.config.get("timeout", 30000) / 1000.0
extra_headers = {
"User-Agent": "aria2-json-rpc-client/1.0",
}
self.connection = await asyncio.wait_for(
websockets.connect(self.ws_url, extra_headers=extra_headers),
timeout=timeout_sec,
)
print(f"✓ WebSocket connected to {self.ws_url}")
except asyncio.TimeoutError:
raise Aria2WebSocketError(
f"Connection timeout to {self.ws_url} (timeout: {timeout_sec}s)"
)
except Exception as e:
raise Aria2WebSocketError(f"Failed to connect to {self.ws_url}: {e}")
async def disconnect(self):
"""Close the WebSocket connection."""
if self.connection:
await self.connection.close()
self.connection = None
print("✓ WebSocket disconnected")
async def send_request(self, method: str, params: List[Any] = None) -> Any:
"""
Send a JSON-RPC request over WebSocket and wait for response.
Args:
method: RPC method name (e.g., "aria2.addUri")
params: Method parameters (list)
Returns:
Result from aria2
Raises:
Aria2WebSocketError: If not connected or request fails
"""
if not self.connection:
raise Aria2WebSocketError("Not connected to aria2 WebSocket")
if params is None:
params = []
# Inject token for aria2.* methods
if method.startswith("aria2."):
params = self._inject_token(params)
request = {
"jsonrpc": "2.0",
"id": self._generate_request_id(),
"method": method,
"params": params,
}
request_id = request["id"]
try:
# Send request
await self.connection.send(json.dumps(request))
# Wait for response
while True:
response_str = await self.connection.recv()
response = json.loads(response_str)
# Check if this is a notification (no id field)
if "id" not in response:
# Handle server notification
await self._handle_notification(response)
continue
# Check if this response matches our request
if response.get("id") == request_id:
return self._parse_response(response, request_id)
except websockets.exceptions.ConnectionClosed as e:
raise Aria2WebSocketError(f"Connection closed: {e}")
except json.JSONDecodeError as e:
raise Aria2WebSocketError(f"Invalid JSON response: {e}")
except Exception as e:
raise Aria2WebSocketError(f"Request failed: {e}")
def _parse_response(self, response: Dict[str, Any], request_id: str) -> Any:
"""
Parse JSON-RPC 2.0 response.
Args:
response: JSON-RPC response dictionary
request_id: Request ID for correlation
Returns:
Result value from response
Raises:
Aria2WebSocketError: If response contains an error
"""
if "error" in response:
error = response["error"]
raise Aria2WebSocketError(
f"aria2 RPC error [{error.get('code', -1)}]: {error.get('message', 'Unknown error')}"
)
if "result" not in response:
raise Aria2WebSocketError(
"Invalid JSON-RPC response: missing both result and error fields"
)
return response["result"]
async def _handle_notification(self, notification: Dict[str, Any]):
"""
Handle server-side notification (event).
Args:
notification: Notification message from aria2
"""
method = notification.get("method")
params = notification.get("params", [])
if method and method in self.event_handlers:
# Call all registered handlers for this event
for handler in self.event_handlers[method]:
try:
# Call handler (can be sync or async)
if asyncio.iscoroutinefunction(handler):
await handler(params)
else:
handler(params)
except Exception as e:
print(f"Error in event handler for {method}: {e}", file=sys.stderr)
async def listen_for_events(self):
"""
Listen for server notifications indefinitely.
This method should be run in a background task to handle
aria2 event notifications like onDownloadStart, onDownloadComplete, etc.
"""
if not self.connection:
raise Aria2WebSocketError("Not connected to aria2 WebSocket")
self._running = True
try:
while self._running:
try:
message_str = await self.connection.recv()
message = json.loads(message_str)
# Handle notifications (messages without id field)
if "id" not in message and "method" in message:
await self._handle_notification(message)
except websockets.exceptions.ConnectionClosed:
if self.reconnect_enabled and self._running:
print(
f"Connection lost. Reconnecting in {self.reconnect_delay} seconds..."
)
await asyncio.sleep(self.reconnect_delay)
await self.connect()
else:
break
except json.JSONDecodeError as e:
print(f"Invalid JSON notification: {e}", file=sys.stderr)
except Exception as e:
print(f"Error in event listener: {e}", file=sys.stderr)
except asyncio.CancelledError:
print("Event listener cancelled")
finally:
self._running = False
def stop_listening(self):
"""Stop the event listener loop."""
self._running = False
def check_websocket_available() -> bool:
"""
Check if WebSocket functionality is available.
Returns:
bool: True if websockets library is installed, False otherwise
"""
return WEBSOCKET_AVAILABLE
# Example usage
async def example_usage():
"""Example of using the WebSocket client."""
config = {
"host": "localhost",
"port": 6800,
"secret": None,
"secure": False,
"timeout": 30000,
}
client = Aria2WebSocketClient(config)
# Register event handlers
def on_download_start(params):
gid = params[0]["gid"] if params else "unknown"
print(f"Download started: GID {gid}")
def on_download_complete(params):
gid = params[0]["gid"] if params else "unknown"
print(f"Download completed: GID {gid}")
client.register_event_handler("aria2.onDownloadStart", on_download_start)
client.register_event_handler("aria2.onDownloadComplete", on_download_complete)
try:
# Connect
await client.connect()
# Start event listener in background
listener_task = asyncio.create_task(client.listen_for_events())
# Send a request
result = await client.send_request(
"aria2.addUri", [["http://example.com/file.zip"]]
)
print(f"Download added: GID {result}")
# Keep running to receive events
await asyncio.sleep(60)
# Stop listening and disconnect
client.stop_listening()
await listener_task
await client.disconnect()
except Aria2WebSocketError as e:
print(f"WebSocket error: {e}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
if not WEBSOCKET_AVAILABLE:
print("ERROR: websockets library not available")
print("Install with: pip install websockets")
sys.exit(1)
print("Running WebSocket client example...")
asyncio.run(example_usage())
```