feat: Enhance Agent and Crew functionality with context management and execution plan support

- Added context management capabilities to the Agent class, allowing for dynamic context configuration.
- Introduced ExecutionPlan class for managing multi-step workflows, including initialization and progress tracking.
- Updated Crew class to synchronize message deletions in subgraphs and validate HITL configurations based on execution mode.
- Enhanced various tools and message handling for improved integration and user notifications.
- Added new dependencies in pyproject.toml for additional functionality.

Author: lihao

examples/components/mcp/src/mcp/crew.py

@@ -8,6 +8,7 @@
 from langcrew.project import CrewBase, agent, crew, task
 
 
+
 @CrewBase
 class MapCrew:
     """Map crew"""
@@ -31,7 +32,7 @@ def planner(self) -> Agent:
         return Agent(
             config=self.agents_config["planner"],
             mcp_servers=mcp_server_configs,
-            mcp_tool_filter=["xxx"],  # 只引入固定的tool
+            mcp_tool_filter=["xxx"], #只引入固定的tool
             llm=self._get_default_llm(),
             verbose=True,
         )
@@ -82,7 +83,7 @@ def planner(self) -> Agent:
             "url": f"https://mcp.amap.com/mcp?key={os.getenv('AMAP_TOKEN')}",
             "transport": "streamable_http",
         }
