Sending commands via the OPS API

The interface for operating Open Cosmos satellites follows a command and response pattern. Commands are sent and can receive one or more responses. Both commands and responses are formatted in the API as JSON, with structures depending on the specific command.

Listing available commands

This section of the help centre contains documentation on some of the commands available. A full list of the commands that are available for a mission can be retrieved by sending a GET request to app.open-cosmos.com/api/ops/v0/commands/mission/<mission-id>. The response will contain an array of JSON Schema objects with the name of each command and the arguments that it takes.

Connecting to a terminal session

Commands are sent to a specific terminal session, which allows commands sent for a particular purpose to be grouped together. Terminal sessions are specific to a given mission, and each mission has multiple terminal sessions.

The terminal session API can be accessed at api/ops/v0/terminal/sessions and provides endpoints for fetching, listing and creating terminal sessions.

The following python code demonstrates how to find or create a terminal session with a given name, given a mission ID (see Home) and an authenticated HTTP session (see Authentication):

from dataclasses import dataclass

import requests
from pydantic import BaseModel, TypeAdapter

OPS_BASE_URL = "https://app.open-cosmos.com"
TERMINAL_SESSION_URL = f"{OPS_BASE_URL}/api/ops/v0/terminal"


class _CreateSessionRequest(BaseModel):
    """Request format for the create session endpoint."""

    name: str
    missionId: str


class _TerminalSession(BaseModel):
    """Response format for the list sessions endpoint."""

    id: int
    name: str
    open: bool


@dataclass
class TerminalSessionServiceClient:
    """Client for interacting with the terminal session service."""

    mission: int
    http_client: requests.Session
    terminal_session_url: str = TERMINAL_SESSION_URL

    def get_or_create(self, session_name: str) -> int:
        """Get or create a terminal session ID with the given name.

        This method lists all of the terminal sessions for the mission and
        looks for an existing one with the given name. If it finds one, it
        returns the ID for that session. If it does not find one, it creates
        a new session with the given name and returns the ID for that session.

        Args:
            session_name: The name of the session to get or create.

        Returns:
            Session ID for the terminal session with the given name.
        """
        sessions = self._list_sessions()
        for session in sessions:
            if session.name == session_name:
                return session.id

        session = self._create_session(session_name)
        return session.id

    def _list_sessions(self) -> list[_TerminalSession]:
        url = f"{self.terminal_session_url}/sessions"
        response = self.http_client.get(
            url,
            # Filter for sessions that are open and accessible to customers of the mission
            # open=True returns sessions that are open in the sense of availability, deleted sessions are soft deleted and considered closed.
            # userAccessible=True returns sessions that are accessible to the customer, to see all sessions use userAccessible=False
            params={"missionId": self.mission, "open": True, "userAccessible": True},
        )
        response.raise_for_status()
        return TypeAdapter(list[_TerminalSession]).validate_python(response.json()["data"])

    def _create_session(self, name: str) -> _TerminalSession:
        url = f"{self.terminal_session_url}/sessions"
        data = _CreateSessionRequest(name=name, missionId=str(self.mission))
        response = self.http_client.post(url, json=data.model_dump())
        response.raise_for_status()
        return _TerminalSession.model_validate(response.json()["data"])

HTTP command API

The simplest way to send commands and receive responses is via the terminal session HTTP API. This can be accessed at api/ops/v0/terminal/sessions/<session-id>/command, using a session ID obtained as in the previous section. The API accepts a POST request with the following JSON body structure:

{
    "sequenceId": <random-integer>,
    "payload": {
        "command": "<command-name>",
        "payload": <payload>
    }
}

The sequenceId field should be set to a random integer generated by the client. This will be included with all responses to allow commands and responses to be correlated. This is less relevant for the HTTP API but is essential for the websocket API where responses to multiple commands are sent on the same channel.

The command name and payload fields should be set to match one of the available commands for the mission, as returned from the list commands API or from the command documentation.

The following python code shows how, given a terminal session ID, a client can be created to send commands to the terminal session and receive the responses.

import random
from dataclasses import dataclass
from typing import Sequence

import requests
from pydantic import BaseModel, TypeAdapter

OPS_BASE_URL = "https://app.open-cosmos.com"
TERMINAL_SESSION_URL = f"{OPS_BASE_URL}/api/ops/v0/terminal"


class SendCommandPayload(BaseModel):
    """Payload format for the send command endpoint."""

    command: str
    payload: dict


class CommandResponse(BaseModel):
    """Format for command responses."""

    type: str
    data: dict
    sequenceId: int


class _SendCommandRequest(BaseModel):
    """Request format for the send command endpoint."""

    sequenceId: int
    payload: SendCommandPayload


@dataclass
class SessionClient:
    """Client for sending commands to a given terminal session."""

    mission_id: int
    session_id: int

    http_client: requests.Session
    terminal_session_url: str = TERMINAL_SESSION_URL

    def send_command(
        self,
        command: SendCommandPayload,
        timeout_s: float | None = None,
    ) -> Sequence[dict]:
        """Send a command to the terminal session.

        Args:
            command: The command to send.
            timeout_s: The timeout for the command.

        Returns:
            The response from the terminal session.
        """
        url = f"{self.terminal_session_url}/sessions/{self.session_id}/command"
        body = _SendCommandRequest(sequenceId=random.randint(0, 1000000), payload=command)
        response = self.http_client.post(url, json=body.model_dump(), timeout=timeout_s)
        response.raise_for_status()
        return TypeAdapter(list[CommandResponse]).validate_python(response.json()["data"]["responses"])

The structure of the data field in the responses depends on the type and will be different for different types of command.

Websocket API

For more complex use cases where there is a need to have multiple commands executing asynchronously at the same time and to receive+ responses in real-time, there is also a websocket interface to the terminal session service.

The websocket interface can be accessed at api/ops/v0/terminal/sessions/ws/<session-id>. Once connected, commands can be sent by sending the following JSON structure to the websocket:

{
    "action": "SEND_UNARMMED_COMMAND",
    "sequenceId": <random-integer>,
    "payload": {
        "command": "<command-name>",
        "payload": <payload>
    }
}

This is the same as the one used above for the HTTP API, with the addition of the action field. It's important to save the sequenceId so that responses can be matched to the command they are responding to.

All messages that are read from the websocket have at least the following fields:

{
    "data": {...},
    "sequenceId": integer,
    "type": "string"
}

Once a command has been sent, at least three messages will be returned on the websocket with matching sequenceIds. The first, with type commandAcknowledgement, acknowledges that the platform has received the command and (if applicable) has sent it to the satellite. Following this, one or more responses to the command will be returned. The value of type and the structure of data will differ for different commands here. Finally, once all responses have been delivered, a message with type commandCompleted will be delivered to inform the client that no more messages should be expected for this command.