Add frontend agent (#442)

Co-authored-by: sohamganatra <[email protected]>
Co-authored-by: Prathit-tech <[email protected]>

python/composio/tools/env/filemanager/manager.py

@@ -155,6 +155,12 @@ def create(self, path: t.Union[str, Path]) -> File:
         self._recent = self._files[path]
         return self._recent
 
+    def create_directory(self, path: t.Union[str, Path]) -> Path:
+        """Create a new directory."""
+        path = self.working_dir / path
+        path.mkdir(parents=True, exist_ok=False)
+        return path
+
     def grep(
         self,
         word: str,

python/composio/tools/local/filetool/actions/create.py

@@ -9,73 +9,82 @@
 
 
 class CreateFileRequest(BaseFileRequest):
-    """Request to create a file."""
+    """Request to create a file or directory."""
 
-    file_path: str = Field(
+    path: str = Field(
         ...,
-        description="""File path to create in the editor.
-        If file already exists, it will be overwritten""",
+        description="""Path to create in the editor.
+        If file/directory already exists, it will be overwritten""",
+    )
+    is_directory: bool = Field(
+        False,
+        description="Whether to create a directory instead of a file",
     )
 
-    @field_validator("file_path")
+    @field_validator("path")
     @classmethod
-    def validate_file_path(cls, v: str) -> str:
+    def validate_path(cls, v: str) -> str:
         if v.strip() == "":
-            raise ValueError("File name cannot be empty or just whitespace")
+            raise ValueError("Path cannot be empty or just whitespace")
         if v in (".", ".."):
-            raise ValueError('File name cannot be "." or ".."')
+            raise ValueError('Path cannot be "." or ".."')
         return v
 
 
 class CreateFileResponse(BaseFileResponse):
-    """Response to create a file."""
+    """Response to create a file or directory."""
 
-    file: str = Field(
+    path: str = Field(
         default=None,
-        description="Path the created file.",
+        description="Path of the created file or directory.",
     )
     success: bool = Field(
         default=False,
-        description="Whether the file was created successfully",
+        description="Whether the file or directory was created successfully",
     )
 
 
 class CreateFile(BaseFileAction):
     """
-    Creates a new file within a shell session.
+    Creates a new file or directory within a shell session.
     Example:
-        - To create a file, provide the path of the new file. If the path you provide
+        - To create a file or directory, provide the path of the new file/directory. If the path you provide
         is relative, it will be created relative to the current working directory.
-        - The response will indicate whether the file was created successfully and list any errors.
+        - Specify is_directory=True to create a directory instead of a file.
+        - The response will indicate whether the file/directory was created successfully and list any errors.
     Raises:
-        - ValueError: If the file path is not a string or if the file path is empty.
-        - FileExistsError: If the file already exists.
-        - PermissionError: If the user does not have permission to create the file.
-        - FileNotFoundError: If the directory does not exist.
+        - ValueError: If the path is not a string or if the path is empty.
+        - FileExistsError: If the file/directory already exists.
+        - PermissionError: If the user does not have permission to create the file/directory.
+        - FileNotFoundError: If the parent directory does not exist.
         - OSError: If an OS-specific error occurs.
     """
 
-    _display_name = "Create a new file"
+    _display_name = "Create a new file or directory"
     _request_schema = CreateFileRequest
     _response_schema = CreateFileResponse
 
     def execute_on_file_manager(
         self, file_manager: FileManager, request_data: CreateFileRequest  # type: ignore
     ) -> CreateFileResponse:
         try:
+            if request_data.is_directory:
+                created_path = file_manager.create_directory(path=request_data.path)
+            else:
+                created_path = file_manager.create(path=request_data.path).path
             return CreateFileResponse(
-                file=str(
-                    file_manager.create(
-                        path=request_data.file_path,
-                    ).path
-                ),
+                path=str(created_path),
                 success=True,
             )
         except FileExistsError as e:
-            return CreateFileResponse(error=f"File already exists: {str(e)}")
+            return CreateFileResponse(
+                error=f"File or directory already exists: {str(e)}"
+            )
         except PermissionError as e:
             return CreateFileResponse(error=f"Permission denied: {str(e)}")
         except FileNotFoundError as e:
-            return CreateFileResponse(error=f"Directory does not exist: {str(e)}")
+            return CreateFileResponse(
+                error=f"Parent directory does not exist: {str(e)}"
+            )
         except OSError as e:
             return CreateFileResponse(error=f"OS error occurred: {str(e)}")

python/composio/tools/local/filetool/actions/edit.py

@@ -65,13 +65,17 @@ class EditFile(BaseFileAction):
     If a lint error occurs, the edit will not be applied.
     Review the error message, adjust your edit accordingly.
 
-    Examples A -
+    If start line and end line are the same,
+    the new text will be added at the start line &
+    text at end line will be still in the new edited file.
+
+    Examples A - Start line == End line
     Start line: 1
     End line: 1
     Text: "print(x)"
     Result: As Start line == End line, print(x) will be added as first line in the file. Rest of the file will be unchanged.
 
-    Examples B -
+    Examples B - Start line != End line
     Start line: 1
     End line: 3
     Text: "print(x)"
@@ -99,6 +103,7 @@ def execute_on_file_manager(
                     path=request_data.file_path,
                 )
             )
+
             if file is None:
                 raise FileNotFoundError(f"File not found: {request_data.file_path}")
 

python/composio/tools/local/shelltool/shell_exec/actions/exec.py

@@ -119,12 +119,16 @@ def execute(
         output = shell.exec(cmd=request_data.cmd)
         # run pwd
         output_dir = shell.exec(cmd="pwd")
-        pwd = output_dir[STDOUT]
+        current_shell_pwd = f"Currently in {output_dir[STDOUT]}"
+        if output[STDERR] != "":
+            current_shell_pwd = (
+                f"Failed to get current directory, error: {output_dir[STDERR]}"
+            )
         return ShellExecResponse(
             stdout=output[STDOUT],
             stderr=output[STDERR],
             exit_code=int(output[EXIT_CODE]),
-            current_shell_pwd=f"Currently in {pwd}",
+            current_shell_pwd=current_shell_pwd,
         )
 
 

python/dockerfiles/Dockerfile.dev

@@ -46,5 +46,7 @@ RUN useradd -rm -d /home/user -s /bin/bash -g root -G sudo -u 1000 user
 # Define entry point
 COPY entrypoint.sh /root/entrypoint.sh
 
+COPY ./IMG_20171230_084709_Bokeh.jpeg /root/karan_image.jpeg
+
 # Run entrypoint.sh
 ENTRYPOINT [ "/root/entrypoint.sh" ]

python/plugins/langchain/composio_langchain/toolset.py

@@ -123,6 +123,12 @@ def _wrap_tool(
         description = schema["description"]
         schema_params = schema["parameters"]
 
+        if len(description) > 1024:
+            self.logger.debug(
+                f"Description for {action} is > 1024 characters. Truncating it."
+            )
+            description = description[:1024]
+
         action_func = self._wrap_action(
             action=action,
             description=description,

python/setup.py

@@ -54,6 +54,7 @@ def scan_for_package_data(
     "e2b-code-interpreter",  # E2B workspace
     "gql",  # FlyIO workspace
     "requests_toolbelt",  # FlyIO workspace
+    "uvicorn",
 ]
 
 tools_require = [

python/swe/examples/langgraph_agent/__init__.py

@@ -0,0 +1 @@
+"""Composio-LangGraph SWE Agent"""

python/swe/examples/langgraph_agent/agent.py

@@ -0,0 +1,194 @@
+"""CrewAI SWE Agent"""
+
+import operator
+import os
+from typing import Annotated, Literal, Sequence, TypedDict
+
+import dotenv
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_openai import ChatOpenAI
+from langgraph.graph import END, START, StateGraph
+from langgraph.prebuilt import ToolNode
+from prompts import frontend_engineer_prompt, pm_prompt
+
+from composio_langgraph import Action, App, ComposioToolSet, WorkspaceType
+
+
+# Load environment variables from .env
+dotenv.load_dotenv()
+
+# Initialize tool.
+openai_client = ChatOpenAI(
+    api_key=os.environ["OPENAI_API_KEY"],  # type: ignore
+    model="gpt-4-turbo",
+)
+composio_toolset = ComposioToolSet(
+    workspace_config=WorkspaceType.Docker(
+        image="composio/composio:dev", persistent=True
+    )
+)
+
+# Get required tools
+coder_tools = [
+    *composio_toolset.get_actions(
+        actions=[
+            Action.FILETOOL_CHANGE_WORKING_DIRECTORY,
+            Action.FILETOOL_FIND_FILE,
+            Action.FILETOOL_CREATE_FILE,
+            Action.FILETOOL_EDIT_FILE,
+            Action.FILETOOL_OPEN_FILE,
+            Action.FILETOOL_SCROLL,
+            Action.FILETOOL_WRITE,
+            Action.FILETOOL_LIST_FILES,
+        ]
+    ),
+    *composio_toolset.get_tools(
+        apps=[
+            App.SHELLTOOL,
+            App.BROWSERTOOL,
+        ]
+    ),
+]
+
+coder_tool_node = ToolNode(coder_tools)
+
+pm_tools = composio_toolset.get_tools(
+    apps=[
+        App.BROWSERTOOL,
+        App.IMAGEANALYSERTOOL,
+    ]
+)
+
+pm_tool_node = ToolNode(pm_tools)
+
+
+# Define AgentState
+class AgentState(TypedDict):
+    messages: Annotated[Sequence[BaseMessage], operator.add]
+    sender: str
+
+
+# Agent names
+coding_agent_name = "Coder"
+coder_tool_node_name = "coder_tool"
+pm_agent_name = "PM"
+pm_tool_node_name = "pm_tool"
+
+
+# Helper function for agent nodes
+def create_agent_node(agent, name):
+    def agent_node(state):
+        result = agent.invoke(state)
+        if not isinstance(result, ToolMessage):
+            result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)
+        return {"messages": [result], "sender": name}
+
+    return agent_node
+
+
+# Create agents
+def create_agent(system_prompt, tools):
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", system_prompt),
+            MessagesPlaceholder(variable_name="messages"),
+        ]
+    )
+    llm = ChatOpenAI(temperature=0, streaming=True, model="gpt-4-1106-preview")
+    return prompt | llm.bind_tools(tools)
+
+
+coding_agent = create_agent(frontend_engineer_prompt, coder_tools)
+coding_node = create_agent_node(coding_agent, coding_agent_name)
+
+pm_agent = create_agent(pm_prompt, pm_tools)
+pm_node = create_agent_node(pm_agent, pm_agent_name)
+
+
+# Router function
+def router(
+    state,
+) -> Literal["call_tool", "pm", "__end__", "continue",]:
+    last_message = state["messages"][-1]
+    sender = state["sender"]
+    if last_message.tool_calls:
+        return "call_tool"
+    if (
+        "LANDING PAGE READY FOR REVIEW" in last_message.content
+        and sender == coding_agent_name
+    ):
+        return "pm"
+    if "LANDING PAGE LOOKS GOOD" in last_message.content and sender == pm_agent_name:
+        return "__end__"
+    return "continue"
+
+
+# Create workflow
+workflow = StateGraph(AgentState)
+
+# add agents
+workflow.add_node(coding_agent_name, coding_node)
+workflow.add_node(coder_tool_node_name, coder_tool_node)
+workflow.add_node(pm_agent_name, pm_node)
+workflow.add_node(pm_tool_node_name, pm_tool_node)
+
+# add start and end
+workflow.add_edge(START, coding_agent_name)
+
+# add conditional edges for tool calling
+workflow.add_conditional_edges(
+    coder_tool_node_name,
+    lambda x: x["sender"],
+    {coding_agent_name: coding_agent_name},
+)
+
+workflow.add_conditional_edges(
+    pm_tool_node_name, lambda x: x["sender"], {pm_agent_name: pm_agent_name}
+)
+
+workflow.add_conditional_edges(
+    coding_agent_name,
+    router,
+    {
+        "continue": coding_agent_name,
+        "call_tool": coder_tool_node_name,
+        "pm": pm_agent_name,
+    },
+)
+
+workflow.add_conditional_edges(
+    pm_agent_name,
+    router,
+    {
+        "continue": coding_agent_name,
+        "call_tool": pm_tool_node_name,
+        "__end__": END,
+    },
+)
+
+graph = workflow.compile()
+
+if __name__ == "__main__":
+    try:
+        final_state = graph.invoke(
+            {
+                "messages": [
+                    HumanMessage(
+                        content="""Create a personal website for Karan Vaidya, co-founder of Composio.
+He graduated from IIT Bombay in 2017 with a B.Tech in Computer Science and Engineering.
+He started his career in Rubrik as a SWE and later became founding PM at Nirvana Insurance.
+At Composio, he is leading Tech and Product teams and is responsible for building products
+around AI Agents.
+Make the site with multiple pages, include a blog, a contact page, and a home page.
+Make the website as classy as possible, use a minimalist approach, think through the design before you start coding.
+Image of Karan: /root/karan_image.jpeg"""
+                    )
+                ]
+            },
+            {"recursion_limit": 75},
+        )
+
+        print(final_state["messages"][-1].content)
+    except Exception as e:
+        print("Error:", e)