Framework Adapter Plugins

from praisonaiagents import Agent

agent = Agent(name="adapter-agent", instructions="Use framework adapter plugins.")
agent.start("Enable the LangChain adapter plugin for this workflow.")

# Example: after pip install praisonai-langgraph
$ praisonai --framework langgraph agents.yaml   # accepted
$ praisonai doctor
✓ agents.yaml framework='langgraph' is a registered adapter

class MyFrameworkAdapter:
    name = "myframework"
    install_hint = 'pip install "myframework-praisonai[gpu]"'

    def is_available(self) -> bool: ...
    def run(self, config, llm_config, topic, *, tools_dict=None,
            agent_callback=None, task_callback=None, cli_config=None): ...
    def cleanup(self) -> None: ...

ImportError: 'myframework' was requested but is not installed.
Install with: pip install "myframework-praisonai[gpu]"

from praisonai.framework_adapters import (
    list_framework_choices,   # canonical name list (sorted, optionally filtered to available)
    get_install_hint,         # adapter-declared hint, or fallback
    get_default_registry,
)

TypeError: FrameworkAdapter '<name>' does not implement the protocol:
missing keyword-only parameters ['agent_callback', 'cli_config', 'task_callback', 'tools_dict']

from praisonaiagents import BaseFrameworkAdapter

class MyFrameworkAdapter(BaseFrameworkAdapter):
    name = "myframework"
    SUPPORTS_ASYNC = True             # native async arun() below
    _THREAD_OFFLOAD_MAX_WORKERS = 8   # only used when SUPPORTS_ASYNC is False

    def is_available(self) -> bool: ...

    def run(self, config, llm_config, topic, *, tools_dict=None,
            agent_callback=None, task_callback=None, cli_config=None): ...

    async def arun(self, config, llm_config, topic, *, tools_dict=None,
                   agent_callback=None, task_callback=None, cli_config=None): ...

from praisonaiagents import BaseFrameworkAdapter

class MyWorkflowFrameworkAdapter(BaseFrameworkAdapter):
    name = "myframework"
    SUPPORTS_ASYNC = True              # native async run() path
    SUPPORTS_WORKFLOW = True           # accept workflow YAML dispatch
    SUPPORTS_RUNTIME_FEATURES = True   # accept cli_backend / runtime keys

    def is_available(self) -> bool:
        return True

    def run(self, config, llm_config, topic, *,
            tools_dict=None, agent_callback=None,
            task_callback=None, cli_config=None) -> str:
        return f"myframework ran: {topic}"

import os
import logging
from typing import Dict, Any, Optional
from praisonaiagents import BaseFrameworkAdapter

logger = logging.getLogger(__name__)

class AutoGenFamilyAdapter(BaseFrameworkAdapter):
    """Router adapter for AutoGen family (v0.2, v0.4, AG2)."""

    name = "autogen"
    is_router = True

    def is_available(self) -> bool:
        v2 = AutoGenAdapter()
        v4 = AutoGenV4Adapter()
        ag2 = AG2Adapter()
        return v2.is_available() or v4.is_available() or ag2.is_available()

    def resolve_alias(self, config: Optional[Dict[str, Any]] = None) -> str:
        """Pick adapter name from config or AUTOGEN_VERSION env var."""
        requested = str(
            (config or {}).get("autogen_version")
            or os.getenv("AUTOGEN_VERSION", "auto")
        ).strip().lower()
        v2_available = AutoGenAdapter().is_available()
        v4_available = AutoGenV4Adapter().is_available()
        ag2_available = AG2Adapter().is_available()

        if requested == "v0.2":
            if v2_available:
                return "autogen_v2"
            raise ImportError(
                "AUTOGEN_VERSION=v0.2 was requested, but the AutoGen v0.2 adapter "
                "is not available. Install with: pip install 'praisonai-frameworks[autogen]'."
            )
        if requested == "v0.4":
            if v4_available:
                return "autogen_v4"
            raise ImportError(
                "AUTOGEN_VERSION=v0.4 was requested, but the v0.4 adapter is not "
                "registered or available. Install/register an autogen_v4 adapter "
                "(pip install 'praisonai-frameworks[autogen-v4]'), or unset AUTOGEN_VERSION "
                "to use auto-selection."
            )
        if requested == "ag2":
            if ag2_available:
                return "ag2"
            raise ImportError(
                "AUTOGEN_VERSION=ag2 was requested, but the AG2 adapter is not "
                "registered or available. Install/register an AG2 adapter "
                "(pip install 'praisonai-frameworks[ag2]'), or unset AUTOGEN_VERSION to use "
                "auto-selection."
            )

        # auto: prefer v0.2 (v0.4 and ag2 are currently unimplemented)
        if v2_available:
            return "autogen_v2"
        if v4_available:
            return "autogen_v4"
        if ag2_available:
            return "ag2"

        raise ImportError(
            "No runnable AutoGen variant is available. Install with:\n"
            "  pip install 'praisonai-frameworks[autogen]' for v0.2\n"
            "  pip install 'praisonai-frameworks[autogen-v4]' for v0.4\n"
            "  pip install 'praisonai-frameworks[ag2]' for AG2"
        )

    def resolve(self, *, config: Optional[Dict[str, Any]] = None) -> "BaseFrameworkAdapter":
        adapter_name = self.resolve_alias(config)
        from praisonai.framework_adapters.registry import get_default_registry
        return get_default_registry().create(adapter_name)

    def run(
        self,
        config,
        llm_config,
        topic,
        *,
        tools_dict=None,
        agent_callback=None,
        task_callback=None,
        cli_config=None,
    ) -> str:
        raise RuntimeError(
            "AutoGenFamilyAdapter.run() should not be called directly. "
            "The resolve() method should have been called first to get the concrete adapter."
        )

def resolve(self, *, config: Optional[Dict[str, Any]] = None) -> "FrameworkAdapter":
    """Pick the concrete adapter variant (e.g. autogen v0.2 vs v0.4).

    Args:
        config: YAML configuration that may contain version preferences

    Returns:
        The resolved adapter instance (self or a different adapter)
    """
    return self

from praisonaiagents import BaseFrameworkAdapter

def resolve(self, *, config: Optional[Dict[str, Any]] = None) -> "BaseFrameworkAdapter":
    version = "auto"
    if config and config.get("autogen_version"):
        version = str(config["autogen_version"]).lower()
    else:
        version = os.environ.get("AUTOGEN_VERSION", "auto").lower()

    v4_adapter = AutoGenV4Adapter()  # direct instantiation — no registry.create()
    v2_adapter = self

    if version == "v0.4" and v4_adapter.is_available():
        return v4_adapter
    elif version == "v0.2" and v2_adapter.is_available():
        return v2_adapter
    # ... auto-detect and fallback ...

def setup(self, *, framework_tag: str) -> None:
    """Framework-specific pre-run hooks (SDK init, etc.).
    
    Default implementation is a no-op. Use this for one-time per-run setup
    that depends on the resolved framework name (logging context, SDK init).
    Note: observability init is handled centrally by init_observability(),
    you don't need to call agentops.init() here.
    """
    pass

from typing import Any, Callable, Dict, List, Optional
from praisonaiagents import BaseFrameworkAdapter
from praisonai.framework_adapters.registry import get_default_registry

class CustomCrewAIAdapter(BaseFrameworkAdapter):
    name = "crewai"

    def is_available(self) -> bool:
        try:
            import crewai
            return True
        except ImportError:
            return False

    def run(
        self,
        config: Dict[str, Any],
        llm_config: List[Dict],
        topic: str,
        *,
        tools_dict: Optional[Dict[str, Any]] = None,
        agent_callback: Optional[Callable] = None,
        task_callback: Optional[Callable] = None,
        cli_config: Optional[Dict[str, Any]] = None,
    ) -> str:
        # Custom CrewAI execution logic
        return "Custom CrewAI result"

    def cleanup(self) -> None:
        super().cleanup()

