Function _import_monitor

agentops/instrumentation/__init__.py:375–448

Monitor imports and instrument packages as they are imported. This replaces the built-in import function to intercept package imports.

(name: str, globals_dict=None, locals_dict=None, fromlist=(), level=0)

def _import_monitor(name: str, globals_dict=None, locals_dict=None, fromlist=(), level=0):
    """
    Monitor imports and instrument packages as they are imported.
    This replaces the built-in import function to intercept package imports.
    """
    global _instrumenting_packages, _has_agentic_library

    # If an agentic library is already instrumented, skip all further instrumentation
    if _has_agentic_library:
        return _original_builtins_import(name, globals_dict, locals_dict, fromlist, level)

    # First, do the actual import
    module = _original_builtins_import(name, globals_dict, locals_dict, fromlist, level)

    # Check for exact matches first (handles package.module like google.adk)
    packages_to_check = set()

    # Check the imported module itself
    if name in TARGET_PACKAGES:
        packages_to_check.add(name)
    else:
        # Check if any target package is a prefix of the import name
        for target in TARGET_PACKAGES:
            if name.startswith(target + ".") or name == target:
                packages_to_check.add(target)

    # For "from X import Y" style imports, also check submodules
    if fromlist:
        for item in fromlist:
            # Construct potential full name, e.g., "google.adk" from name="google", item="adk"
            # Or if name="os", item="path", full_name="os.path"
            # If the original name itself is a multi-part name like "a.b", and item is "c", then "a.b.c"
            # This logic needs to correctly identify the root package if 'name' is already a sub-package.
            # The existing TARGET_PACKAGES check is simpler: it checks against pre-defined full names.

            # Check full name if item forms part of a target package name
            full_item_name_candidate = f"{name}.{item}"

            if full_item_name_candidate in TARGET_PACKAGES:
                packages_to_check.add(full_item_name_candidate)
            else:  # Fallback to checking if 'name' itself is a target
                for target in TARGET_PACKAGES:
                    if name == target or name.startswith(target + "."):
                        packages_to_check.add(target)  # Check the base target if a submodule is imported from it.

    # Instrument all matching packages
    for package_to_check in packages_to_check:
        if package_to_check not in _instrumenting_packages and not _is_package_instrumented(package_to_check):
            target_module_obj = sys.modules.get(package_to_check)

            if target_module_obj:
                is_sdk = _is_installed_package(target_module_obj, package_to_check)
                if not is_sdk:
                    logger.debug(
                        f"AgentOps: Target '{package_to_check}' appears to be a local module/directory. Skipping AgentOps SDK instrumentation for it."
                    )
                    continue
            else: