Framework Adapter Plugins

Extend PraisonAI with custom framework adapters via Python entry points

Framework adapter plugins enable third-party developers to add new execution frameworks to PraisonAI without modifying core code.

from praisonaiagents import Agent

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

The user installs a third-party adapter package; entry-point discovery registers it for CLI framework selection.

Quick Start

1. Programmatic Registration

   Register a framework adapter directly in your code:

   from typing import Any, Callable, Dict, List, Optional
   from praisonaiagents import BaseFrameworkAdapter
   from praisonai.framework_adapters.registry import get_default_registry
   
   class MyFrameworkAdapter(BaseFrameworkAdapter):
       name = "myframework"
   
       def is_available(self) -> bool:
           try:
               import myframework
               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:
           return "Framework execution result"
   
   # Register the adapter (preferred for app code)
   registry = get_default_registry()
   registry.register("myframework", MyFrameworkAdapter)

2. Entry-Point Plugin (installable)

   Create a pip-installable plugin — no manual registry.register() call needed:

   # my_framework_adapter.py
   from praisonaiagents import BaseFrameworkAdapter
   
   class MyFrameworkAdapter(BaseFrameworkAdapter):
       name = "my_framework"
       install_hint = "pip install my-framework-bridge"
       requires_tools_extra = False
   
       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"my_framework ran: {topic}"

   # pyproject.toml
   [build-system]
   requires = ["setuptools>=61.0"]
   build-backend = "setuptools.build_meta"
   
   [project]
   name = "my-framework-bridge"
   version = "0.0.1"
   dependencies = ["praisonaiagents"]
   
   [tool.setuptools]
   py-modules = ["my_framework_adapter"]
   
   [project.entry-points."praisonai.framework_adapters"]
   my_framework = "my_framework_adapter:MyFrameworkAdapter"

   pip install -e ./examples/python/frameworks/custom_adapter
   praisonai --framework my_framework agents.yaml

   After pip install, the adapter appears in praisonai --framework choices and praisonai doctor automatically — no manual registry.register() call needed.

What the Registry Controls

Registering through the praisonai.framework_adapters entry-point group gives your adapter automatic visibility across three surfaces — no extra wiring required:

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

See Framework Availability — Dynamic Discovery for more details on how the registry feeds these surfaces.

How It Works

The framework adapter registry provides a central point for managing framework implementations:

Unsupported-field warnings (PraisonAI #3771)

When a YAML config declares a field a specific framework adapter doesn’t wire — for example approval: on the CrewAI or AutoGen adapter — the adapter now emits a logging.warning naming the dropped field(s). Previously the field was silently ignored, so users who wrote approval: and wondered why nothing happened had no signal in the logs.

• PraisonAI-native (framework: praisonai) — no warning; the native runner supports every declared field.
• CrewAI / AutoGen — warning logged for each unsupported field per agent/task.

The SUPPORTED_YAML_FIELDS contract (PraisonAI #4651)

Each adapter declares the extended YAML fields it honours as a class attribute — the single source of truth is the adapter that reads a field. PraisonAI PR #4651 replaced the old module-level _ADAPTER_SUPPORTED_FIELDS dict with this per-class set.

from typing import ClassVar, FrozenSet
from praisonai.framework_adapters.base import BaseFrameworkAdapter

class MyFrameworkAdapter(BaseFrameworkAdapter):
    name = "myframework"
    # Extended fields this backend actually consumes, beyond the structural
    # keys (tools, tasks, role, goal, backstory). Empty set = "supports
    # everything" (the praisonai backend).
    SUPPORTED_YAML_FIELDS: ClassVar[FrozenSet[str]] = frozenset({"llm"})

warn_unsupported_fields(adapter, extras) reads adapter.SUPPORTED_YAML_FIELDS directly. Any per-agent YAML key not in this set — after the structural keys tools, tasks, role, goal, backstory are removed — is logged once per agent:

framework='crewai' ignores YAML field(s) ['approval'] for agent 'Researcher'; these are only honoured by framework=praisonai.

The two built-in overrides are working templates for third-party authors:

# crewai_adapter.py
SUPPORTED_YAML_FIELDS = frozenset({
    "allow_delegation", "max_iter", "max_rpm", "max_execution_time",
    "verbose", "cache", "system_template", "prompt_template",
    "response_template", "llm", "function_calling_llm",
})

# autogen_adapter.py — shared by AutoGenAdapter, AutoGenV4Adapter,
# AG2Adapter, and AutoGenFamilyAdapter
SUPPORTED_YAML_FIELDS = frozenset({"llm", "function_calling_llm"})

AutoGen callback parity

• agent_callback now fires per agent (PraisonAI #3771). The AutoGen adapter previously declared but never invoked agent_callback. It now runs once per agent, matching the CrewAI adapter’s behaviour.
• task_callback is now wired for the AutoGen v0.2 adapter (PraisonAI PR #5403), restoring parity with CrewAI. AutoGen v0.2’s initiate_chats has no per-step hook, so the adapter fans the callback out itself: it fires {"event": "task_prepared", ...} once per prepared chat task before initiate_chats, and {"event": "task_completed", ...} once per chat after initiate_chats returns. Sync callbacks and async def callbacks both work — an async callback is routed through dispatch_maybe_awaitable so the returned coroutine actually runs instead of being discarded. Callback exceptions are caught and logged; they never abort the chat.

from praisonaiagents import Agent

def on_task(event):
    # event ∈ {"task_prepared", "task_completed"}
    print(event["event"], event.get("summary"))

Wire task_callback the same way as the other adapter callbacks — through the orchestrator / adapter contract (see Auto-Discovery in the CLI and the protocol table below for the exact plumbing).

task_callback event payload

Auto-Discovery in the CLI

Once your adapter is registered (in-process via register() or via the praisonai.framework_adapters entry-point group), it appears automatically wherever framework names are listed or validated — no further wiring needed:

The single source of truth is praisonai.framework_adapters.list_framework_choices(). When include_unavailable=False (the default) only adapters whose is_available() returns True are returned; the CLI uses include_unavailable=True so users see every registered framework and get a friendly error if its dependencies are missing.

Declaring an install hint

Surface a tailored install command for your adapter so users don’t see the generic fallback:

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: ...

When myframework is selected but not installed, the CLI now raises:

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

If you omit install_hint, the wrapper falls back to pip install 'praisonai-frameworks[myframework]'.

Missing optional dependencies no longer leak

FrameworkAdapterRegistry.is_available() now catches ImportError raised when an adapter’s constructor touches a missing optional dependency, alongside the existing ValueError/TypeError. A plugin with unmet deps reports as unavailable rather than crashing the CLI list, the doctor check, or pick_default().

The registry now distinguishes structural failures (ImportError, ValueError, TypeError — the optional dep is genuinely not installed, cached for the process) from transient failures (any other exception — e.g. a racing entry-point scan, a boot-time healthcheck timeout, cached only for _NEG_CACHE_TTL = 60 s). A one-off flaky probe therefore no longer pins a framework to unavailable for the rest of the process.

Adapter authors: keep is_available() cheap and side-effect-free. If you must raise on infrastructure hiccups (network probes, entry-point scans mid-write), raising the default Exception subclass is correct — the registry will retry after the TTL. Do not raise ImportError for transient causes, or the framework will be hidden for the whole process.

Adapter Resolution Errors

When the requested framework can’t be resolved — the entry-point package isn’t installed, the adapter dependency is missing, the framework name is typo’d, or the generator is running in minimal/mock mode without an _adapter_registry — PraisonAI logs a Could not resolve framework adapter for <framework>: <error> warning and falls back to native execution. Previously this was swallowed silently. If you see this warning: install the plugin, correct the framework name, or drop the --framework flag to run natively.

Helpers exposed at the package root

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,
)

Configuration

Registry Methods

Adapter Protocol

Framework adapters must implement the FrameworkAdapter protocol:

Protocol Validation (PR #2083)

Since PR #2083, FrameworkAdapterRegistry.create() validates every adapter at construction time. If the run() method does not accept the four parameters tools_dict, agent_callback, task_callback, and cli_config, instantiation fails with a per-method TypeError:

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

The error names the failing method (.run or .arun), so arun() is now validated with the same rules whenever it exists. A run() (or arun()) that declares a **kwargs catch-all satisfies the check without listing each protocol keyword explicitly.

Adapters that fail validation also report registry.is_available("<name>") == False instead of raising — so a broken plugin will simply disappear from the available-framework list rather than crashing the CLI.

**kwargs-only adapters are accepted but warned

A signature that relies on kwargs still validates for backward compatibility — an adapter can legitimately forward kwargs to a delegate. But if run() / arun() does not name every one of tools_dict, agent_callback, task_callback, and cli_config explicitly and instead leans on **kwargs, registration now logs a warning naming the un-declared kwargs. If those kwargs aren’t forwarded, their safety guarantees — tool_timeout wraps, agent_callback / task_callback observability hooks, and cli_config approval / guardrails / allowed-tools knobs — are silently dropped.

FrameworkAdapter 'my_adapter'.run names none of the adapter kwargs
['agent_callback', 'cli_config', 'task_callback', 'tools_dict']
explicitly and relies on **kwargs; if these are not forwarded, their
safety guarantees (tool_timeout, callbacks, approval/guardrails via
cli_config) will be silently dropped.

The warning fires whenever the signature has a kwargs catch-all and** is missing one or more of the four named kwargs. The log line lists exactly the missing ones. Built-in adapters (crewai, autogen, autogen-v4, ag2, praisonai) name all four explicitly, so no warning fires for them.

Default arun() offloads run() to a bounded per-adapter thread pool, allowing sync adapters to be called from async contexts without blocking the event loop. Unlike asyncio.to_thread (which shares the process-wide executor), each adapter gets its own ThreadPoolExecutor sized by _THREAD_OFFLOAD_MAX_WORKERS (default 4), so one slow framework run cannot starve every other async caller.

Override arun() with a native async def and set SUPPORTS_ASYNC = True to skip the thread offload entirely and use the framework’s own async API.

Async capability flags

Set SUPPORTS_ASYNC = True when you provide a native async def arun(); leave it False to inherit the bounded thread-pool offload.

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): ...

Capability Flags

Capability flags are class-level booleans that tell PraisonAI which features your adapter can execute.

Four flags gate distinct wrapper features. All default to False, so an adapter opts in only to what it can actually run.

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
    SUPPORTS_CORE_TOOL_TIMEOUT = True  # only if run() forwards tool_config to a core Agent

    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}"

The built-in PraisonAIAdapter sets SUPPORTS_WORKFLOW, SUPPORTS_RUNTIME_FEATURES, and SUPPORTS_CORE_TOOL_TIMEOUT to True. Third-party adapters registered via the praisonai.framework_adapters entry-point group — and any subclass of PraisonAIAdapter, such as a renamed fork — opt in by setting the flags on their class.

A renamed fork inherits every flag with no extra work:

from praisonai.framework_adapters.praisonai_adapter import PraisonAIAdapter

class CompanyNativeAdapter(PraisonAIAdapter):
    # Inherits SUPPORTS_WORKFLOW, SUPPORTS_RUNTIME_FEATURES, SUPPORTS_ASYNC,
    # SUPPORTS_CORE_TOOL_TIMEOUT, SUPPORTS_SESSION_CONTINUITY, SUPPORTS_STREAM_BRIDGE
    name = "company_native"

Each flag maps to the wrapper call site that reads it:

Adding an Async Path

1. Sync-only adapter (default)

   Leave arun() unoverridden and SUPPORTS_ASYNC at its False default. The base class offloads run() to the adapter’s bounded thread pool automatically:

   from praisonaiagents import BaseFrameworkAdapter
   
   class MySyncAdapter(BaseFrameworkAdapter):
       name = "my_sync_framework"
   
       def is_available(self) -> bool:
           try:
               import my_sync_framework
               return True
           except ImportError:
               return False
   
       def run(
           self,
           config,
           llm_config,
           topic,
           *,
           tools_dict=None,
           agent_callback=None,
           task_callback=None,
           cli_config=None,
       ) -> str:
           import my_sync_framework
           return my_sync_framework.execute(config, topic)

2. Native async adapter

   Override arun() with a real async def and set SUPPORTS_ASYNC = True to use the framework’s native async API:

   from praisonaiagents import BaseFrameworkAdapter
   
   class MyAsyncAdapter(BaseFrameworkAdapter):
       name = "my_async_framework"
       SUPPORTS_ASYNC = True
   
       def is_available(self) -> bool:
           try:
               import my_async_framework
               return True
           except ImportError:
               return False
   
       def run(
           self,
           config,
           llm_config,
           topic,
           *,
           tools_dict=None,
           agent_callback=None,
           task_callback=None,
           cli_config=None,
       ) -> str:
           import asyncio
           return asyncio.run(self.arun(config, llm_config, topic))
   
       async def arun(
           self,
           config,
           llm_config,
           topic,
           *,
           tools_dict=None,
           agent_callback=None,
           task_callback=None,
           cli_config=None,
       ) -> str:
           import my_async_framework
           return await my_async_framework.aexecute(config, topic)

Family Router Pattern — AutoGenFamilyAdapter as Canonical Example

A family adapter routes to a concrete sibling adapter based on environment or config. It overrides resolve_alias() to return an adapter name, then resolve() uses that name to look up and return the concrete instance.

The router consults the owning registry — the registry that constructed this adapter — with get_default_registry() as the fallback when the adapter was constructed outside a registry (standalone use). FrameworkAdapterRegistry.create() post-attaches itself as adapter._owning_registry, so router adapters honour the same registry that instantiated them without any explicit plumbing. A module-level _owning_registry(adapter) helper returns adapter._owning_registry when present, else the process-default registry; is_available(), resolve_alias(), and resolve() all route through it.

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)
        # _owning_registry(self) returns self._owning_registry when the registry
        # that built this adapter post-attached it, else the process default.
        return _owning_registry(self).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."
        )