# Override built-in CrewAI adapter
registry = get_default_registry()
registry.register("crewai", CustomCrewAIAdapter)

import subprocess
from typing import Any, Callable, Dict, List, Optional
from praisonaiagents import BaseFrameworkAdapter

class NonPythonAdapter(BaseFrameworkAdapter):
    name = "external_framework"

    def is_available(self) -> bool:
        try:
            subprocess.run(["external_framework", "--version"],
                         capture_output=True, check=True)
            return True
        except (subprocess.CalledProcessError, FileNotFoundError):
            return False

    def run(
        self,
        config: Dict[str, Any],
        llm_config: List[Dict],
        topic: str,
        *,
        tools_dict: Optional[Dict[str, Any]] = None,
        agent_callback: Optional[Callable] = None,
        task_callback: Optional[Callable] = None,
        cli_config: Optional[Dict[str, Any]] = None,
    ) -> str:
        # Convert config to external format
        cmd = ["external_framework", "run", "--topic", topic]
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout

import logging
from typing import Any, Callable, Dict, List, Optional
from praisonaiagents import BaseFrameworkAdapter

logger = logging.getLogger(__name__)

class OptionalDepsAdapter(BaseFrameworkAdapter):
    name = "advanced_framework"

    def is_available(self) -> bool:
        try:
            # Check multiple optional dependencies
            import advanced_framework
            import optional_plugin
            return True
        except ImportError as e:
            logger.debug(f"Framework not available: {e}")
            return False

    def run(
        self,
        config: Dict[str, Any],
        llm_config: List[Dict],
        topic: str,
        *,
        tools_dict: Optional[Dict[str, Any]] = None,
        agent_callback: Optional[Callable] = None,
        task_callback: Optional[Callable] = None,
        cli_config: Optional[Dict[str, Any]] = None,
    ) -> str:
        if not self.is_available():
            raise RuntimeError("Framework dependencies not installed")

        import advanced_framework
        return advanced_framework.execute(config, llm_config, topic)

from praisonai.framework_adapters.registry import (
    list_framework_choices,
    list_available_frameworks,
    get_install_hint,
)

from praisonai.framework_adapters.registry import list_framework_choices, get_install_hint

# See which frameworks appear in --framework choices
choices = list_framework_choices()
print(choices)  # ['crewai', 'praisonai', 'autogen', 'my_framework']

# Get install instructions for a missing framework
hint = get_install_hint("crewai")
print(hint)  # pip install 'praisonai-frameworks[crewai]'

DEFAULT_PRIORITY: tuple[str, ...] = ("crewai", "praisonai", "autogen")

from praisonai.framework_adapters.registry import get_default_registry

registry = get_default_registry()
default = registry.pick_default()   # returns "crewai", "praisonai", etc.
print(f"Using framework: {default}")

from praisonai.framework_adapters.registry import get_default_registry

# Scenario: only your plugin adapter is installed
registry = get_default_registry()
default = registry.pick_default()  # returns your plugin adapter name

from praisonai.framework_adapters.registry import FrameworkAdapterRegistry
from praisonai.agents_generator import AgentsGenerator

# Tenant-isolated registry — NOT shared with other tenants in the same process
tenant_registry = FrameworkAdapterRegistry()
tenant_registry.register("crewai", CustomCrewAIAdapter)

generator = AgentsGenerator(
    agent_file="agents.yaml",
    framework="crewai",
    config_list=[...],
    adapter_registry=tenant_registry,  # <-- new in #1639
)

from praisonai.framework_adapters.registry import FrameworkAdapterRegistry
from praisonai.auto import AutoGenerator

reg = FrameworkAdapterRegistry()
auto = AutoGenerator(
    topic="...",
    agent_file="test.yaml",
    framework="crewai",
    config_list=[...],
    adapter_registry=reg,
)