文章目录
- 1. GetToken
- 2. GetFolders
- 3. GetReleases
- 4. StartJobs
- End. 一站式启动机器人代码
注意: 本文内容皆为Python环境演示
your_domain
:你的UiPath Orchestrator 地址【不懂得去登录Orchestrator来回登录各个界面,始终不变的那一段就是 -_- 】
1. GetToken
此接口用于获取
access_token
请求类型 (Request Method):
POST
请求地址 (Request Url):
{your_domain}/identity/connect/token
请求正文 (Request Body):
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
grant_type | str | client_credentials | True | 填写默认就行 |
client_id | str | True | API ID | |
client_secret | str | True | API SECRET | |
scope | str | True | 作用域,根据后续调用API需要的权限进行配置,多个值之间空格进行分割,文章皆为会进行简单介绍 |
示例如下:
def get_access_token():url = f"{your_domain}/identity/connect/token"req_data = {"grant_type": "client_credentials","client_id": client_id,"client_secret": client_secret,"scope": "OR.Jobs OR.Settings",}result = requests.post(url, data=req_data)print(f"access_token 获取状态: {result.status_code}")print(f"access_token: {json.loads(result.text)['access_token']}")
响应正文:
- code 200: 成功
{"access_token": "", // token内容"expires_in": 3600,"token_type": "Bearer", // token头"scope": "" // 你刚开始授权的作用域
}
2. GetFolders
未通过此接口获取文件夹key时, 执行其他api操作通常会产生错误:
此操作需要一个组织单位
请求类型 (Request Method):
GET
请求地址 (Request Url):
{your_domain}/odata/Folders
请求头 (Request Headers):
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
Authorization | str | Bearer {access_token} | True | access_token 来自/identity/connect/token |
Accept | str | application/json | True |
示例如下:
def get_folders(access_token):url = f"{your_domain}/odata/Folders"headers = {"Authorization": f"Bearer {access_token}","Accept": "application/json",}result = requests.get(url, headers=headers)print(f"响应状态:{result.status_code}")print(f"响应内容:{result.text}")
响应正文:
- code 200: 成功
响应内容:
{"@odata.context": "https://xxxxx/odata/$metadata#Folders","@odata.count": 21, // 返回value数量"value": [{"Key": "xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx", // 文件夹key, 调用其他接口所需"DisplayName": "文件夹名称", // 文件夹展示名称"FullyQualifiedName": "文件夹名称", // 文件夹全称"FullyQualifiedNameOrderable": "文件夹名称","Description": null,"FolderType": "Standard","ProvisionType": "Automatic","PermissionModel": "FineGrained","ParentId": null,"ParentKey": null,"IsActive": true,"FeedType": "Processes","Id": 4,}]}
3. GetReleases
此接口用于获取启动作业所需的
release_key
请求类型 (Request Method):
GET
请求地址 (Request Url):
{your_domain}/odata/Releases
请求头 (Request Headers):
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
Authorization | str | Bearer {access_token} | True | access_token 来自/identity/connect/token |
Accept | str | application/json | True | |
X-UIPATH-FolderKey | str | True | 操作文件夹对应的key, 通过/odata/Folders 获取 |
示例如下:
"""
access_token: token
folder_key: 操作文件夹对应的key, 来自/odata/Folders
"""
def get_realeases(access_token, folder_key):url = f"{your_domain}/odata/Releases"headers = {"Authorization": f"Bearer {access_token}","accept": "application/json","X-UIPATH-FolderKey": folder_key,}result = requests.get(url, headers=headers)print(f"响应状态: {result.status_code}")print(f"响应正文: {result.text}")
响应正文:
- code 200: 成功
响应内容:
{"@odata.context": "https://xxxxxxxx/odata/$metadata#Releases","@odata.count": 2,"value": [{"Key": "xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx", // 流程key, 用于启动作业"ProcessKey": "包来源名称-测试", // 包来源名称"ProcessVersion": "1.0.6", // 包版本"IsLatestVersion": false,"IsProcessDeleted": false,"Description": "描述", // 流程描述"Name": "流程名称", // 流程名称"EnvironmentId": null,"EnvironmentName": "","EntryPointId": 2,"EntryPointPath": null,"InputArguments": null,"ProcessType": "Process","SupportsMultipleEntryPoints": true,"RequiresUserInteraction": true,"IsAttended": false,"IsCompiled": false,"AutomationHubIdeaUrl": null,"AutoUpdate": false,"HiddenForAttendedUser": false,"FeedId": "xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx","JobPriority": "Normal","SpecificPriorityValue": 45,"OrganizationUnitId": 21,"OrganizationUnitFullyQualifiedName": "文件夹名称","TargetFramework": "Legacy","RobotSize": null,"AutoCreateConnectedTriggers": true,"RemoteControlAccess": "None","LastModificationTime": "时间","LastModifierUserId": 9,"CreationTime": "时间","CreatorUserId": 9,"Id": 2,}]
}
4. StartJobs
此接口用于
启动作业
请求类型 (Request Method):
POST
请求地址 (Request Url):
{your_domain}/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs
请求头 (Request Headers):
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
Authorization | str | Bearer {access_token} | True | access_token 来自/identity/connect/token |
Accept | str | application/json | True | |
Content-Type | str | application/json | True | |
X-UIPATH-FolderKey | str | True | 操作文件夹对应的key, 通过/odata/Folders 获取 |
请求正文 (Request Body):
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
startInfo | dict | startInfo | True | 千万注意层级哦 |
- startInfo
参数名称 | 类型 | 默认值 | 不为空 | 描述 |
---|---|---|---|---|
ReleaseKey | str | True | release_key 来自/odata/Releases | |
JobsCount | num | 1 | True | 作业数 |
JobPriority | str | Normal | True | 优先级 |
Strategy | str | ModernJobsCount | True | 策略 |
ResumeOnSameContext | boolean | False | True | |
RuntimeType | str | Unattended | True |
示例如下:
"""
access_token: token
release_key: 流程key
folder_key: 文件夹key
"""
def start_jobs(access_token, release_key, folder_key):url = f"{your_domain}/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs"req_data = {"startInfo": {"ReleaseKey": release_key, # 流程id"JobsCount": 1, # 作业数"JobPriority": "Normal", # 优先级"Strategy": "ModernJobsCount","ResumeOnSameContext": False,"RuntimeType": "Unattended",}}headers = {"Accept": "application/json","Content-Type": "application/json","Authorization": f"Bearer {access_token}","X-UIPATH-FolderKey": folder_key,}result = requests.post(url=url, data=json.dumps(req_data), headers=headers)print(f"响应状态: {result.status_code}")print(f"响应正文: {result.text}")
响应正文:
- code 200: 成功
{"@odata.context": "https://xxxxxx/odata/$metadata#Jobs","value": [{"Key": "xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx","StartTime": null,"EndTime": null,"State": "Pending","JobPriority": "Normal","SpecificPriorityValue": 45,"ResourceOverwrites": null,"Source": "Manual","SourceType": "Manual","BatchExecutionKey": "xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx","Info": null,"CreationTime": "时间","StartingScheduleId": null,"ReleaseName": "Java","Type": "Unattended","InputArguments": null,"OutputArguments": null,"HostMachineName": null,"HasMediaRecorded": false,"HasVideoRecorded": null,"PersistenceId": null,"ResumeVersion": null,"StopStrategy": null,"RuntimeType": "Unattended","RequiresUserInteraction": true,"ReleaseVersionId": null,"EntryPointPath": "Main.xaml","OrganizationUnitId": 21,"OrganizationUnitFullyQualifiedName": null,"Reference": "","ProcessType": "Process","ProfilingOptions": null,"ResumeOnSameContext": false,"LocalSystemAccount": "","OrchestratorUserIdentity": null,"RemoteControlAccess": "None","MaxExpectedRunningTimeSeconds": null,"ServerlessJobType": null,"Id": 4262,}],
}
End. 一站式启动机器人代码
以下参数修改成自己的即可
your_domain : OC地址
scopes : 权限作用域
client_id : API ID
client_secret : API 密钥
in_folders_name : 操作流程所在文件夹
in_process_name : 流程任务名称
"""
API 启动Uipath机器人
"""
import requests
import jsonyour_domain = "https://xxxx.xxxx.com.cn"
# 权限作用域 根据个人需求填写
scopes = "OR.Administration OR.Administration.Read OR.Administration.Write OR.Analytics OR.Analytics.Read OR.Analytics.Write OR.Assets OR.Assets.Read OR.Assets.Write OR.Audit OR.Audit.Read OR.Audit.Write OR.BackgroundTasks OR.BackgroundTasks.Read OR.BackgroundTasks.Write OR.Execution OR.Execution.Read OR.Execution.Write OR.Folders OR.Folders.Read OR.Folders.Write OR.Hypervisor OR.Hypervisor.Read OR.Hypervisor.Write OR.Jobs OR.Jobs.Read OR.Jobs.Write OR.License OR.License.Read OR.License.Write OR.Machines OR.Machines.Read OR.Machines.Write OR.ML OR.ML.Read OR.ML.Write OR.Monitoring OR.Monitoring.Read OR.Monitoring.Write OR.Queues OR.Queues.Read OR.Queues.Write OR.Robots OR.Robots.Read OR.Robots.Write OR.Settings OR.Users OR.Users.Read OR.Users.Write "
client_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx"
client_secret = "xxxxxxxxxxxx"# 获取access token
def get_access_token() -> dict:url = f"{your_domain}/identity/connect/token"req_data = {"grant_type": "client_credentials","client_id": client_id,"client_secret": client_secret,"scope": scopes,}result = requests.post(url, data=req_data)if result.status_code != 200:raise Exception(f"get_access_token 异常[{result.status_code}]: {result.text}") print(f"access_token: {result.json()['access_token']}")return result.json()['access_token']# 获取文件夹folders
def get_folders(access_token, need_folder_name):url = f"{your_domain}/odata/Folders"headers = {"Authorization": f"Bearer {access_token}","accept": "application/json",}result = requests.get(url, headers=headers)if result.status_code != 200:raise Exception(f"get_folders 异常[{result.status_code}]: {result.text}")filter_folder_list = list(filter(lambda x: x["DisplayName"] == need_folder_name, result.json()["value"]))if len(filter_folder_list) != 1:raise Exception(f"get_folders 异常[筛选文件夹不唯一]: {filter_folder_list}")print(f"{need_folder_name} 的key: {filter_folder_list[0]['Key']}")return filter_folder_list[0]['Key']# 获取流程key
def get_realease_key(access_token, folder_key, process_name):url = f"{your_domain}/odata/Releases"headers = {"Authorization": f"Bearer {access_token}","accept": "application/json","X-UIPATH-FolderKey": folder_key,}result = requests.get(url, headers=headers)if result.status_code != 200:raise Exception(f"get_release_key 异常[{result.status_code}]: {result.text}")filter_releases_list = list(filter(lambda x: x["Name"] == process_name, result.json()["value"]))if len(filter_releases_list) != 1:raise Exception(f"get_release_key 异常[筛选文件夹不唯一]: {filter_releases_list}")print(f"release_key: {filter_releases_list[0]['Key']}")return filter_releases_list[0]['Key']# 启动jobs
def start_jobs(access_token, release_key, folder_key):url = f"{your_domain}/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs"req_data = {"startInfo": {"ReleaseKey": release_key, # 流程id"JobsCount": 1, # 作业数"JobPriority": "Normal", # 优先级"Strategy": "ModernJobsCount","ResumeOnSameContext": False,"RuntimeType": "Unattended",}}headers = {"Accept": "application/json","Content-Type": "application/json","Authorization": f"Bearer {access_token}","X-UIPATH-FolderKey": folder_key,}result = requests.post(url=url, data=json.dumps(req_data), headers=headers)print(f"响应状态: {result.status_code}")print(f"响应正文: {result.text}")if __name__ == "__main__":in_folders_name = "文件夹名称"in_process_name = "流程任务名称"access_token = get_access_token()folders_key = get_folders(access_token, in_folders_name)release_key = get_realease_key(access_token, folders_key, in_process_name)start_jobs(access_token,release_key, folders_key)