Merge pull request #1334 from MervinPraison/claude/issue-1331-20260409-0930

feat: Recipe and Gateway Integration with Inbox & Stepper Pattern

Author: MervinPraison

src/praisonai-agents/praisonaiagents/gateway/adapters/__init__.py

@@ -0,0 +1,9 @@
+"""
+Gateway adapters package for PraisonAI.
+
+This package contains adapters that bridge different systems with the Gateway infrastructure.
+"""
+
+from .recipe_adapter import RecipeBotAdapter
+
+__all__ = ["RecipeBotAdapter"]

src/praisonai-agents/praisonaiagents/gateway/adapters/recipe_adapter.py

@@ -0,0 +1,56 @@
+"""
+Adapter to wrap a RecipeRuntime behind the standard Agent duck-typing for Gateway and BotOS.
+"""
+
+import logging
+from typing import Optional, Any
+from praisonaiagents.streaming import StreamEventEmitter
+
+logger = logging.getLogger(__name__)
+
+class RecipeBotAdapter:
+    """Wraps RecipeRuntime to act like an Agent for gateway/bot compatibility."""
+    
+    def __init__(self, recipe_name: str, **kwargs):
+        self.recipe_name = recipe_name
+        self.config = kwargs
+        self._runtime = None
+        self.stream_emitter = StreamEventEmitter()
+    
+    @property
+    def name(self) -> str:
+        return f"recipe-{self.recipe_name}"
+        
+    def _get_runtime(self):
+        """Lazy load and instantiate RecipeRuntime."""
+        if self._runtime is None:
+            try:
+                from praisonai.recipe.core import RecipeRuntime
+                self._runtime = RecipeRuntime(recipe_name=self.recipe_name, **self.config)
+            except ImportError as e:
+                raise ImportError(
+                    f"Failed to load PraisonAI recipe components: {e}. "
+                    "Make sure you have praisonai installed."
+                ) from e
+        return self._runtime
+        
+    def chat(self, message: str) -> str:
+        """Process the message via the recipe runtime."""
+        try:
+            runtime = self._get_runtime()
+            
+            # Forward the user_input to the recipe's variables
+            variables = runtime.context.variables if hasattr(runtime, 'context') else {}
+            variables["user_input"] = message
+            
+            # Currently we rely on string return matching standard Agent
+            # Note: We don't have direct streaming from recipe to emitter yet; this binds them generically.
+            result = runtime.run()
+            
+            if hasattr(result, "final_output"):
+                return str(result.final_output)
+            return str(result)
+            
+        except Exception as e:
+            logger.error(f"Recipe execution failed: {e}")
+            return f"Error executing recipe: {e}"

src/praisonai/praisonai/cli/commands/serve.py

