Use generic types (#1680)

agenthub/README.md

@@ -16,14 +16,18 @@ Every agent also has a `self.llm` which it can use to interact with the LLM conf
 See the [LiteLLM docs for `self.llm.completion`](https://docs.litellm.ai/docs/completion).
 
 ## State
+
 The `state` contains:
-* A history of actions taken by the agent, as well as any observations (e.g. file content, command output) from those actions
-* A list of actions/observations that have happened since the most recent step
-* A [`plan`](https://github.com/OpenDevin/OpenDevin/blob/main/opendevin/plan.py), which contains the main goal
-  * The agent can add and modify subtasks through the `AddTaskAction` and `ModifyTaskAction`
+
+- A history of actions taken by the agent, as well as any observations (e.g. file content, command output) from those actions
+- A list of actions/observations that have happened since the most recent step
+- A [`plan`](https://github.com/OpenDevin/OpenDevin/blob/main/opendevin/plan.py), which contains the main goal
+  - The agent can add and modify subtasks through the `AddTaskAction` and `ModifyTaskAction`
 
 ## Actions
+
 Here is a list of available Actions, which can be returned by `agent.step()`:
+
 - [`CmdRunAction`](../opendevin/action/bash.py) - Runs a command inside a sandboxed terminal
 - [`CmdKillAction`](../opendevin/action/bash.py) - Kills a background command
 - [`IPythonRunCellAction`](../opendevin/action/bash.py) - Execute a block of Python code interactively (in Jupyter notebook) and receives `CmdOutputObservation`. Requires setting up `jupyter` [plugin](../opendevin/sandbox/plugins) as a requirement.
@@ -43,11 +47,13 @@ Here is a list of available Actions, which can be returned by `agent.step()`:
 You can use `action.to_dict()` and `action_from_dict` to serialize and deserialize actions.
 
 ## Observations
+
 There are also several types of Observations. These are typically available in the step following the corresponding Action.
 But they may also appear as a result of asynchronous events (e.g. a message from the user, logs from a command running
 in the background).
 
 Here is a list of available Observations:
+
 - [`CmdOutputObservation`](../opendevin/observation/run.py)
 - [`BrowserOutputObservation`](../opendevin/observation/browse.py)
 - [`FileReadObservation`](../opendevin/observation/files.py)
@@ -59,19 +65,24 @@ Here is a list of available Observations:
 You can use `observation.to_dict()` and `observation_from_dict` to serialize and deserialize observations.
 
 ## Interface
+
 Every agent must implement the following methods:
 
 ### `step`
+
 ```
 def step(self, state: "State") -> "Action"
 ```
+
 `step` moves the agent forward one step towards its goal. This probably means
 sending a prompt to the LLM, then parsing the response into an `Action`.
 
 ### `search_memory`
+
 ```
-def search_memory(self, query: str) -> List[str]:
+def search_memory(self, query: str) -> list[str]:
 ```
+
 `search_memory` should return a list of events that match the query. This will be used
 for the `recall` action.
 

agenthub/SWE_agent/agent.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from opendevin.controller.agent import Agent
 from opendevin.controller.state.state import State
 from opendevin.events.action import (
@@ -32,7 +30,7 @@ class SWEAgent(Agent):
         super().__init__(llm)
         self.memory_window = 4
         self.max_retries = 2
-        self.running_memory: List[str] = []
+        self.running_memory: list[str] = []
         self.cur_file: str = ''
         self.cur_line: int = 0
 
@@ -41,7 +39,7 @@ class SWEAgent(Agent):
         memory = MEMORY_FORMAT(action.to_memory(), observation.to_memory())
         self.running_memory.append(memory)
 
-    def _think_act(self, messages: List[dict]) -> tuple[Action, str]:
+    def _think_act(self, messages: list[dict]) -> tuple[Action, str]:
         resp = self.llm.completion(
             messages=messages,
             temperature=0.05,
@@ -99,7 +97,7 @@ class SWEAgent(Agent):
         self.latest_action = action
         return action
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         return [item for item in self.running_memory if query in item]
 
     def reset(self) -> None:

agenthub/delegator_agent/agent.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from opendevin.controller.agent import Agent
 from opendevin.controller.state.state import State
 from opendevin.events.action import Action, AgentDelegateAction, AgentFinishAction
@@ -81,5 +79,5 @@ class DelegatorAgent(Agent):
         else:
             raise Exception('Invalid delegate state')
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         return []

agenthub/dummy_agent/agent.py

@@ -1,5 +1,5 @@
 import time
-from typing import List, TypedDict
+from typing import TypedDict
 
 from opendevin.controller.agent import Agent
 from opendevin.controller.state.state import State
@@ -35,7 +35,7 @@ FIXME: There are a few problems this surfaced
 """
 
 ActionObs = TypedDict(
-    'ActionObs', {'action': Action, 'observations': List[Observation]}
+    'ActionObs', {'action': Action, 'observations': list[Observation]}
 )
 
 BACKGROUND_CMD = 'echo "This is in the background" && sleep .1 && echo "This too"'
@@ -49,7 +49,7 @@ class DummyAgent(Agent):
 
     def __init__(self, llm: LLM):
         super().__init__(llm)
-        self.steps: List[ActionObs] = [
+        self.steps: list[ActionObs] = [
             {
                 'action': AddTaskAction(parent='0', goal='check the current directory'),
                 'observations': [NullObservation('')],
@@ -160,5 +160,5 @@ class DummyAgent(Agent):
                     ), f'Expected observation {expected_obs}, got {hist_obs}'
         return self.steps[state.iteration]['action']
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         return ['I am a computer.']

agenthub/micro/agent.py

@@ -1,5 +1,4 @@
 import json
-from typing import Dict, List
 
 from jinja2 import BaseLoader, Environment
 
@@ -59,7 +58,7 @@ def to_json(obj, **kwargs):
 
 class MicroAgent(Agent):
     prompt = ''
-    agent_definition: Dict = {}
+    agent_definition: dict = {}
 
     def __init__(self, llm: LLM):
         super().__init__(llm)
@@ -83,5 +82,5 @@ class MicroAgent(Agent):
         action = parse_response(action_resp)
         return action
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         return []

agenthub/micro/instructions.py

@@ -1,7 +1,6 @@
 import os
-from typing import Dict
 
-instructions: Dict = {}
+instructions: dict = {}
 
 base_dir = os.path.dirname(os.path.abspath(__file__)) + '/_instructions'
 for root, dirs, files in os.walk(base_dir):

agenthub/monologue_agent/agent.py

@@ -1,5 +1,3 @@
-from typing import List
-
 import agenthub.monologue_agent.utils.prompts as prompts
 from agenthub.monologue_agent.utils.monologue import Monologue
 from opendevin.controller.agent import Agent
@@ -239,7 +237,7 @@ class MonologueAgent(Agent):
         self.latest_action = action
         return action
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         """
         Uses VectorIndexRetriever to find related memories within the long term memory.
         Uses search to produce top 10 results.
@@ -248,7 +246,7 @@ class MonologueAgent(Agent):
         - query (str): The query that we want to find related memories for
 
         Returns:
-        - List[str]: A list of top 10 text results that matched the query
+        - list[str]: A list of top 10 text results that matched the query
         """
         if self.memory is None:
             return []

agenthub/monologue_agent/utils/memory.py

@@ -171,7 +171,7 @@ class LongTermMemory:
         - k (int): Number of top results to return
 
         Returns:
-        - List[str]: List of top k results found in current memory
+        - list[str]: list of top k results found in current memory
         """
         retriever = VectorIndexRetriever(
             index=self.index,

agenthub/monologue_agent/utils/monologue.py

@@ -36,7 +36,7 @@ class Monologue:
         Get the current thoughts of the agent.
 
         Returns:
-        - List: The list of thoughts that the agent has.
+        - list: The list of thoughts that the agent has.
         """
         return self.thoughts
 

agenthub/monologue_agent/utils/prompts.py

@@ -1,6 +1,5 @@
 import re
 from json import JSONDecodeError
-from typing import List
 
 from opendevin.core.config import config
 from opendevin.core.exceptions import LLMOutputError
@@ -98,7 +97,7 @@ You can also use the same action and args from the source monologue.
 """
 
 
-def get_summarize_monologue_prompt(thoughts: List[dict]):
+def get_summarize_monologue_prompt(thoughts: list[dict]):
     """
     Gets the prompt for summarizing the monologue
 
@@ -112,16 +111,16 @@ def get_summarize_monologue_prompt(thoughts: List[dict]):
 
 def get_request_action_prompt(
     task: str,
-    thoughts: List[dict],
-    background_commands_obs: List[CmdOutputObservation] = [],
+    thoughts: list[dict],
+    background_commands_obs: list[CmdOutputObservation] = [],
 ):
     """
     Gets the action prompt formatted with appropriate values.
 
     Parameters:
     - task (str): The current task the agent is trying to accomplish
-    - thoughts (List[dict]): The agent's current thoughts
-    - background_commands_obs (List[CmdOutputObservation]): List of all observed background commands running
+    - thoughts (list[dict]): The agent's current thoughts
+    - background_commands_obs (list[CmdOutputObservation]): list of all observed background commands running
 
     Returns:
     - str: Formatted prompt string with hint, task, monologue, and background included
@@ -203,7 +202,7 @@ def parse_action_response(response: str) -> Action:
     return action_from_dict(action_dict)
 
 
-def parse_summary_response(response: str) -> List[dict]:
+def parse_summary_response(response: str) -> list[dict]:
     """
     Parses a summary of the monologue
 
@@ -211,7 +210,7 @@ def parse_summary_response(response: str) -> List[dict]:
     - response (str): The response string to be parsed
 
     Returns:
-    - List[dict]: The list of summaries output by the model
+    - list[dict]: The list of summaries output by the model
     """
     parsed = json.loads(response)
     return parsed['new_monologue']

agenthub/planner_agent/agent.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from opendevin.controller.agent import Agent
 from opendevin.controller.state.state import State
 from opendevin.events.action import Action, AgentFinishAction
@@ -46,5 +44,5 @@ class PlannerAgent(Agent):
         action = parse_response(action_resp)
         return action
 
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         return []

agenthub/planner_agent/prompt.py

@@ -1,5 +1,4 @@
 import json
-from typing import List, Tuple
 
 from opendevin.controller.state.plan import Plan
 from opendevin.core.logger import opendevin_logger as logger
@@ -124,14 +123,14 @@ def get_hint(latest_action_id: str) -> str:
     return hints.get(latest_action_id, '')
 
 
-def get_prompt(plan: Plan, history: List[Tuple[Action, Observation]]) -> str:
+def get_prompt(plan: Plan, history: list[tuple[Action, Observation]]) -> str:
     """
     Gets the prompt for the planner agent.
     Formatted with the most recent action-observation pairs, current task, and hint based on last action
 
     Parameters:
     - plan (Plan): The original plan outlined by the user with LLM defined tasks
-    - history (List[Tuple[Action, Observation]]): List of corresponding action-observation pairs
+    - history (list[tuple[Action, Observation]]): list of corresponding action-observation pairs
 
     Returns:
     - str: The formatted string prompt with historical values

docs/modules/usage/agents.md

@@ -8,12 +8,13 @@ sidebar_position: 3
 
 ### Description
 
-This agent implements the CodeAct idea ([paper](https://arxiv.org/abs/2402.13463), [tweet](https://twitter.com/xingyaow_/status/1754556835703751087)) that consolidates LLM agents’ **act**ions into a unified **code** action space for both *simplicity* and *performance* (see paper for more details).
+This agent implements the CodeAct idea ([paper](https://arxiv.org/abs/2402.13463), [tweet](https://twitter.com/xingyaow_/status/1754556835703751087)) that consolidates LLM agents’ **act**ions into a unified **code** action space for both _simplicity_ and _performance_ (see paper for more details).
 
 The conceptual idea is illustrated below. At each turn, the agent can:
 
 1. **Converse**: Communicate with humans in natural language to ask for clarification, confirmation, etc.
 2. **CodeAct**: Choose to perform the task by executing code
+
 - Execute any valid Linux `bash` command
 - Execute any valid `Python` code with [an interactive Python interpreter](https://ipython.org/). This is simulated through `bash` command, see plugin system below for more details.
 
@@ -22,6 +23,7 @@ The conceptual idea is illustrated below. At each turn, the agent can:
 ### Plugin System
 
 To make the CodeAct agent more powerful with only access to `bash` action space, CodeAct agent leverages OpenDevin's plugin system:
+
 - [Jupyter plugin](https://github.com/OpenDevin/OpenDevin/tree/main/opendevin/runtime/plugins/jupyter): for IPython execution via bash command
 - [SWE-agent tool plugin](https://github.com/OpenDevin/OpenDevin/tree/main/opendevin/runtime/plugins/swe_agent_commands): Powerful bash command line tools for software development tasks introduced by [swe-agent](https://github.com/princeton-nlp/swe-agent).
 
@@ -29,8 +31,7 @@ To make the CodeAct agent more powerful with only access to `bash` action space,
 
 https://github.com/OpenDevin/OpenDevin/assets/38853559/f592a192-e86c-4f48-ad31-d69282d5f6ac
 
-*Example of CodeActAgent with `gpt-4-turbo-2024-04-09` performing a data science task (linear regression)*
-
+_Example of CodeActAgent with `gpt-4-turbo-2024-04-09` performing a data science task (linear regression)_
 
 ### Actions
 
@@ -50,18 +51,17 @@ https://github.com/OpenDevin/OpenDevin/assets/38853559/f592a192-e86c-4f48-ad31-d
 
 ### Methods
 
-| Method          | Description                                                                                                                                                                                                                                             |
-| --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `__init__`      | Initializes an agent with `llm` and a list of messages `List[Mapping[str, str]]`                                                                                                                                                                        |
+| Method          | Description                                                                                                                                     |
+| --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- |
+| `__init__`      | Initializes an agent with `llm` and a list of messages `list[Mapping[str, str]]`                                                                |
 | `step`          | Performs one step using the CodeAct Agent. This includes gathering info on previous steps and prompting the model to make a command to execute. |
-| `search_memory` | Not yet implemented                                                                                                                                                                                                                                     |
+| `search_memory` | Not yet implemented                                                                                                                             |
 
 ### Work-in-progress & Next step
 
 [] Support web-browsing
 [] Complete the workflow for CodeAct agent to submit Github PRs
 
-
 ## Monologue Agent
 
 ### Description

opendevin/controller/agent.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Dict, List, Type
+from typing import TYPE_CHECKING, Type
 
 if TYPE_CHECKING:
     from opendevin.controller.state.state import State
@@ -20,8 +20,8 @@ class Agent(ABC):
     It tracks the execution status and maintains a history of interactions.
     """
 
-    _registry: Dict[str, Type['Agent']] = {}
-    sandbox_plugins: List[PluginRequirement] = []
+    _registry: dict[str, Type['Agent']] = {}
+    sandbox_plugins: list[PluginRequirement] = []
 
     def __init__(
         self,
@@ -49,7 +49,7 @@ class Agent(ABC):
         pass
 
     @abstractmethod
-    def search_memory(self, query: str) -> List[str]:
+    def search_memory(self, query: str) -> list[str]:
         """
         Searches the agent's memory for information relevant to the given query.
 

opendevin/controller/state/plan.py

@@ -1,5 +1,3 @@
-from typing import List
-
 from opendevin.core.exceptions import PlanInvalidStateError
 from opendevin.core.logger import opendevin_logger as logger
 
@@ -21,14 +19,14 @@ class Task:
     id: str
     goal: str
     parent: 'Task | None'
-    subtasks: List['Task']
+    subtasks: list['Task']
 
     def __init__(
         self,
         parent: 'Task | None',
         goal: str,
         state: str = OPEN_STATE,
-        subtasks: List = [],
+        subtasks: list = [],
     ):
         """Initializes a new instance of the Task class.
 
@@ -186,7 +184,7 @@ class Plan:
             task = task.subtasks[part]
         return task
 
-    def add_subtask(self, parent_id: str, goal: str, subtasks: List = []):
+    def add_subtask(self, parent_id: str, goal: str, subtasks: list = []):
         """Adds a subtask to a parent task.
 
         Args:

opendevin/controller/state/state.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass, field
-from typing import Dict, List, Tuple
 
 from opendevin.controller.state.plan import Plan
 from opendevin.events.action import (
@@ -17,8 +16,8 @@ class State:
     iteration: int = 0
     # number of characters we have sent to and received from LLM so far for current task
     num_of_chars: int = 0
-    background_commands_obs: List[CmdOutputObservation] = field(default_factory=list)
-    history: List[Tuple[Action, Observation]] = field(default_factory=list)
-    updated_info: List[Tuple[Action, Observation]] = field(default_factory=list)
-    inputs: Dict = field(default_factory=dict)
-    outputs: Dict = field(default_factory=dict)
+    background_commands_obs: list[CmdOutputObservation] = field(default_factory=list)
+    history: list[tuple[Action, Observation]] = field(default_factory=list)
+    updated_info: list[tuple[Action, Observation]] = field(default_factory=list)
+    inputs: dict = field(default_factory=dict)
+    outputs: dict = field(default_factory=dict)

opendevin/events/action/agent.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field
-from typing import ClassVar, Dict
+from typing import ClassVar
 
 from opendevin.core.schema import ActionType
 
@@ -43,7 +43,7 @@ class AgentSummarizeAction(Action):
 
 @dataclass
 class AgentFinishAction(Action):
-    outputs: Dict = field(default_factory=dict)
+    outputs: dict = field(default_factory=dict)
     thought: str = ''
     action: str = ActionType.FINISH
 
@@ -54,7 +54,7 @@ class AgentFinishAction(Action):
 
 @dataclass
 class AgentRejectAction(Action):
-    outputs: Dict = field(default_factory=dict)
+    outputs: dict = field(default_factory=dict)
     thought: str = ''
     action: str = ActionType.REJECT
 

opendevin/events/observation/recall.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass
-from typing import List
 
 from opendevin.core.schema import ObservationType
 
@@ -12,7 +11,7 @@ class AgentRecallObservation(Observation):
     This data class represents a list of memories recalled by the agent.
     """
 
-    memories: List[str]
+    memories: list[str]
     role: str = 'assistant'
     observation: str = ObservationType.RECALL
 

opendevin/events/stream.py

@@ -1,7 +1,7 @@
 import asyncio
 from datetime import datetime
 from enum import Enum
-from typing import Callable, Dict, List
+from typing import Callable
 
 from opendevin.core.logger import opendevin_logger as logger
 
@@ -21,8 +21,8 @@ class EventSource(str, Enum):
 
 
 class EventStream:
-    _subscribers: Dict[str, Callable] = {}
-    _events: List[Event] = []
+    _subscribers: dict[str, Callable] = {}
+    _events: list[Event] = []
     _lock = asyncio.Lock()
 
     def subscribe(self, id: EventStreamSubscriber, callback: Callable):

opendevin/runtime/docker/exec_box.py

@@ -7,7 +7,6 @@ import time
 import uuid
 from collections import namedtuple
 from glob import glob
-from typing import Dict, List, Tuple
 
 import docker
 
@@ -32,7 +31,7 @@ class DockerExecBox(Sandbox):
     docker_client: docker.DockerClient
 
     cur_background_id = 0
-    background_commands: Dict[int, Process] = {}
+    background_commands: dict[int, Process] = {}
 
     def __init__(
         self,
@@ -95,7 +94,7 @@ class DockerExecBox(Sandbox):
             if exit_code != 0:
                 raise Exception(f'Failed to setup devin user: {logs}')
 
-    def get_exec_cmd(self, cmd: str) -> List[str]:
+    def get_exec_cmd(self, cmd: str) -> list[str]:
         if self.run_as_devin:
             return ['su', 'devin', '-c', cmd]
         else:
@@ -107,7 +106,7 @@ class DockerExecBox(Sandbox):
         bg_cmd = self.background_commands[id]
         return bg_cmd.read_logs()
 
-    def execute(self, cmd: str) -> Tuple[int, str]:
+    def execute(self, cmd: str) -> tuple[int, str]:
         # TODO: each execute is not stateful! We need to keep track of the current working directory
         def run_command(container, command):
             return container.exec_run(

opendevin/runtime/docker/local_box.py

@@ -2,7 +2,6 @@ import atexit
 import os
 import subprocess
 import sys
-from typing import Dict, Tuple
 
 from opendevin.core.config import config
 from opendevin.core.logger import opendevin_logger as logger
@@ -29,12 +28,12 @@ class LocalBox(Sandbox):
     def __init__(self, timeout: int = 120):
         os.makedirs(config.workspace_base, exist_ok=True)
         self.timeout = timeout
-        self.background_commands: Dict[int, Process] = {}
+        self.background_commands: dict[int, Process] = {}
         self.cur_background_id = 0
         atexit.register(self.cleanup)
         super().__init__()
 
-    def execute(self, cmd: str) -> Tuple[int, str]:
+    def execute(self, cmd: str) -> tuple[int, str]:
         try:
             completed_process = subprocess.run(
                 cmd,

opendevin/runtime/docker/process.py

@@ -1,6 +1,5 @@
 import select
 import sys
-from typing import Tuple
 
 from opendevin.runtime.process import Process
 
@@ -33,7 +32,7 @@ class DockerProcess(Process):
     def command(self) -> str:
         return self._command
 
-    def parse_docker_exec_output(self, logs: bytes) -> Tuple[bytes, bytes]:
+    def parse_docker_exec_output(self, logs: bytes) -> tuple[bytes, bytes]:
         """
             When you execute a command using `exec` in a docker container, the output produced will be in bytes. this function parses the output of a Docker exec command.
 
@@ -61,7 +60,7 @@ class DockerProcess(Process):
             logs (bytes): The raw output logs of the command.
 
         Returns:
-            Tuple[bytes, bytes]: A tuple containing the parsed output and any remaining data.
+            tuple[bytes, bytes]: A tuple containing the parsed output and any remaining data.
         """
         res = b''
         tail = b''

opendevin/runtime/docker/ssh_box.py

@@ -7,7 +7,6 @@ import time
 import uuid
 from collections import namedtuple
 from glob import glob
-from typing import Dict, List, Tuple, Union
 
 import docker
 from pexpect import pxssh
@@ -41,7 +40,7 @@ class DockerSSHBox(Sandbox):
     _ssh_port: int
 
     cur_background_id = 0
-    background_commands: Dict[int, Process] = {}
+    background_commands: dict[int, Process] = {}
 
     def __init__(
         self,
@@ -210,7 +209,7 @@ class DockerSSHBox(Sandbox):
         self.ssh.sendline(f'cd {self.sandbox_workspace_dir}')
         self.ssh.prompt()
 
-    def get_exec_cmd(self, cmd: str) -> List[str]:
+    def get_exec_cmd(self, cmd: str) -> list[str]:
         if self.run_as_devin:
             return ['su', 'opendevin', '-c', cmd]
         else:
@@ -222,7 +221,7 @@ class DockerSSHBox(Sandbox):
         bg_cmd = self.background_commands[id]
         return bg_cmd.read_logs()
 
-    def execute(self, cmd: str) -> Tuple[int, str]:
+    def execute(self, cmd: str) -> tuple[int, str]:
         cmd = cmd.strip()
         # use self.ssh
         self.ssh.sendline(cmd)
@@ -411,7 +410,7 @@ class DockerSSHBox(Sandbox):
             raise ex
 
         try:
-            network_kwargs: Dict[str, Union[str, Dict[str, int]]] = {}
+            network_kwargs: dict[str, str | dict[str, int]] = {}
             if self.use_host_network:
                 network_kwargs['network_mode'] = 'host'
             else:
@@ -440,9 +439,11 @@ class DockerSSHBox(Sandbox):
                     mount_dir: {'bind': self.sandbox_workspace_dir, 'mode': 'rw'},
                     # mount cache directory to /home/opendevin/.cache for pip cache reuse
                     config.cache_dir: {
-                        'bind': '/home/opendevin/.cache'
-                        if self.run_as_devin
-                        else '/root/.cache',
+                        'bind': (
+                            '/home/opendevin/.cache'
+                            if self.run_as_devin
+                            else '/root/.cache'
+                        ),
                         'mode': 'rw',
                     },
                 },

opendevin/runtime/e2b/sandbox.py

@@ -1,7 +1,6 @@
 import os
 import tarfile
 from glob import glob
-from typing import Dict, Tuple
 
 from e2b import Sandbox as E2BSandbox
 from e2b.sandbox.exception import (
@@ -18,7 +17,7 @@ from opendevin.runtime.sandbox import Sandbox
 class E2BBox(Sandbox):
     closed = False
     cur_background_id = 0
-    background_commands: Dict[int, Process] = {}
+    background_commands: dict[int, Process] = {}
     _cwd: str = '/home/user'
 
     def __init__(
@@ -73,7 +72,7 @@ class E2BBox(Sandbox):
         assert isinstance(proc, E2BProcess)
         return '\n'.join([m.line for m in proc.output_messages])
 
-    def execute(self, cmd: str) -> Tuple[int, str]:
+    def execute(self, cmd: str) -> tuple[int, str]:
         process = self.sandbox.process.start(cmd, env_vars=self._env)
         try:
             process_output = process.wait(timeout=self.timeout)

opendevin/runtime/files.py

@@ -1,16 +1,16 @@
 from pathlib import Path
-from typing import Any, Dict, List
+from typing import Any
 
 
 class WorkspaceFile:
     name: str
-    children: List['WorkspaceFile']
+    children: list['WorkspaceFile']
 
-    def __init__(self, name: str, children: List['WorkspaceFile']):
+    def __init__(self, name: str, children: list['WorkspaceFile']):
         self.name = name
         self.children = children
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self) -> dict[str, Any]:
         """Converts the File object to a dictionary.
 
         Returns:

opendevin/runtime/plugins/mixin.py

@@ -1,5 +1,5 @@
 import os
-from typing import List, Protocol, Tuple
+from typing import Protocol
 
 from opendevin.core.logger import opendevin_logger as logger
 from opendevin.runtime.plugins.requirement import PluginRequirement
@@ -8,7 +8,7 @@ from opendevin.runtime.plugins.requirement import PluginRequirement
 class SandboxProtocol(Protocol):
     # https://stackoverflow.com/questions/51930339/how-do-i-correctly-add-type-hints-to-mixin-classes
 
-    def execute(self, cmd: str) -> Tuple[int, str]: ...
+    def execute(self, cmd: str) -> tuple[int, str]: ...
 
     def copy_to(self, host_src: str, sandbox_dest: str, recursive: bool = False): ...
 
@@ -16,7 +16,7 @@ class SandboxProtocol(Protocol):
 class PluginMixin:
     """Mixin for Sandbox to support plugins."""
 
-    def init_plugins(self: SandboxProtocol, requirements: List[PluginRequirement]):
+    def init_plugins(self: SandboxProtocol, requirements: list[PluginRequirement]):
         """Load a plugin into the sandbox."""
         for requirement in requirements:
             # copy over the files

opendevin/runtime/plugins/swe_agent_commands/__init__.py

@@ -1,6 +1,5 @@
 import os
 from dataclasses import dataclass, field
-from typing import List
 
 from opendevin.runtime.plugins.requirement import PluginRequirement
 from opendevin.runtime.plugins.swe_agent_commands.parse_commands import (
@@ -39,7 +38,7 @@ class SWEAgentCommandsRequirement(PluginRequirement):
     sandbox_dest: str = '/opendevin/plugins/swe_agent_commands'
     bash_script_path: str = 'setup_default.sh'
 
-    scripts_filepaths: List[str | None] = field(
+    scripts_filepaths: list[str | None] = field(
         default_factory=lambda: DEFAULT_SCRIPT_FILEPATHS
     )
     documentation: str = DEFAULT_DOCUMENTATION
@@ -66,7 +65,7 @@ class SWEAgentCursorCommandsRequirement(PluginRequirement):
     sandbox_dest: str = '/opendevin/plugins/swe_agent_commands'
     bash_script_path: str = 'setup_cursor_mode.sh'
 
-    scripts_filepaths: List[str | None] = field(
+    scripts_filepaths: list[str | None] = field(
         default_factory=lambda: CURSOR_SCRIPT_FILEPATHS
     )
     documentation: str = CURSOR_DOCUMENTATION

opendevin/runtime/runtime.py

@@ -1,5 +1,4 @@
 from abc import abstractmethod
-from typing import List
 
 from opendevin.core.config import config
 from opendevin.events.action import (
@@ -62,7 +61,7 @@ class Runtime:
         self.sandbox = create_sandbox(sid, config.sandbox_type)
         self.browser = BrowserEnv()
 
-    def init_sandbox_plugins(self, plugins: List[PluginRequirement]) -> None:
+    def init_sandbox_plugins(self, plugins: list[PluginRequirement]) -> None:
         self.sandbox.init_plugins(plugins)
 
     async def run_action(self, action: Action) -> Observation:
@@ -83,7 +82,7 @@ class Runtime:
         observation = await getattr(self, action_id)(action)
         return observation
 
-    def get_background_obs(self) -> List[CmdOutputObservation]:
+    def get_background_obs(self) -> list[CmdOutputObservation]:
         """
         Returns all observations that have accumulated in the runtime's background.
         Right now, this is just background commands, but could include e.g. asyncronous

opendevin/runtime/sandbox.py

@@ -1,14 +1,13 @@
 import os
 from abc import ABC, abstractmethod
-from typing import Dict, Tuple
 
 from opendevin.runtime.docker.process import Process
 from opendevin.runtime.plugins.mixin import PluginMixin
 
 
 class Sandbox(ABC, PluginMixin):
-    background_commands: Dict[int, Process] = {}
-    _env: Dict[str, str] = {}
+    background_commands: dict[int, Process] = {}
+    _env: dict[str, str] = {}
 
     def __init__(self, **kwargs):
         for key in os.environ:
@@ -20,7 +19,7 @@ class Sandbox(ABC, PluginMixin):
         self._env[key] = value
 
     @abstractmethod
-    def execute(self, cmd: str) -> Tuple[int, str]:
+    def execute(self, cmd: str) -> tuple[int, str]:
         pass
 
     @abstractmethod

opendevin/server/auth/auth.py

@@ -1,5 +1,4 @@
 import os
-from typing import Dict
 
 import jwt
 from jwt.exceptions import InvalidTokenError
@@ -36,7 +35,7 @@ def get_sid_from_token(token: str) -> str:
     return ''
 
 
-def sign_token(payload: Dict[str, object]) -> str:
+def sign_token(payload: dict[str, object]) -> str:
     """Signs a JWT token."""
     # payload = {
     #     "sid": sid,

opendevin/server/session/manager.py

@@ -1,7 +1,7 @@
 import atexit
 import json
 import os
-from typing import Callable, Dict
+from typing import Callable
 
 from fastapi import WebSocket
 
@@ -15,7 +15,7 @@ SESSION_CACHE_FILE = os.path.join(CACHE_DIR, 'sessions.json')
 
 
 class SessionManager:
-    _sessions: Dict[str, Session] = {}
+    _sessions: dict[str, Session] = {}
 
     def __init__(self):
         self._load_sessions()
@@ -38,7 +38,7 @@ class SessionManager:
         logger.info('Saving sessions...')
         self._save_sessions()
 
-    async def send(self, sid: str, data: Dict[str, object]) -> bool:
+    async def send(self, sid: str, data: dict[str, object]) -> bool:
         """Sends data to the client."""
         message_stack.add_message(sid, 'assistant', data)
         if sid not in self._sessions:

opendevin/server/session/msg_stack.py

@@ -2,7 +2,6 @@ import atexit
 import json
 import os
 import uuid
-from typing import Dict
 
 from opendevin.core.logger import opendevin_logger as logger
 from opendevin.core.schema.action import ActionType
@@ -14,9 +13,9 @@ MSG_CACHE_FILE = os.path.join(CACHE_DIR, 'messages.json')
 class Message:
     id: str = str(uuid.uuid4())
     role: str  # "user"| "assistant"
-    payload: Dict[str, object]
+    payload: dict[str, object]
 
-    def __init__(self, role: str, payload: Dict[str, object]):
+    def __init__(self, role: str, payload: dict[str, object]):
         self.role = role
         self.payload = payload
 
@@ -24,14 +23,14 @@ class Message:
         return {'id': self.id, 'role': self.role, 'payload': self.payload}
 
     @classmethod
-    def from_dict(cls, data: Dict):
+    def from_dict(cls, data: dict):
         m = cls(data['role'], data['payload'])
         m.id = data['id']
         return m
 
 
 class MessageStack:
-    _messages: Dict[str, list[Message]] = {}
+    _messages: dict[str, list[Message]] = {}
 
     def __init__(self):
         self._load_messages()
@@ -41,7 +40,7 @@ class MessageStack:
         logger.info('Saving messages...')
         self._save_messages()
 
-    def add_message(self, sid: str, role: str, message: Dict[str, object]):
+    def add_message(self, sid: str, role: str, message: dict[str, object]):
         if sid not in self._messages:
             self._messages[sid] = []
         self._messages[sid].append(Message(role, message))
@@ -51,7 +50,7 @@ class MessageStack:
             return
         del self._messages[sid]
 
-    def get_messages(self, sid: str) -> list[Dict[str, object]]:
+    def get_messages(self, sid: str) -> list[dict[str, object]]:
         if sid not in self._messages:
             return []
         return [msg.to_dict() for msg in self._messages[sid]]

opendevin/server/session/session.py

@@ -1,5 +1,5 @@
 import time
-from typing import Callable, Dict
+from typing import Callable
 
 from fastapi import WebSocket, WebSocketDisconnect
 
@@ -44,7 +44,7 @@ class Session:
                 self.is_alive = False
             logger.exception('Error in loop_recv: %s', e)
 
-    async def send(self, data: Dict[str, object]) -> bool:
+    async def send(self, data: dict[str, object]) -> bool:
         try:
             if self.websocket is None or not self.is_alive:
                 return False
@@ -68,7 +68,7 @@ class Session:
         self.is_alive = True
         self.last_active_ts = int(time.time())
 
-    def load_from_data(self, data: Dict) -> bool:
+    def load_from_data(self, data: dict) -> bool:
         self.last_active_ts = data.get('last_active_ts', 0)
         if self.last_active_ts < int(time.time()) - DEL_DELT_SEC:
             return False