Key points:

• resolve_alias(config) accepts optional config dict — config autogen_version key wins over AUTOGEN_VERSION env var
• resolve() passes config to resolve_alias(), then fetches the concrete adapter from the owning registry (get_default_registry() fallback for standalone use)
• run() has the full four keyword-only kwargs and raises RuntimeError — the orchestrator calls resolve() first, then calls run() on the returned concrete adapter

Optional Adapter Hooks

After the existing “Adapter protocol” section, two new optional methods were added in PR #1763:

**resolve(*, config=None)** — pick a concrete variant (PR #2070).

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

Worked example — AutoGenAdapter.resolve: the orchestrator passes YAML config; the adapter instantiates siblings directly:

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 ...

**setup(*, framework_tag)** — pre-run hook.

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

As of PR #3152, setup() is invoked inside observability_session, so any events emitted by setup() — and any exception it raises — are captured and finalized. Custom adapters should therefore not open their own observability session inside setup().

Orchestrator Pipeline (Post-#3152)

As of PR #3152, AgentsGenerator owns the observability lifecycle. Both the sync (generate_crew_and_kickoff) and async (agenerate_crew_and_kickoff) run paths bracket adapter.setup() and adapter.run()/adapter.arun() in a single observability_session, so setup and run share one session and finalize is impossible to forget. The orchestrator now follows this sequence:

Common Patterns

Override Built-in Adapter

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)

Subprocess Framework Wrapper

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

Conditional Dependencies

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)