@@ -478,6 +478,7 @@ def serve_scheduler(
 def serve_recipe(
     host: str = typer.Option("127.0.0.1", "--host", "-h", help="Host to bind to"),
     port: int = typer.Option(8765, "--port", "-p", help="Port to bind to"),
+    recipe: Optional[str] = typer.Option(None, "--recipe", "-r", help="Recipe name to execute"),
     config: Optional[str] = typer.Option(None, "--config", "-c", help="Config file path"),
     reload: bool = typer.Option(False, "--reload", help="Enable auto-reload"),
 ):
@@ -490,6 +491,8 @@ def serve_recipe(
     try:
         from ..features.serve import handle_serve_command
         args = ["recipe", "--host", host, "--port", str(port)]
+        if recipe:
+            args.extend(["--recipe", recipe])
         if config:
             args.extend(["--config", config])
         if reload:

src/praisonai/praisonai/cli/features/bots_cli.py

@@ -59,6 +59,9 @@ class BotCapabilities:
     exec_enabled: bool = False
     auto_approve: bool = False
 
+    # Optional primitive override
+    recipe: Optional[str] = None
+    
     # Model settings
     model: Optional[str] = None
     thinking: Optional[str] = None  # off, minimal, low, medium, high
@@ -750,6 +753,11 @@ def _load_agent(
         """
         from praisonaiagents import Agent
 
+        # Intercept Recipe Adapter
+        if capabilities and capabilities.recipe:
+            from praisonaiagents.gateway.adapters.recipe_adapter import RecipeBotAdapter
+            return RecipeBotAdapter(recipe_name=capabilities.recipe)
+            
         # Build tools list from capabilities
         tools = self._build_tools(capabilities) if capabilities else []
 
@@ -977,6 +985,7 @@ def _add_capability_args(parser) -> None:
     """Add capability arguments to a parser."""
     # Agent configuration
     parser.add_argument("--agent", help="Path to agent YAML configuration file")
+    parser.add_argument("--recipe", help="Recipe name to execute as the bot")
     parser.add_argument("--model", "-m", help="LLM model to use")
 
     # Browser
@@ -1050,6 +1059,7 @@ def _build_capabilities_from_args(args) -> BotCapabilities:
         stt_model=getattr(args, "stt_model", None),
         session_id=getattr(args, "session_id", None),
         user_id=getattr(args, "user_id", None),
+        recipe=getattr(args, "recipe", None),
     )
 
 

src/praisonai/praisonai/cli/features/serve.py

@@ -310,34 +310,33 @@ async def root():
         return app
 
     def cmd_recipe(self, args: List[str]) -> int:
-        """Launch recipe runner server."""
+        """Launch recipe runner server via WebSocketGateway."""
         spec = {
             "host": {"default": self.DEFAULT_HOST},
             "port": {"default": self.DEFAULT_PORT, "type": "int"},
-            "config": {"default": None},
+            "recipe": {"default": None},
             "reload": {"flag": True, "default": False},
-            "api_key": {"default": None},
-            "preload": {"flag": True, "default": False},
         }
         parsed = self._parse_args(args, spec)
 
-        try:
-            self._print_success(f"Starting recipe server on {parsed['host']}:{parsed['port']}")
-            print("  Discovery: /__praisonai__/discovery")
-            
-            from praisonai.recipe.serve import serve, load_config
+        recipe_name = parsed.get("recipe")
+        if not recipe_name:
+            self._print_error("A recipe name is required (e.g. --recipe my-recipe)")
+            return self.EXIT_VALIDATION_ERROR
 
-            # Load config
-            config = load_config(parsed["config"]) if parsed["config"] else {}
-            if parsed["api_key"]:
-                config["api_key"] = parsed["api_key"]
+        try:
+            self._print_success(f"Starting recipe gateway on {parsed['host']}:{parsed['port']}")
+            import asyncio
+            from praisonai.gateway import WebSocketGateway
+            from praisonaiagents.gateway.adapters.recipe_adapter import RecipeBotAdapter
 
-            serve(
-                host=parsed["host"],
-                port=parsed["port"],
-                reload=parsed["reload"],
-                config=config,
-            )
+            async def _run():
+                gateway = WebSocketGateway(host=parsed['host'], port=parsed['port'])
+                agent = RecipeBotAdapter(recipe_name=recipe_name)
+                gateway.register_agent(agent)
+                await gateway.start()
+                
+            asyncio.run(_run())
 
         except ImportError as e:
             self._print_error(f"Missing dependency: {e}")

src/praisonai/praisonai/gateway/server.py

@@ -44,6 +44,10 @@ class GatewaySession:
     _messages: List[GatewayMessage] = field(default_factory=list)
     _max_messages: int = 1000
 
+    # Stepper & Concurrency logic
+    _inbox: asyncio.Queue = field(default_factory=asyncio.Queue)
+    _is_executing: bool = False
+    
     @property
     def session_id(self) -> str:
         return self._session_id
@@ -89,6 +93,20 @@ def get_messages(self, limit: Optional[int] = None) -> List[GatewayMessage]:
     def close(self) -> None:
         self._is_active = False
 
+    async def queue_message(self, message: str) -> None:
+        """Queue a user message for execution after the current operation."""
+        await self._inbox.put(message)
+        
+    def get_next_message(self) -> Optional[str]:
+        """Fetch the next queued message if available without blocking."""
+        if self._inbox.empty():
+            return None
+        return self._inbox.get_nowait()
+        
+    def mark_executing(self, status: bool) -> None:
+        """Mark the session as currently executing an agent workflow."""
+        self._is_executing = status
+
 
 class WebSocketGateway:
     """WebSocket gateway server for multi-agent coordination.
@@ -564,40 +582,72 @@ async def _process_agent_message(
             return "Agent not available"
 
         client_id = session.client_id
+        content = message.content if isinstance(message.content, str) else str(message.content)
 
-        try:
-            content = message.content if isinstance(message.content, str) else str(message.content)
-            
-            # Wire streaming relay if agent has a stream_emitter
-            relay_callback = None
-            emitter = getattr(agent, 'stream_emitter', None)
-            if emitter is not None and client_id:
-                relay_callback = self._make_stream_relay(client_id, session)
-                emitter.add_callback(relay_callback)
-            
-            try:
-                loop = asyncio.get_event_loop()
-                response = await loop.run_in_executor(None, agent.chat, content)
-            finally:
-                # Always clean up the relay callback
-                if relay_callback and emitter is not None:
-                    try:
-                        emitter.remove_callback(relay_callback)
-                    except (ValueError, AttributeError):
-                        pass
-            
-            response_message = GatewayMessage(
-                content=response,
-                sender_id=session.agent_id,
-                session_id=session.session_id,
-                reply_to=message.message_id,
+        # Inbox & Stepper logic
+        if session._is_executing:
+            # Send an ephemeral status event
+            await self._send_to_client(
+                client_id,
+                {
+                    "type": "status",
+                    "source": session.agent_id,
+                    "message": "Thinking... (I've added your new message to the queue to process next)."
+                }
             )
-            session.add_message(response_message)
+            await session.queue_message(content)
+            return "Message queued."
 
-            return response
-        except Exception as e:
-            logger.error(f"Agent error: {e}")
-            return f"Error: {str(e)}"
+        session.mark_executing(True)
+        await session.queue_message(content)
+        
+        # Start background task to process the queue
+        asyncio.create_task(self._run_session_queue(session, agent, client_id))
+        return "Started processing."
+
+    async def _run_session_queue(self, session: GatewaySession, agent: Any, client_id: str) -> None:
+        """Background task loop that constantly pulls from `_inbox` and executes the agent task."""
+        try:
+            while True:
+                content = session.get_next_message()
+                if not content:
+                    break  # Queue is empty, exit loop
+                
+                # Wire streaming relay if agent has a stream_emitter
+                relay_callback = None
+                emitter = getattr(agent, 'stream_emitter', None)
+                if emitter is not None and client_id:
+                    relay_callback = self._make_stream_relay(client_id, session)
+                    emitter.add_callback(relay_callback)
+                
+                try:
+                    loop = asyncio.get_event_loop()
+                    response = await loop.run_in_executor(None, agent.chat, content)
+                except Exception as e:
+                    logger.error(f"Agent error in queue processor: {e}")
+                    response = f"Error: {str(e)}"
+                finally:
+                    # Always clean up the relay callback
+                    if relay_callback and emitter is not None:
+                        try:
+                            emitter.remove_callback(relay_callback)
+                        except (ValueError, AttributeError):
+                            pass
+                
+                response_message = GatewayMessage(
+                    content=response,
+                    sender_id=session.agent_id,
+                    session_id=session.session_id,
+                )
+                session.add_message(response_message)
+                
+                await self._send_to_client(client_id, {
+                    "type": "response",
+                    "content": response,
+                    "session_id": session.session_id,
+                })
+        finally:
+            session.mark_executing(False)
 
     def _make_stream_relay(
         self, client_id: str, session: "GatewaySession"