Dynamic Package Import

Nekro Agent provides dynamic package import functionality, allowing plugins to install and import Python third-party packages on-demand at runtime. This feature enables plugins to flexibly utilize various tools and frameworks from the Python ecosystem without modifying the system environment.

Feature Overview

The dynamic package import functionality has the following characteristics:

• On-demand Installation: Downloads and installs packages only when needed, saving system resources
• Version Control: Supports precise version specifications and dependency constraints
• Isolated Storage: Packages are installed in plugin-specific directories without affecting the system Python environment
• Mirror Support: Supports configuring PyPI mirror sources to accelerate domestic downloads
• Error Handling: Provides friendly error prompts and exception handling mechanisms

Basic Usage

Import Function

Import the dynamic package import function from nekro_agent.api.plugin:

from nekro_agent.api.plugin import dynamic_import_pkg
from nekro_agent.api.schemas import AgentCtx

Simple Import Example

The simplest usage is to directly provide the package name:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "use_requests", "Use requests library to send HTTP requests")
async def fetch_url(_ctx: AgentCtx, url: str) -> str:
    """Get web page content using dynamically imported requests library"""

    # Dynamically import requests package
    requests = dynamic_import_pkg("requests")

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return f"Successfully fetched page, status code: {response.status_code}, content length: {len(response.text)} characters"
    except Exception as e:
        return f"Request failed: {e}"

Version Specifications

Specifying Version Ranges

Supports standard Python package version specification syntax:

# Exact version
pandas = dynamic_import_pkg("pandas==2.0.0")

# Minimum version requirement
numpy = dynamic_import_pkg("numpy>=1.24.0")

# Version range
requests = dynamic_import_pkg("requests>=2.25.0,<3.0.0")

# Exclude specific version
flask = dynamic_import_pkg("flask>=2.0,!=2.0.1")

Version Specification Example

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "data_analysis", "Execute data analysis")
async def analyze_data(_ctx: AgentCtx, data: list) -> str:
    """Perform data analysis using pandas"""

    # Import specific version of pandas
    pd = dynamic_import_pkg(
        "pandas>=2.0.0,<3.0.0",
        import_name="pandas"  # Specify module name for import
    )

    try:
        df = pd.DataFrame(data)
        summary = df.describe().to_string()
        return f"Data analysis results:\n{summary}"
    except Exception as e:
        return f"Analysis failed: {e}"

Advanced Parameter Configuration

Complete Parameter Description

dynamic_import_pkg(
    package_spec: str,                    # Package specification, e.g., "requests>=2.25.0"
    import_name: Optional[str] = None,    # Import name, defaults to package name
    mirror: Optional[str] = "https://pypi.tuna.tsinghua.edu.cn/simple",  # PyPI mirror source
    trusted_host: bool = True,            # Whether to trust mirror source host
    timeout: int = 300,                   # Installation timeout (seconds)
    repo_dir: Optional[Path] = None       # Custom installation directory
)

Using Domestic Mirrors for Acceleration

Uses Tsinghua University PyPI mirror source by default, but you can also specify other mirrors:

# Use Alibaba Cloud mirror
beautifulsoup4 = dynamic_import_pkg(
    "beautifulsoup4",
    mirror="https://mirrors.aliyun.com/pypi/simple/",
    trusted_host=True
)

# Use Tencent Cloud mirror
pillow = dynamic_import_pkg(
    "Pillow>=10.0.0",
    mirror="https://mirrors.cloud.tencent.com/pypi/simple",
    trusted_host=True
)

# Use official PyPI (slower)
scipy = dynamic_import_pkg(
    "scipy",
    mirror="https://pypi.org/simple",
    trusted_host=False
)

Practical Usage Examples

Example 1: Web Scraping

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scrape_webpage", "Scrape web page content")
async def scrape_webpage(_ctx: AgentCtx, url: str, selector: str) -> str:
    """Use BeautifulSoup to scrape specified elements from web pages"""

    # Dynamically import required packages
    requests = dynamic_import_pkg("requests>=2.25.0")
    bs4 = dynamic_import_pkg("beautifulsoup4>=4.9.0", import_name="bs4")

    try:
        # Get web page content
        response = requests.get(url, timeout=10)
        response.raise_for_status()

        # Parse HTML
        soup = bs4.BeautifulSoup(response.text, 'html.parser')
        elements = soup.select(selector)

        if not elements:
            return f"No elements found matching selector '{selector}'"

        # Extract text content
        results = [elem.get_text(strip=True) for elem in elements[:5]]  # Return up to 5
        return f"Found {len(elements)} elements, first {len(results)} contents:\n" + "\n".join(results)

    except Exception as e:
        return f"Scraping failed: {e}"

