Skip to main content

openhands.server.listen

load_file_upload_config​

def load_file_upload_config() -> tuple[int, bool, list[str]]

Load file upload configuration from the config object.

This function retrieves the file upload settings from the global config object. It handles the following settings:

  • Maximum file size for uploads
  • Whether to restrict file types
  • List of allowed file extensions

It also performs sanity checks on the values to ensure they are valid and safe.

Returns:

  • tuple - A tuple containing:
    • max_file_size_mb (int): Maximum file size in MB. 0 means no limit.
    • restrict_file_types (bool): Whether file type restrictions are enabled.
    • allowed_extensions (set): Set of allowed file extensions.

is_extension_allowed​

def is_extension_allowed(filename)

Check if the file extension is allowed based on the current configuration.

This function supports wildcards and files without extensions. The check is case-insensitive for extensions.

Arguments:

  • filename str - The name of the file to check.

Returns:

  • bool - True if the file extension is allowed, False otherwise.

attach_session​

@app.middleware('http')
async def attach_session(request: Request, call_next)

Middleware to attach session information to the request.

This middleware checks for the Authorization header, validates the token, and attaches the corresponding session to the request state.

Arguments:

  • request Request - The incoming request object.
  • call_next Callable - The next middleware or route handler in the chain.

Returns:

  • Response - The response from the next middleware or route handler.

websocket_endpoint​

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket)

WebSocket endpoint for receiving events from the client (i.e., the browser). Once connected, the client can send various actions:

  • Initialize the agent: session management, and event streaming. websocket (WebSocket): The WebSocket connection object.
  • Start a new development task:
  • Send a message:
  • Write contents to a file:
  • Read the contents of a file:
  • Run a command:
  • Run an IPython command:
  • Open a web page:
  • Add a task to the root_task:
  • Update a task in the root_task:
  • Change the agent's state:
  • Finish the task: 
    {"action": "initialize", "args": {"LLM_MODEL": "ollama/llama3", "AGENT": "CodeActAgent", "LANGUAGE": "en", "LLM_API_KEY": "ollama"}}

Args:

```json
{"action": "start", "args": {"task": "write a bash script that prints hello"}}
{"action": "message", "args": {"content": "Hello, how are you?", "image_urls": ["base64_url1", "base64_url2"]}}
{"action": "write", "args": {"path": "./greetings.txt", "content": "Hello, OpenHands?"}}
{"action": "read", "args": {"path": "./greetings.txt"}}
{"action": "run", "args": {"command": "ls -l", "thought": "", "confirmation_state": "confirmed"}}
{"action": "run_ipython", "args": {"command": "print('Hello, IPython!')"}}
{"action": "browse", "args": {"url": "https://arxiv.org/html/2402.01030v2"}}
{"action": "add_task", "args": {"task": "Implement feature X"}}
{"action": "modify_task", "args": {"id": "0", "state": "in_progress", "thought": ""}}
{"action": "start", "args": {"task": "write a bash script that prints hello"}}
```0
```json
{"action": "start", "args": {"task": "write a bash script that prints hello"}}
```1

#### get\_litellm\_models

```python
@app.get('/api/options/models')
async def get_litellm_models() -> list[str]

Get all models supported by LiteLLM.

This function combines models from litellm and Bedrock, removing any error-prone Bedrock models.

To get the models:

curl http://localhost:3000/api/litellm-models

Returns:

  • list - A sorted list of unique model names.

get_agents​

@app.get('/api/options/agents')
async def get_agents()

Get all agents supported by LiteLLM.

To get the agents:

curl http://localhost:3000/api/agents

Returns:

  • list - A sorted list of agent names.

get_security_analyzers​

@app.get('/api/options/security-analyzers')
async def get_security_analyzers()

Get all supported security analyzers.

To get the security analyzers:

curl http://localhost:3000/api/security-analyzers

Returns:

  • list - A sorted list of security analyzer names.

list_files​

@app.get('/api/list-files')
async def list_files(request: Request, path: str | None = None)

List files in the specified path.

This function retrieves a list of files from the agent's runtime file store, excluding certain system and hidden files/directories.

To list files:

curl http://localhost:3000/api/list-files

Arguments:

  • request Request - The incoming request object.
  • path str, optional - The path to list files from. Defaults to None.

Returns:

  • list - A list of file names in the specified path.

Raises:

  • HTTPException - If there's an error listing the files.

select_file​

@app.get('/api/select-file')
async def select_file(file: str, request: Request)

Retrieve the content of a specified file.

To select a file:

curl http://localhost:3000/api/select-file?file=<file_path>

Arguments:

  • file str - The path of the file to be retrieved. Expect path to be absolute inside the runtime.
  • request Request - The incoming request object.

Returns:

  • dict - A dictionary containing the file content.

Raises:

  • HTTPException - If there's an error opening the file.

sanitize_filename​

def sanitize_filename(filename)

Sanitize the filename to prevent directory traversal

get_remote_runtime_config​

@app.get('/api/conversation')
async def get_remote_runtime_config(request: Request)

Retrieve the remote runtime configuration.

Currently, this is the runtime ID.

upload_file​

@app.post('/api/upload-files')
async def upload_file(request: Request, files: list[UploadFile])

Upload a list of files to the workspace.

To upload a files:

curl -X POST -F "file=@<file_path1>" -F "file=@<file_path2>" http://localhost:3000/api/upload-files

Arguments:

  • request Request - The incoming request object.
  • files list[UploadFile] - A list of files to be uploaded.

Returns:

  • dict - A message indicating the success of the upload operation.

Raises:

  • HTTPException - If there's an error saving the files.

submit_feedback​

@app.post('/api/submit-feedback')
async def submit_feedback(request: Request)

Submit user feedback.

This function stores the provided feedback data.

To submit feedback:

curl -X POST -d '{"email": "test@example.com"}' -H "Authorization:"

Arguments:

  • request Request - The incoming request object.
  • feedback FeedbackDataModel - The feedback data to be stored.

Returns:

  • dict - The stored feedback data.

Raises:

  • HTTPException - If there's an error submitting the feedback.

appconfig_defaults​

@app.get('/api/defaults')
async def appconfig_defaults()

Retrieve the default configuration settings.

To get the default configurations:

curl http://localhost:3000/api/defaults

Returns:

  • dict - The default configuration settings.

save_file​

@app.post('/api/save-file')
async def save_file(request: Request)

Save a file to the agent's runtime file store.

This endpoint allows saving a file when the agent is in a paused, finished, or awaiting user input state. It checks the agent's state before proceeding with the file save operation.

Arguments:

  • request Request - The incoming FastAPI request object.

Returns:

  • JSONResponse - A JSON response indicating the success of the operation.

Raises:

HTTPException:

  • 403 error if the agent is not in an allowed state for editing.
  • 400 error if the file path or content is missing.
  • 500 error if there's an unexpected error during the save operation.

security_api​

@app.route('/api/security/{path:path}',
           methods=['GET', 'POST', 'PUT', 'DELETE'])
async def security_api(request: Request)

Catch-all route for security analyzer API requests.

Each request is handled directly to the security analyzer.

Arguments:

  • request Request - The incoming FastAPI request object.

Returns:

  • Any - The response from the security analyzer.

Raises:

  • HTTPException - If the security analyzer is not initialized.

get_vscode_url​

@app.get('/api/vscode-url')
async def get_vscode_url(request: Request)

Get the VSCode URL.

This endpoint allows getting the VSCode URL.

Arguments:

  • request Request - The incoming FastAPI request object.

Returns:

  • JSONResponse - A JSON response indicating the success of the operation.