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?", "images_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": "", "is_confirmed": "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
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, feedback: FeedbackDataModel)
Submit user feedback.
This function stores the provided feedback data.
To submit feedback:
curl -X POST -F "email=test@example.com" -F "token=abc" -F "feedback=positive" -F "permissions=private" -F "trajectory={}" http://localhost:3000/api/submit-feedback