Best Practices

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

def is_available(self) -> bool:
    try:
        # Check all required dependencies
        import required_framework
        import optional_dependency

        # Verify minimum versions if needed
        if hasattr(required_framework, 'version'):
            version = required_framework.version
            if version < (1, 0, 0):
                return False

        return True
    except ImportError:
        # Log debug info, don't raise
        logging.getLogger(__name__).debug(
            "Framework dependencies not available"
        )
        return False

# ❌ Bad - imports at module level
import heavy_framework
from expensive.module import Component

class BadAdapter(BaseFrameworkAdapter):
    pass

# ✅ Good - lazy imports
class GoodAdapter(BaseFrameworkAdapter):
    def run(self, config, llm_config, topic):
        # Import only when needed
        import heavy_framework
        from expensive.module import Component
        return heavy_framework.run(config)

import logging
from praisonaiagents import BaseFrameworkAdapter

class LoggingAdapter(BaseFrameworkAdapter):
    def __init__(self):
        super().__init__()
        self.logger = logging.getLogger(__name__)

    def run(self, config, llm_config, topic):
        self.logger.info(f"Starting {self.name} execution for topic: {topic}")
        try:
            result = self._execute_framework(config, llm_config, topic)
            self.logger.info(f"Execution completed successfully")
            return result
        except Exception as e:
            self.logger.error(f"Framework execution failed: {e}")
            raise

class ResourceAwareAdapter(BaseFrameworkAdapter):
    def __init__(self):
        super().__init__()
        self._connections = []
        self._temp_files = []

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

        temp_file = self._create_temp_file()
        self._temp_files.append(temp_file)

        return result

    def cleanup(self) -> None:
        for conn in self._connections:
            try:
                conn.close()
            except Exception:
                pass

        for temp_file in self._temp_files:
            try:
                temp_file.unlink()
            except Exception:
                pass

        self._connections.clear()
        self._temp_files.clear()
        super().cleanup()

