03 — Key Abstractions & Interfaces

class Agent(ABC):
    _registry: dict[str, type['Agent']] = {}
    sandbox_plugins: list[PluginRequirement] = []
 
    def __init__(self, config: AgentConfig, llm_registry: LLMRegistry):
        self.llm = llm_registry.get_llm_from_agent_config('agent', config)
        self.config = config
        self.tools: list = []
        self.mcp_tools: dict[str, ChatCompletionToolParam] = {}
 
    @abstractmethod
    def step(self, state: 'State') -> 'Action':
        """Given current state, produce the next action."""
        pass
 
    def reset(self) -> None: ...
    def get_system_message(self) -> 'SystemMessageAction | None': ...
 
    @classmethod
    def register(cls, name: str, agent_cls: type['Agent']) -> None: ...
 
    @classmethod
    def get_cls(cls, name: str) -> type['Agent']: ...
 
    @classmethod
    def list_agents(cls) -> list[str]: ...

class Runtime(FileEditRuntimeMixin):
    """Abstract base class for agent runtime environments."""
 
    @abstractmethod
    async def connect(self) -> None: ...
 
    @abstractmethod
    def run(self, action: CmdRunAction) -> Observation: ...
 
    @abstractmethod
    def run_ipython(self, action: IPythonRunCellAction) -> Observation: ...
 
    @abstractmethod
    def read(self, action: FileReadAction) -> Observation: ...
 
    @abstractmethod
    def write(self, action: FileWriteAction) -> Observation: ...
 
    @abstractmethod
    def edit(self, action: FileEditAction) -> Observation: ...
 
    @abstractmethod
    def browse(self, action: BrowseURLAction) -> Observation: ...
 
    @abstractmethod
    def browse_interactive(self, action: BrowseInteractiveAction) -> Observation: ...
 
    @abstractmethod
    async def call_tool_mcp(self, action: MCPAction) -> Observation: ...
 
    @abstractmethod
    def copy_to(self, host_src: str, sandbox_dest: str, recursive: bool = False): ...
 
    @abstractmethod
    def list_files(self, path: str | None = None) -> list[str]: ...
 
    @abstractmethod
    def copy_from(self, path: str) -> Path: ...

@dataclass
class Event:
    INVALID_ID = -1
 
    @property
    def id(self) -> int: ...           # Auto-assigned by EventStream
 
    @property
    def timestamp(self) -> str | None: ...  # ISO datetime
 
    @property
    def source(self) -> EventSource | None: ...  # AGENT, USER, ENVIRONMENT
 
    @property
    def cause(self) -> int | None: ...  # ID of causing action
 
    @property
    def llm_metrics(self) -> Metrics | None: ...  # Cost/token data
 
    @property
    def tool_call_metadata(self) -> ToolCallMetadata | None: ...

class FileStore(ABC):
    @abstractmethod
    def write(self, path: str, contents: str | bytes) -> None: ...
 
    @abstractmethod
    def read(self, path: str) -> str: ...
 
    @abstractmethod
    def list(self, path: str) -> list[str]: ...
 
    @abstractmethod
    def delete(self, path: str) -> None: ...

class LLM(RetryMixin, DebugMixin):
    def __init__(self, config: LLMConfig, service_id: str,
                 metrics: Metrics | None = None,
                 retry_listener: Callable | None = None):
        self.config: LLMConfig = copy.deepcopy(config)
        self.metrics: Metrics = metrics or Metrics(model_name=config.model)
        self._completion = partial(litellm_completion, model=config.model, ...)

class LLMRegistry:
    def __init__(self, config: OpenHandsConfig, ...):
        self.service_to_llm: dict[str, LLM] = {}
        self.active_agent_llm: LLM = self.get_llm('agent', llm_config)