Example 2: Image Processing

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image", "Process image files")
async def process_image(_ctx: AgentCtx, image_path: str, operation: str) -> str:
    """Use Pillow to process images"""

    # Dynamically import Pillow
    PIL = dynamic_import_pkg("Pillow>=10.0.0", import_name="PIL")
    from PIL import Image, ImageFilter

    try:
        # Get actual file path
        real_path = _ctx.fs.get_file(image_path)

        # Open image
        img = Image.open(real_path)

        # Process based on operation type
        if operation == "blur":
            processed = img.filter(ImageFilter.BLUR)
        elif operation == "grayscale":
            processed = img.convert('L')
        elif operation == "resize":
            processed = img.resize((800, 600))
        else:
            return f"Unsupported operation: {operation}"

        # Save processed image
        output_path = _ctx.fs.shared_path / f"processed_{operation}.jpg"
        processed.save(output_path)

        # Return sandbox path to AI
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"Image processing completed, result: {sandbox_path}"

    except Exception as e:
        return f"Image processing failed: {e}"

Example 3: Data Processing and Visualization

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_chart", "Create data chart")
async def create_data_chart(_ctx: AgentCtx, data: dict, chart_type: str) -> str:
    """Create data charts using matplotlib"""

    # Dynamically import data processing and visualization libraries
    pd = dynamic_import_pkg("pandas>=2.0.0", import_name="pandas")
    plt = dynamic_import_pkg("matplotlib>=3.5.0", import_name="matplotlib.pyplot")

    try:
        # Create DataFrame
        df = pd.DataFrame(data)

        # Create chart
        fig, ax = plt.subplots(figsize=(10, 6))

        if chart_type == "bar":
            df.plot(kind='bar', ax=ax)
        elif chart_type == "line":
            df.plot(kind='line', ax=ax)
        elif chart_type == "pie":
            df.plot(kind='pie', y=df.columns[0], ax=ax)
        else:
            return f"Unsupported chart type: {chart_type}"

        # Save chart
        output_path = _ctx.fs.shared_path / f"chart_{chart_type}.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close(fig)

        # Return sandbox path
        sandbox_path = _ctx.fs.forward_file(output_path)
        return f"Chart generated: {sandbox_path}"

    except Exception as e:
        return f"Chart generation failed: {e}"

Example 4: Scientific Computing

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "scientific_compute", "Execute scientific computing")
async def scientific_compute(_ctx: AgentCtx, operation: str, values: list) -> str:
    """Perform scientific computing using NumPy and SciPy"""

    # Dynamically import scientific computing libraries
    np = dynamic_import_pkg("numpy>=1.24.0", import_name="numpy")
    scipy = dynamic_import_pkg("scipy>=1.10.0")

    try:
        data = np.array(values)

        if operation == "fft":
            # Fast Fourier Transform
            result = np.fft.fft(data)
            return f"FFT results (first 5): {result[:5]}"

        elif operation == "stats":
            # Statistical analysis
            from scipy import stats
            mean = np.mean(data)
            std = np.std(data)
            skew = stats.skew(data)
            kurtosis = stats.kurtosis(data)

            return f"""Statistical results:
- Mean: {mean:.4f}
- Standard deviation: {std:.4f}
- Skewness: {skew:.4f}
- Kurtosis: {kurtosis:.4f}"""

        elif operation == "interpolate":
            # Data interpolation
            from scipy import interpolate
            x = np.arange(len(data))
            f = interpolate.interp1d(x, data, kind='cubic')
            x_new = np.linspace(0, len(data)-1, len(data)*2)
            y_new = f(x_new)
            return f"Interpolation completed, original data points: {len(data)}, after interpolation: {len(y_new)}"

        else:
            return f"Unsupported operation: {operation}"

    except Exception as e:
        return f"Computation failed: {e}"

Error Handling

Dynamic imports may fail due to network issues, non-existent packages, or other reasons

Common Error Types

# RuntimeError: Installation failed
# - Network connection problems
# - Package does not exist or version unavailable
# - SSL certificate verification failed
# - Mirror source access denied

# ImportError: Import failed
# - Package installed but module cannot be found
# - Import name does not match actual module name

# ValueError: Package specification format error
# - Incorrect version number format
# - Unsupported version operators used

# subprocess.TimeoutExpired: Installation timeout
# - Package size too large
# - Network speed too slow

Considerations

Security Considerations

1. Only import trusted packages: Only install packages from official PyPI or trusted mirror sources
2. Version locking: Use exact version numbers in production environments to avoid unexpected package updates
3. Review dependencies: Understand transitive dependencies of packages to avoid introducing unnecessary risks
4. Document declaration: Declare used packages in plugin documentation and note version numbers

Performance Considerations

1. First installation overhead: The first time a package is used, there will be download and installation time
2. Disk space: Each package occupies disk space; be mindful of controlling the number and size of packages

Compatibility Considerations

1. Python version: Ensure packages support the Python version running Nekro Agent
2. System dependencies: Some packages may require system-level dependency libraries (e.g., OpenCV requires specific C++ libraries)
3. Package conflicts: Be aware of potential dependency conflicts between different packages

Troubleshooting

Installation Succeeded but Import Failed

Error: ImportError: Installation succeeded but cannot import module_name

Solution:
1. Check if the import_name parameter is correct
2. Some packages have different import names than package names (e.g., Pillow -> PIL)
3. Package may require additional system dependencies

By properly utilizing the dynamic package import functionality, your plugin can flexibly leverage Python's rich ecosystem to implement more powerful and diverse features.