import importlib.util

def is_available(self) -> bool:
    return importlib.util.find_spec("myframework") is not None

from praisonaiagents import BaseFrameworkAdapter

class ScalableAsyncAdapter(BaseFrameworkAdapter):
    name = "scalable_framework"
    SUPPORTS_ASYNC = True

    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:
        import asyncio
        return asyncio.run(self.arun(config, llm_config, topic))

    async def arun(self, config, llm_config, topic, *, tools_dict=None,
                   agent_callback=None, task_callback=None, cli_config=None) -> str:
        import scalable_framework
        return await scalable_framework.arun(config, topic)

Registry Helpers

Registry-level helpers for building CLIs and plugins:

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 Framework Selection

FrameworkAdapterRegistry.pick_default() selects the default framework when none is specified.

DEFAULT_PRIORITY is a class attribute on FrameworkAdapterRegistry:

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

pick_default() walks this tuple and returns the first available adapter. If none of the three built-ins are installed, it falls back to any other registered adapter — including entry-point plugins. If no adapter is available at all, it raises RuntimeError.

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}")

_entrypoint.run() and arun() now delegate to pick_default(). This means an entry-point adapter can become the default framework once the four built-ins are uninstalled:

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

resolve_or_default() — for optional framework inputs

When a caller may or may not pass a framework name, use resolve_or_default() — it returns the caller’s choice if given, or the registry’s pick_default() otherwise.

FrameworkAdapterRegistry.resolve_or_default(name: Optional[str]) -> str

resolve_or_default() now:

• strips whitespace from truthy strings (" autogen " → "autogen"), matching how pick_default() normalises its own output.
• falls back to DEFAULT_PRIORITY[0] when pick_default() raises RuntimeError (no adapter installed), instead of propagating — so downstream availability validation gets a chance to raise its own actionable error rather than this helper masking it.

from praisonai.framework_adapters.registry import get_default_registry

registry = get_default_registry()

registry.resolve_or_default("crewai")   # → "crewai"
registry.resolve_or_default(None)       # → whatever pick_default() picks

As of PR #5039 (Gap 1 of #5016), every wrapper entry point that used to fall back to a hardcoded "praisonai" now goes through the registry:

Inject Your Own Registry (Multi-tenant / Tests)

The new adapter_registry= kwarg on AgentsGenerator and AutoGenerator enables dependency injection for multi-tenant applications and testing:

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
)

For AutoGenerator:

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,
)

For workflow YAML, use WorkflowAutoGenerator — its framework= kwarg is honoured end-to-end (see AutoAgents).

Framework validation honours the injected registry

As of PR #3152, framework-availability validation is DI-aware — a scoped/per-tenant adapter registered on the generator’s own registry is no longer rejected just because it is absent from the process-default registry.

• assert_framework_available(name, , registry=None) and get_install_hint(name, , registry=None) now accept an optional registry kwarg (default: process-default).
• AgentsGenerator._prepare_for_run threads its injected _adapter_registry through into assert_framework_available, so validation consults the tenant registry, not the process-default.
• The install hint is derived from the same registry that failed validation (DI end to end).
• Fallback for resolved built-in names: if the injected registry does not know a name but the process-default does (typical when a family router resolves autogen → autogen_v2), validation falls back to the default and the framework is accepted. This preserves the DI seam without rejecting resolved built-ins.

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

tenant_registry = FrameworkAdapterRegistry()
tenant_registry.register("tenant_scoped_fw", TenantScopedAdapter)

gen = AgentsGenerator(
    agent_file="agents.yaml",
    framework="tenant_scoped_fw",
    config_list=[...],
    adapter_registry=tenant_registry,   # DI seam
)
# Validation now consults tenant_registry, not the process-default.

Troubleshooting

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

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

def run(self, config, llm_config, topic, **kwargs):  # ✅ now accepted
    tools_dict = kwargs.get("tools_dict")
    agent_callback = kwargs.get("agent_callback")
    ...

class MyAdapter(BaseFrameworkAdapter):
    SUPPORTS_ASYNC = True

    async def arun(self, config, llm_config, topic, *, tools_dict=None,
                   agent_callback=None, task_callback=None, cli_config=None) -> str:
        return await my_native_async_call(config, topic)