-
+        
         mcp_server_configs = {"amap-streamable_http": server_config}
         return Agent(
             config=self.agents_config["planner"],

examples/components/mcp/src/mcp/main.py

@@ -7,34 +7,36 @@
 
 load_dotenv(os.path.join(os.path.dirname(__file__), "..", "..", ".env.example"))
 
-
 # sse use case
 async def map_sse_mcp():
-    inputs = {"user_instruction": "Optimal route from Shanghai to Beijing"}
-    crew = MapCrew().crew()
-    result = await crew.akickoff(inputs=inputs)
+    inputs = {
+        "user_instruction": "Optimal route from Shanghai to Beijing"
+    }
+    crew =  MapCrew().crew()
+    result= await crew.akickoff(inputs=inputs)
     print(f"map result: {result}")
 
-
 # streamHttp use case
 async def map_streamHttp_mcp():
-    inputs = {"user_instruction": "Optimal route from Shanghai to Beijing"}
-    crew = MapStreamHttpCrew().crew()
-    result = await crew.akickoff(inputs=inputs)
+    inputs = {
+        "user_instruction": "Optimal route from Shanghai to Beijing"
+    }
+    crew =  MapStreamHttpCrew().crew()
+    result= await crew.akickoff(inputs=inputs)
     print(f"map result: {result}")
 
-
 # stdio use case
 async def calculator_stdio_mcp():
     inputs = {
         "user_instruction": "Calculate 100+100 using tools, and explain which tool you used"
     }
-    crew = CalculatorCrew().crew()
-    result = await crew.akickoff(inputs=inputs)
+    crew =  CalculatorCrew().crew()
+    result= await crew.akickoff(inputs=inputs)
     print(f"calculator result: {result}")
 
 
 if __name__ == "__main__":
     # asyncio.run(map_sse_mcp())
     # asyncio.run(map_streamHttp_mcp())
     asyncio.run(calculator_stdio_mcp())
+    

examples/components/mcp/src/mcp/tools/calculator_stdio.py

@@ -7,7 +7,7 @@
 
 @mcp.tool()
 def add(a: int, b: int) -> int:
-    """Add two numbers together"""
+    """Add two numbers together"""    
     return a + b
 
 
@@ -33,4 +33,4 @@ def divide(a: float, b: float) -> float:
 
 if __name__ == "__main__":
     # Start server using stdio transport
-    mcp.run(transport="stdio")
+    mcp.run(transport="stdio")

examples/components/web/super_agent/.env.example

@@ -30,4 +30,9 @@ IN_DOCKER=true
 ANONYMIZED_TELEMETRY=false
 BROWSER_USE_LOGGING_LEVEL=debug
 BROWSER_USE_CLOUD_SYNC=false
-IS_IN_EVALS=true
+IS_IN_EVALS=true
+
+# cloud phone
+AGENTBOX_API_KEY=
+AGENTBOX_TEMPLATE=
+AGENTBOX_TIMEOUT=1800

examples/components/web/super_agent/README.md

@@ -216,7 +216,7 @@ pnpm install
 pnpm dev
 ```
 
-Visit http://localhost:3000/chat for graphical interaction.
+Visit http://localhost:3600/chat for graphical interaction.
 
 ## Agent Capabilities
 

examples/components/web/super_agent/compose-super-agent.yaml

@@ -5,7 +5,7 @@ services:
     ports:
       - "3600:3600"
     volumes:
-      - ./web:/app
+      - ../../../../web:/app
       - /app/node_modules
       - /app/.pnpm-store
     working_dir: /app
@@ -27,7 +27,7 @@ services:
   # Super Agent API service
   super-agent:
     build:
-      context: .
+      context: ../../../../
       dockerfile: examples/components/web/super_agent/Dockerfile
     ports:
       - "8000:8000"

examples/components/web/super_agent/pyproject.toml

@@ -31,6 +31,8 @@ dependencies = [
     "tenacity>=8.2.0",
     # Browser automation
     "browser-use==0.5.5",
+    "openai>=1.81.0,<1.82.0",
+    "adb-shell>=0.4.4",
 ]
 
 [tool.uv.sources]
@@ -50,4 +52,4 @@ packages = ["src/super_agent"]
 requires = [
     "hatchling",
 ]
-build-backend = "hatchling.build"
+build-backend = "hatchling.build"

examples/components/web/super_agent/src/super_agent/agent/crew.py

@@ -15,7 +15,10 @@
 from langgraph.checkpoint.base import BaseCheckpointSaver
 
 from super_agent.common.session_state import SessionState
-from super_agent.common.sandbox_config import create_sandbox_source_by_session_id
+from super_agent.common.sandbox_config import (
+    create_cloud_phone_sandbox_by_session_id,
+    create_sandbox_source_by_session_id,
+)
 from super_agent.config.config import SuperAgentConfig, default_config
 from langcrew_tools.utils.s3.factory import create_s3_client
 from langcrew_tools.hitl.langchain_tools import UserInputTool
@@ -30,6 +33,7 @@
 from langcrew_tools.image_gen import ImageGenerationTool
 from langcrew_tools.code_interpreter import CodeInterpreterTool
 from langcrew_tools.commands import RunCommandTool
+from super_agent.tool.cloud_phone_streaming_tool import CloudPhoneStreamingTool
 
 logger = logging.getLogger(__name__)
 
@@ -96,14 +100,19 @@ def get_llm_client(self) -> BaseChatModel:
         )
 
     def get_tools(self) -> list[BaseTool]:
-        """Create browser tools"""
         # Web and search tools
         tools = [
             BrowserStreamingTool(
                 vl_llm=self.get_browser_llm(),
                 async_s3_client=self.async_s3_client,
                 # sandbox_source=none_sandbox,
             ),
+            CloudPhoneStreamingTool(
+                base_model=self.get_llm_client(),
+                sandbox_source=create_cloud_phone_sandbox_by_session_id(
+                    self.session_id, checkpointer=self.checkpointer
+                ),
+            ),
             UserInputTool(),
             WebSearchTool(),
             WebFetchTool(),

examples/components/web/super_agent/src/super_agent/agent/enhanced_crew.py

@@ -186,10 +186,29 @@ async def _aprocess_output(self, output: Any) -> Any:
         # Execute function callbacks
         for callback_fn in callback_copy:
             try:
-                if inspect.iscoroutinefunction(callback_fn):
-                    prev_result = await callback_fn(prev_result)
-                else:
-                    prev_result = callback_fn(prev_result)
+                if prev_result:
+                    # Handle the case where prev_result might be a list
+                    if isinstance(prev_result, list):
+                        # Process each item in the list through the callback
+                        processed_items = []
+                        for item in prev_result:
+                            if inspect.iscoroutinefunction(callback_fn):
+                                processed_item = await callback_fn(item)
+                            else:
+                                processed_item = callback_fn(item)
+                            if processed_item:
+                                # If processed_item is also a list, extend instead of append
+                                if isinstance(processed_item, list):
+                                    processed_items.extend(processed_item)
+                                else:
+                                    processed_items.append(processed_item)
+                        prev_result = processed_items
+                    else:
+                        # Original logic for dict input
+                        if inspect.iscoroutinefunction(callback_fn):
+                            prev_result = await callback_fn(prev_result)
+                        else:
+                            prev_result = callback_fn(prev_result)
             except Exception as e:
                 logger.error(f"Error in output processing callback: {e}")
 

examples/components/web/super_agent/src/super_agent/common/sandbox_config.py

@@ -7,7 +7,10 @@
 
 from .checkpointer_state_manager import CheckpointerStateManager
 from langcrew_tools.utils.sandbox import sandbox_toolkit
-
+from agentbox import Sandbox
+from langcrew_tools.cloud_phone.base import enable_a11y
+from langcrew_tools.utils.env_config import env_config
+from langchain_core.callbacks.manager import dispatch_custom_event
 
 logger = logging.getLogger(__name__)
 
@@ -47,3 +50,64 @@ async def _get_async_sandbox() -> "AsyncSandbox":
         return sandbox
 
     return _get_async_sandbox
+
+
+def create_cloud_phone_sandbox_by_session_id(
+    session_id: str,
+    checkpointer: BaseCheckpointSaver | None = None,
+) -> Callable[[], Awaitable["str"]]:
+    async def _get_cloud_phone_async_sandbox() -> "str":
+        # For now, create a new sandbox (placeholder implementation)
+        sandbox_id = await CheckpointerStateManager(checkpointer).get_value(
+            session_id, "cloud_phone_sandbox_id"
+        )
+        if sandbox_id:
+            logger.info(
+                f"cloud_phone_sandbox_id session_id: {session_id} cloud_phone_sandbox_id: {sandbox_id}"
+            )
+            return sandbox_id
+        else:
+            logger.info(f"create cloud_phone_sandbox_id session_id: {session_id}")
+            config = env_config.get_dict("AGENTBOX_")
+            config = env_config.filter_valid_parameters(Sandbox, config)
+            sbx = Sandbox(**config)
+            sbx.adb_shell.connect()
+            await enable_a11y(sbx)
+            auth_info = sbx.get_instance_auth_info(config["timeout"])
+
+            logger.info(
+                f"cloud_phone_sandbox_id session_id: {session_id} auth_info: {auth_info}"
+            )
+
+            try:
+                dispatch_custom_event(
+                    "on_langcrew_agentbox_created",
+                    {
+                        "sandbox_id": sbx.sandbox_id,
+                        "session_id": session_id,
+                        "instance_no": auth_info.instance_no,
+                        **(
+                            {
+                                "access_key": auth_info.access_key,
+                                "access_secret_key": auth_info.access_secret_key,
+                                "expire_time": auth_info.expire_time,
+                                "user_id": auth_info.user_id,
+                            }
+                            if auth_info
+                            else {}
+                        ),
+                    },
+                    config={"configurable": {"thread_id": session_id}},
+                )
+            except Exception as e:
+                logger.warning(
+                    f"create cloud_phone_sandbox_id session_id: {session_id} error: {e}"
+                )
+
+            await CheckpointerStateManager(checkpointer).set_state(
+                session_id, {"cloud_phone_sandbox_id": sbx.sandbox_id}
+            )
+            # Safely call the async callback if provided
+            return sbx.sandbox_id
+
+    return _get_cloud_phone_async_sandbox