| --- |
| title: "Implementing the API" |
| description: "Learn how to implement the Browser Use API in Python" |
| icon: "code" |
| --- |
|
|
| This guide shows how to implement common API patterns using Python. We'll create a complete example that creates and monitors a browser automation task. |
| |
| ## Basic Implementation |
| |
| For all settings see [Run Task](cloud/api-v10/run-task). |
| |
| Here's a simple implementation using Python's `requests` library to stream the task steps: |
| |
| ```python |
| import json |
| import time |
| |
| import requests |
| |
| API_KEY = 'your_api_key_here' |
| BASE_URL = 'https: |
| HEADERS = {'Authorization': f'Bearer {API_KEY}'} |
|
|
|
|
| def create_task(instructions: str): |
| """Create a new browser automation task""" |
| response = requests.post(f'{BASE_URL}/run-task', headers=HEADERS, json={'task': instructions}) |
| return response.json()['id'] |
|
|
|
|
| def get_task_status(task_id: str): |
| """Get current task status""" |
| response = requests.get(f'{BASE_URL}/task/{task_id}/status', headers=HEADERS) |
| return response.json() |
|
|
|
|
| def get_task_details(task_id: str): |
| """Get full task details including output""" |
| response = requests.get(f'{BASE_URL}/task/{task_id}', headers=HEADERS) |
| return response.json() |
|
|
|
|
| def wait_for_completion(task_id: str, poll_interval: int = 2): |
| """Poll task status until completion""" |
| count = 0 |
| unique_steps = [] |
| while True: |
| details = get_task_details(task_id) |
| new_steps = details['steps'] |
| |
| if new_steps != unique_steps: |
| for step in new_steps: |
| if step not in unique_steps: |
| print(json.dumps(step, indent=4)) |
| unique_steps = new_steps |
| count += 1 |
| status = details['status'] |
|
|
| if status in ['finished', 'failed', 'stopped']: |
| return details |
| time.sleep(poll_interval) |
|
|
|
|
| def main(): |
| task_id = create_task('Open https://www.google.com and search for openai') |
| print(f'Task created with ID: {task_id}') |
| task_details = wait_for_completion(task_id) |
| print(f"Final output: {task_details['output']}") |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
| ``` |
|
|
| |
|
|
| Here's how to implement task control with pause/resume functionality: |
| |
| ```python |
| def control_task(): |
| # Create a new task |
| task_id = create_task("Go to google.com and search for Browser Use") |
| |
| # Wait for 5 seconds |
| time.sleep(5) |
| |
| # Pause the task |
| requests.put(f"{BASE_URL}/pause-task?task_id={task_id}", headers=HEADERS) |
| print("Task paused! Check the live preview.") |
| |
| # Wait for user input |
| input("Press Enter to resume...") |
| |
| # Resume the task |
| requests.put(f"{BASE_URL}/resume-task?task_id={task_id}", headers=HEADERS) |
| |
| # Wait for completion |
| result = wait_for_completion(task_id) |
| print(f"Task completed with output: {result['output']}") |
| ``` |
| |
| <Note> |
| Remember to handle your API key securely and implement proper error handling |
| in production code. |
| </Note> |
| |