Robocorp’s Open Source Gen2 RPA is better, faster, and is the most cost effective RPA platform in the world - built for developers and teams that want to work smarter, not harder.
Robocorp’s technology empowers developers to build any automations that their team can dream up. Our programmable approach leverages a powerful open-source ecosystem of libraries and keywords that provides you the flexibility to build automations that you can count on.
Our RPA framework automation stack is built on-top of Python. Now developers are able to quickly build, integrate, deploy, analyze and maintain complex automations with Robocorp.
In addition to Python's intrinsic advantages it also has a large community of developers and vast ecosystem of readily available libraries that make integrations with AI/ML, OCR and other systems a breeze. Simply put, when it comes to building automations, Python is the native language of automation.
*** Settings ***Library RPA.HTTPLibrary RPA.JSONLibrary RPA.Robocorp.WorkItems*** Variables ***${WORK_ITEM_NAME}= traffic_data
Process traffic data ${payload}= Get Work Item Payload ${traffic_data}= Set Variable ${payload}[${WORK_ITEM_NAME}] ${valid}= Validate traffic data ${traffic_data} IF ${valid} Post traffic data to sales system ${traffic_data} ELSE Handle invalid traffic data ${traffic_data} END
def _pop_item(self): # Get the next input work item from the cloud queue. url = url_join( "runs", self._process_run_id, "robotRuns", self._step_run_id, "reserve-next-work-item", ) logging.info("Reserving new input work item from: %s", url) response = self._process_requests.post(url) return response.json()["workItemId"]
0
developersimport copy import email import fnmatch import json import logging import os from abc import ABC, abstractmethod from enum import Enum from pathlib import Path from shutil import copy2 from threading import Event from typing import Callable, Type, Any, Optional, Union, Dict, List, Tuple import yaml from robot.api.deco import library, keyword from robot.libraries.BuiltIn import BuiltIn from RPA.Email.ImapSmtp import ImapSmtp from RPA.FileSystem import FileSystem from RPA.core.helpers import import_by_name, required_env from RPA.core.logger import deprecation from RPA.core.notebook import notebook_print from .utils import ( JSONType, Requests, is_json_equal, json_dumps, resolve_path, truncate, url_join, ) UNDEFINED = object() # Undefined default value class State(Enum): """Work item state. (set when released)""" DONE = "COMPLETED" FAILED = "FAILED" class Error(Enum): """Failed work item error type.""" BUSINESS = "BUSINESS" # wrong/missing data, shouldn't be retried APPLICATION = "APPLICATION" # logic issue/timeout, can be retried class EmptyQueue(IndexError): """Raised when trying to load an input item and none available.""" class BaseAdapter(ABC): """Abstract base class for work item adapters.""" @abstractmethod def reserve_input(self) -> str: """Get next work item ID from the input queue and reserve it.""" raise NotImplementedError @abstractmethod def release_input( self, item_id: str, state: State, exception: Optional[dict] = None ): """Release the lastly retrieved input work item and set state.""" raise NotImplementedError @abstractmethod def create_output(self, parent_id: str, payload: Optional[JSONType] = None) -> str: """Create new output for work item, and return created ID.""" raise NotImplementedError @abstractmethod def load_payload(self, item_id: str) -> JSONType: """Load JSON payload from work item.""" raise NotImplementedError @abstractmethod def save_payload(self, item_id: str, payload: JSONType): """Save JSON payload to work item.""" raise NotImplementedError @abstractmethod def list_files(self, item_id: str) -> List[str]: """List attached files in work item.""" raise NotImplementedError @abstractmethod def get_file(self, item_id: str, name: str) -> bytes: """Read file's contents from work item.""" raise NotImplementedError @abstractmethod def add_file(self, item_id: str, name: str, *, original_name: str, content: bytes): """Attach file to work item.""" raise NotImplementedError @abstractmethod def remove_file(self, item_id: str, name: str): """Remove attached file from work item.""" raise NotImplementedError class RobocorpAdapter(BaseAdapter): """Adapter for saving/loading work items from Robocorp Control Room. Required environment variables: * RC_API_WORKITEM_HOST: Work item API hostname * RC_API_WORKITEM_TOKEN: Work item API access token * RC_API_PROCESS_HOST: Process API hostname * RC_API_PROCESS_TOKEN: Process API access token * RC_WORKSPACE_ID: Control room workspace ID * RC_PROCESS_ID: Control room process ID * RC_PROCESS_RUN_ID: Control room process run ID * RC_ROBOT_RUN_ID: Control room robot run ID * RC_WORKITEM_ID: Control room work item ID (input) """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # IDs identifying the current robot run and its input. self._workspace_id = required_env("RC_WORKSPACE_ID") self._process_run_id = required_env("RC_PROCESS_RUN_ID") self._step_run_id = required_env("RC_ACTIVITY_RUN_ID") self._initial_item_id: Optional[str] = required_env("RC_WORKITEM_ID") self._workitem_requests = self._process_requests = None self._init_workitem_requests() self._init_process_requests() def _init_workitem_requests(self): # Endpoint for old work items API. workitem_host = required_env("RC_API_WORKITEM_HOST") workitem_token = required_env("RC_API_WORKITEM_TOKEN") route_prefix = ( url_join( workitem_host, "json-v1", "workspaces", self._workspace_id, "workitems" ) + "/" ) default_headers = { "Authorization": f"Bearer {workitem_token}", "Content-Type": "application/json", } logging.info("Work item API route prefix: %s", route_prefix) self._workitem_requests = Requests( route_prefix, default_headers=default_headers ) def _init_process_requests(self): # Endpoint for the new process API. process_host = required_env("RC_API_PROCESS_HOST") process_token = required_env("RC_API_PROCESS_TOKEN") process_id = required_env("RC_PROCESS_ID") route_prefix = ( url_join( process_host, "process-v1", "workspaces", self._workspace_id, "processes", process_id, ) + "/" ) default_headers = { "Authorization": f"Bearer {process_token}", "Content-Type": "application/json", } logging.info("Process API route prefix: %s", route_prefix) self._process_requests = Requests(route_prefix, default_headers=default_headers) def _pop_item(self): # Get the next input work item from the cloud queue. url = url_join( "runs", self._process_run_id, "robotRuns", self._step_run_id, "reserve-next-work-item", ) logging.info("Reserving new input work item from: %s", url) response = self._process_requests.post(url) return response.json()["workItemId"] def reserve_input(self) -> str: if self._initial_item_id: item_id = self._initial_item_id self._initial_item_id = None return item_id item_id = self._pop_item() if not item_id: raise EmptyQueue("No work items in the input queue") return item_id def release_input( self, item_id: str, state: State, exception: Optional[dict] = None ): # Release the current input work item in the cloud queue. url = url_join( "runs", self._process_run_id, "robotRuns", self._step_run_id, "release-work-item", ) body = {"workItemId": item_id, "state": state.value} if exception: for key, value in list(exception.items()): if value is None: del exception[key] body["exception"] = exception logging.info( "Releasing %s input work item %r into %r with exception: %s", state.value, item_id, url, exception, ) self._process_requests.post(url, json=body)
Monitor, manage and visualize your automations in
a single pane of glass.
Deploy, monitor, and manage your bots quickly from within Robocorps Control Room — a complete platform that serves as the central command for all of your automations.
Robocorp cloud is a convenient way to manage, distribute, and monitor your robot workforce in real time across your organization from one central cloud command center.
Experience full control of where bot code and data meet your infosec requirements.
Create scalable automations with one-to-many style of processing.
Automatically scale the execution of automations by spinning up containers in parallel processes for faster processing time.
This makes it easy for customers to quickly deploy robots at scale. Bots and or portions of bots can take advantage of cloud infrastructure but you retain the flexibility needed to address the uniqueness of your use case and systems.
No Need to Set Up Any Infrastructure
No Upfront Investments to Procure Hardware
No Maintenance Cost (Reduces OpEx)
Reduce TTV and Accelerate TTM
Software robots built with Robocorp easily interact with the frontend and backend of the programs and platforms you use every day. The result is faster throughput and significantly lower maintenance costs from Gen1 providers.
Worried about having no visibility into the layers below and your proprietary platforms? These robots take screenshots of elements and tell the automation how, when and where to interact with the interface.
When the UI is built from HTML, CSS, JavaScript, images and other assets, our robots can interact with the code to navigate the UI. Giving you a more reliable interaction than the GUI, this works in Windows applications and browsers.
APIs are used to transfer data to and from an application. Robocorp offers libraries that communicate with APIs to perform automations, typically through HTTP requests.
When the application allows, robots can access the persistent data storage. In some cases, Robocorp offers database libraries, and in others, there are APIs.
Quickly scale environment resources up and down based on demand without license overheads.
Blazing fast scale allows you to run parallel processes using multiple bots that perform many different tasks simultaneously.
Splitting large workloads into small items is key to accelerating processing time. With Robocorp, we break the traditional RPA provider model - no constraints or outdated licensing ever.
Robocorp makes it simple for business users to manage their automated processes and track value of their business operations.
Teams that run their automations on Robocorp save up to 80% of what they’ve traditionally spent running Gen1 tools.
No upfront costs to deploy or schedule robots.
Only pay when you run your robots.
Scalable pricing adjusts with your automation requirements.
Predict your costs with transparent pricing.
Lower maintenance, hosting and support costs.
Ensuring your security is at the forefront of what we do, because of this, Robocorp is proud to be SOC II Type 2 certified, HIPAA compliant and we partner with AWS which powers our cloud infrastructure.
We believe that everything worth doing is worth doing well. We follow operational best practices and are committed to keeping your information private and your data secure.
Robocorp is SOC 2 Type II compliant. Our certifications ensure that we are following best practices for security and data handling that were set in place by AICPA.
We are accessible to all. Having our HIPAA compliance allows us to do just that. So, if you’re in the Healthcare industry we have you covered.
The Robocorp community is where passion for automation meets the power of coming together to create something great. Built by developers, for developers, our dedicated community space is available 24/7 for you to ask and answer questions.
With robust documentation, a large keyword library and a portal filled with hundreds of bot templates built on Robocorp’ RPA framework, there are endless possibilities of what you can do.
Feeling inspired or have a great idea? Take that open-source code you just wrote and share it with our global community of developers.