はじめに
- 本記事ではMCP(Model Context Protocol)クライアントからRPA自動化処理を実行する方法について説明します。
- これまでAttended(有人)ロボットとstdioで連携する手順について説明してきました。
- 今回はUnattended(無人)ロボットとstdioで連携する手順について説明したいと思います。
Unattendedロボット連携実装方法
全体構成
- UiPath Automation Cloud: Orchestratorにてプロセス（自動化処理）を一元管理
-
サーバレスロボット: プロセスの実行環境
- もしくはUiPath RobotをOrchestratorにクライアント資格情報で接続し、Unattended Robotとして構成
-
Claude Desktop: MCPクライアントとして利用
- WindowsとmacOSの両方で動作しますが、今回はWindowsでの手順を説明
- Python 3.11 + uv: MCPサーバー実行環境として利用
AttendedロボットとUnattendedロボットの違い
- AttendedとUnattendedは次のような違いがあります。
| Attendedロボット | Unattendedロボット |
|---|---|
| 人間の監督が必要で、ユーザーがログインしている状態で動作します。 | 人間の監督を必要とせず、自立的に動作します。 |
| ユーザーが指示を出して特定のアクティビティを実行する必要があります。 | スケジュール実行などのトリガーを元に実行され、エンドツーエンドで完全に自動化されたプロセスを実行します。 ユーザーの指示による実行も可能です。 |
| ローカルマシン で実行されます。 | あらかじめ設定された リモートマシン で実行されます。 |
- つまりAttendedロボットではMCPクライアントと同じマシンで自動化処理が実行されますが、Unattendedロボットでは別のマシンで実行されます。このためローカルのアプリケーション操作やローカルファイルのI/O処理は実行できないことに注意してください。
- リモートマシンではあらかじめUiPath RobotがインストールされたWindowsマシンを準備するか、クロスプラットフォームで作成したワークフローが実行可能な サーバレスロボット を利用する方法があります。今回はサーバレスロボットを利用したいと思います。
構成の違い
MCP連携という文脈においてはAttendedロボットとUnattendedロボットを比較すると次のような違いになります。
-
Attendedロボット構成: MCPクライアント・MCPサーバー・Attendedロボットはすべて同じマシンで実行されます。
-
Unattendedロボット構成: MCPクライアントとMCPサーバーは同じマシンで実行されますが、Unattendedロボット(サーバレスロボット)はリモートマシンで実行されます。
-
Attendedロボット構成ではMCPサーバーから直接UiPath Robotにコマンドで実行命令を送っていましたが、Unattendedロボット構成ではMCPサーバーからAPIトリガーを実行し、Orchestrator経由でリモートのサーバレスロボットに実行命令を送信しています。
MCPサーバー実行環境の準備
全体構成をご理解いただいたところで具体的な実装手順について説明したいと思います。
-
次のコマンドを実行し、MCPサーバーの実行環境を準備します。
mkdir C:\mcp cd C:\mcp uv init uipath_unattended cd uipath_unattended uv venv .venv\Scripts\activate uv add mcp[cli]
APIトリガー作成と一覧取得
本記事ではAPIトリガーで定義されたプロセス一覧をOrchestrator APIを利用して取得します。APIトリガーの詳細は こちらの記事 を参考にしてください。
-
まずMCPクライアントから実行するプロセスに対してAPIトリガーを作成します。今回は Attendedロボット応用編 で作成した 2つのプロセス(Azure VM一覧を取得する
cloud-resource-reportと Slackに投稿するSlack-Post-Message)に対してそれぞれAPIトリガーを設定します。
- APIトリガー設定時の注意点として、MCPサーバーからAPIトリガーを同期実行して実行結果を受け取りたいため、 呼び出しモードは「同期 (ロングポーリング)」を指定 するようにしてください。
-
Orchestrator API利用のために 個人用アクセストークン (PAT: Personal Access Token)を取得します。次のスコープを選択してください。
- OR.Folders.Read
- OR.Execution.Read
- OR.Jobs
-
Postmanなどを使用して正常にAPIトリガーを実行できることを確認します。
- エンドポイントはAPIトリガー設定時に指定したスラグを使用します。
- 認証は取得したPATをBearerトークンとして指定します。
-
環境変数にてOrchestrator URLとPATを
.envファイルに設定します。ORCH_BASE_URL=https://cloud.uipath.com/{組織名}/{テナント名}/orchestrator_ API_TOKEN={取得したPAT} -
次のPythonコードを基本編で作成したプロジェクトディレクトリ
C:\mcp\uipath_unattendedに配置します。ファイル名はget_uipath_api_trigger.pyとします。このスクリプトによってAPI経由でプロセス一覧を取得し、JSONファイルに保存します。# get_uipath_api_trigger.py import os import json import requests import sys from dotenv import load_dotenv def get_uipath_api_triggers(): """ UiPath Orchestrator APIからAPIトリガー定義を取得し、 .envファイルの環境変数を使用して、それらをuipath_api_trigger.jsonに保存します Release内のIdを使用して追加のリリース情報（DescriptionとTags）を取得します """ # .envファイルを読み込む load_dotenv() # 環境変数を取得 orch_base_url = os.environ.get('ORCH_BASE_URL') api_token = os.environ.get('API_TOKEN') # 環境変数が設定されているか確認 if not orch_base_url: print("Error: ORCH_BASE_URL environment variable is not set in .env file", file=sys.stderr) print("Make sure your .env file contains a line like: ORCH_BASE_URL=https://your-orchestrator-url", file=sys.stderr) return False if not api_token: print("Error: API_TOKEN environment variable is not set in .env file", file=sys.stderr) print("Make sure your .env file contains a line like: API_TOKEN=your_token_here", file=sys.stderr) return False # Bearerトークンでヘッダーを設定 headers = { 'Authorization': f'Bearer {api_token}', 'Content-Type': 'application/json' } try: # フォルダーリストを取得 folders_api = f"{orch_base_url}/odata/Folders" print(f"Calling Folders API: {folders_api}", file=sys.stderr) folders_response = requests.get(folders_api, headers=headers) # リクエストが成功したか確認 folders_response.raise_for_status() # レスポンスJSONを解析 folders_data = folders_response.json() # レスポンスに'value'フィールドが存在するか確認 if 'value' not in folders_data: print("Error: 'value' field not found in Folders API response", file=sys.stderr) print(f"Response: {json.dumps(folders_data, indent=2)}", file=sys.stderr) return False # すべてのAPIトリガーを格納するリスト all_api_triggers = [] # 各フォルダーIDについてHTTPトリガーを取得 for folder in folders_data['value']: folder_id = folder['Id'] # フォルダーIDをヘッダーに追加 folder_headers = headers.copy() folder_headers['x-uipath-organizationunitid'] = f"{folder_id}" # APIリクエストを実行 triggers_api = f"{orch_base_url}/odata/HttpTriggers" print(f"Calling HttpTriggers API for Folder ID {folder_id}: {triggers_api}", file=sys.stderr) triggers_response = requests.get(triggers_api, headers=folder_headers) # リクエストが成功したか確認 triggers_response.raise_for_status() # レスポンスJSONを解析 triggers_data = triggers_response.json() # レスポンスに'value'フィールドが存在するか確認 if 'value' not in triggers_data: print(f"Warning: 'value' field not found in HttpTriggers API response for Folder ID {folder_id}", file=sys.stderr) continue # このフォルダーのAPIトリガーを処理 for trigger in triggers_data['value']: # Release情報を取得 - Release.Id を使用 if 'Release' in trigger and trigger['Release'] and 'Id' in trigger['Release']: release_id = trigger['Release']['Id'] release_api = f"{orch_base_url}/odata/Releases({release_id})?$select=Description,Arguments,Tags&$expand=EntryPoint" print(f"Calling Releases API for Release ID {release_id}: {release_api}", file=sys.stderr) # Release APIを呼び出し release_response = requests.get(release_api, headers=folder_headers) # リクエストが成功したか確認 release_response.raise_for_status() # レスポンスJSONを解析 release_data = release_response.json() # DescriptionとTagsを追加 if 'Description' in release_data: trigger['Description'] = release_data['Description'] else: trigger['Description'] = None if 'Tags' in release_data: trigger['Tags'] = release_data['Tags'] else: trigger['Tags'] = None if 'Arguments' in release_data: trigger['Arguments'] = release_data['Arguments'] else: trigger['Arguments'] = None # トリガーをリストに追加 all_api_triggers.append(trigger) # JSONファイルに保存 output_file = 'uipath_api_trigger.json' with open(output_file, 'w', encoding='utf-8') as f: json.dump(all_api_triggers, f, ensure_ascii=False, indent=2) print(f"Successfully saved {len(all_api_triggers)} API triggers to {output_file}", file=sys.stderr) return True except requests.exceptions.RequestException as e: print(f"API request error: {str(e)}", file=sys.stderr) return False except json.JSONDecodeError as e: print(f"JSON parsing error: {str(e)}", file=sys.stderr) return False except Exception as e: print(f"Unexpected error: {str(e)}", file=sys.stderr) return False if __name__ == "__main__": get_uipath_api_triggers() -
python get_uipath_api_trigger.pyにてPythonスクリプトを実行し、uipath_api_trigger.jsonというファイルにAPIトリガーの定義が保存されることを確認します。
No module named 'requests' などのエラーが出る場合は次の手順を実行します。
-
get-pip.py をダウンロードして
python get-pip.pyを実行します。 -
pip install requests python-dotenvを実行してPythonライブラリーをインストールします。
MCPサーバー実装
-
次にAPIトリガーのプロセス一覧が定義されたJSONファイルを利用してMCPサーバーを実装します。次のPythonコードを
exec_uipath_api_trigger.pyとして保存し、プロジェクトディレクトリC:\mcp\uipath_unattendedに配置します。# exec_uipath_api_trigger.py from typing import Any, Dict import httpx import os import sys import json import logging import datetime from mcp.server.fastmcp import FastMCP from dotenv import load_dotenv # .envファイルを読み込む load_dotenv() # 環境変数を取得 orch_base_url = os.environ.get('ORCH_BASE_URL') api_token = os.environ.get('API_TOKEN') # 環境変数が設定されているか確認 if not orch_base_url: logging.debug("Error: ORCH_BASE_URL environment variable is not set in .env file") logging.debug("Make sure your .env file contains a line like: ORCH_BASE_URL=https://your-orchestrator-url") sys.exit(1) if not api_token: logging.debug("Error: API_TOKEN environment variable is not set in .env file") logging.debug("Make sure your .env file contains a line like: API_TOKEN=your_token_here") sys.exit(1) # ログ設定 # logsディレクトリが存在しない場合は作成 if not os.path.exists('logs'): os.makedirs('logs') log_filename = f"logs/mcp_api_debug_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.log" logging.basicConfig( filename=log_filename, level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s' ) # FastMCPサーバーを初期化 mcp = FastMCP("uipath_unattended") # 環境変数を取得 api_token = os.environ.get("API_TOKEN") def parse_kwargs_string(input_dict): """ 入力辞書からkwargs値を解析して辞書形式で返す """ # kwargs値を抽出 kwargs_value = input_dict.get('kwargs', '') # すでに辞書形式の場合はそのまま返す if isinstance(kwargs_value, dict): return kwargs_value # 文字列の場合、異なる形式に対応 if isinstance(kwargs_value, str): # 入力がJSON文字列か確認（引用符のエスケープの有無に関わらず） if kwargs_value.strip().startswith('{'): try: # JSONとして解析を試みる return json.loads(kwargs_value) except json.JSONDecodeError: # 直接のJSON解析が失敗した場合、エスケープされた引用符の処理を試みる try: # エスケープされた引用符を実際の引用符に置き換え cleaned_string = kwargs_value.replace('\\"', '"') return json.loads(cleaned_string) except json.JSONDecodeError: # JSON解析がまだ失敗する場合、URL形式の解析に戻る pass # URL形式の文字列として処理（オリジナルの方法） result = {} # 文字列を&で分割してキーと値のペアを取得 pairs = kwargs_value.split('&') # 各キーと値のペアを処理 for pair in pairs: # キーと値を=で分割 if '=' in pair: key, value = pair.split('=', 1) # 最初の=でのみ分割 result[key] = value return result # kwargsが辞書でも文字列でもない場合、空の辞書を返す return {} async def call_external_api(api_trigger_url: str, payload: Dict[str, Any], method: str) -> Dict[str, Any] | None: """指定されたHTTPメソッドを使用して、提供されたペイロードで外部APIを呼び出す""" headers = { "Authorization": f"Bearer {api_token}" } # POSTとPUTリクエストにContent-Typeヘッダーを追加 if method.upper() in ["POST", "PUT"]: headers["Content-Type"] = "application/json" inner_payload = parse_kwargs_string(payload) logging.debug(f"Calling API at: {api_trigger_url}") logging.debug(f"With method: {method}") logging.debug(f"With payload: {inner_payload}") logging.debug(f"Using token: {api_token[:5]}...{api_token[-5:] if len(api_token) > 10 else ''}") async with httpx.AsyncClient() as client: try: if method.upper() == "GET": response = await client.get( api_trigger_url, headers=headers, params=inner_payload, timeout=180.0 ) elif method.upper() == "POST": response = await client.post( api_trigger_url, headers=headers, json=inner_payload, timeout=180.0 ) elif method.upper() == "PUT": response = await client.put( api_trigger_url, headers=headers, json=inner_payload, timeout=180.0 ) elif method.upper() == "DELETE": response = await client.delete( api_trigger_url, headers=headers, json=inner_payload, timeout=180.0 ) else: logging.debug(f"Unsupported HTTP method: {method}") return None logging.debug(f"Response status code: {response.status_code}") logging.debug(f"Response headers: {response.headers}") response_text = response.text logging.debug(f"Response body: {response_text[:200]}...") # 最初の200文字だけ表示 response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: logging.debug(f"HTTP Error: {e.response.status_code} - {e.response.text}") return None except httpx.RequestError as e: logging.debug(f"Request Error: {e}") return None except Exception as e: logging.debug(f"Unexpected error: {str(e)}") return None def load_api_definitions(json_file_path: str) -> None: """ JSONファイルからAPI定義を読み込み、動的に関数を定義する """ try: with open(json_file_path, 'r', encoding='utf-8') as file: api_definitions = json.load(file) for api_def in api_definitions: # ProcessTagsにmcp=falseがある場合はスキップ should_skip = False if "Tags" in api_def: for tag in api_def["Tags"]: if tag.get("Name") == "mcp" and tag.get("Value") == "false": should_skip = True break if should_skip: logging.debug(f"Skipping API {api_def.get('Name')} due to mcp=false tag") continue # API URLを構築 # ExternalReferenceから取得 if "ExternalReference" not in api_def or not api_def["ExternalReference"]: logging.debug(f"Skipping API {api_def.get('Name')} due to missing ExternalReference") continue external_ref_parts = api_def["ExternalReference"].split() if len(external_ref_parts) < 3: logging.debug(f"Skipping API {api_def.get('Name')} due to invalid ExternalReference format") continue method = external_ref_parts[0].lower() # HTTPメソッド (GET, POST, など) slug = external_ref_parts[1] # APIエンドポイントスラグ uuid = external_ref_parts[2] # UUID api_url = f"{orch_base_url}/t/{uuid}/{slug}" # 関数の説明を取得 description = api_def.get("Description", "No description provided") # 入力引数と出力引数を解析 input_args = [] if "Arguments" in api_def and api_def["Arguments"].get("Input"): try: input_args_data = json.loads(api_def["Arguments"]["Input"]) for arg in input_args_data: input_args.append(arg["name"]) except (json.JSONDecodeError, KeyError, TypeError) as e: logging.debug(f"Error parsing input arguments for {api_def.get('Name')}: {e}") output_args = [] if "Arguments" in api_def and api_def["Arguments"].get("Output"): try: output_args_data = json.loads(api_def["Arguments"]["Output"]) for arg in output_args_data: output_args.append(arg["name"]) except (json.JSONDecodeError, KeyError, TypeError) as e: logging.debug(f"Error parsing output arguments for {api_def.get('Name')}: {e}") # 関数名を決定（slugからハイフンを削除してアンダースコアに変換） function_name = f"invoke_{'_'.join(slug.split('-'))}" # 引数の説明文を生成 args_description = "" for arg in input_args: args_description += f"\n {arg}: 入力パラメータ" # 関数のドキュメント文字列を生成 docstring = f"""{description} Args:{args_description} """ # キャプチャする値を固定するためのパラメータ method_upper = method.upper() final_api_url = api_url final_input_args = list(input_args) final_output_args = list(output_args) # 動的に関数定義するための関数を生成 def create_tool_function(func_name, doc_string, api_url, method, input_args, output_args): async def func(**kwargs): # APIを呼び出す response_data = await call_external_api(api_url, kwargs, method) if not response_data: return "Unable to get a response from the external API." # レスポンスから出力を抽出 if len(output_args) == 1: output = response_data.get(output_args[0]) if output: return output else: return "No output was provided in the response." else: result = {} for out_arg in output_args: if out_arg in response_data: result[out_arg] = response_data[out_arg] if result: return result else: return "No output was provided in the response." # 重要: 関数名を正しく設定 func.__name__ = func_name func.__doc__ = doc_string func.__qualname__ = func_name # 型ヒントを設定 func.__annotations__ = {arg: str for arg in input_args} func.__annotations__["return"] = str # デコレータを適用 return mcp.tool()(func) # 関数を作成して登録 tool_function = create_tool_function( function_name, docstring, final_api_url, method_upper, final_input_args, final_output_args ) # グローバル名前空間に登録 setattr(sys.modules[__name__], function_name, tool_function) logging.debug(f"Successfully registered function: {function_name}") except Exception as e: logging.debug(f"Error loading API definitions: {e}") logging.debug(f"Error loading API definitions: {e}") # print文をlogging.debugに置き換え # JSONファイルからAPI定義を読み込む load_api_definitions("uipath_api_trigger.json") if __name__ == "__main__": # サーバーを初期化して実行 mcp.run(transport='stdio')
MCPクライアント設定・実行
-
MCPクライアントの設定を行います。Claude Desktopの設定ファイル(%AppData%\Claude\claude_desktop_config.json)を開きます。存在しない場合は新規作成します。次のように設定します。
{ "mcpServers": { "uipath_unattended": { "command": "uv", "args": [ "--directory", "C:\\mcp\\uipath_unattended", "run", "exec_uipath_api_trigger.py" ], "env": { "ORCH_BASE_URL": "https://cloud.uipath.com/{組織名}/{テナント名}/orchestrator_", "API_TOKEN": "{PAT}" } } } } -
Claude Desktopを起動し、「検索とツール」のアイコンをクリックして、APIトリガーが設定されたUiPathプロセス一覧がMCPツールとして追加されていることを確認します。
-
プロンプトにてUiPathプロセスが実行できることを確認します。
「実行中のAzure VMの一覧を作ってSlackのmessage-testチャンネルに投稿して」
-
このプロンプトではAttendedロボット応用編と同じものですが、Azure VM一覧を取得する
cloud-resource-reportと Slackに投稿するSlack-Post-Messageを連鎖的にサーバレスロボットで実行します。
おわりに
- 今回はMCPによるUnattendedロボット実行の方法として、OrchestratorからAPIトリガーを取得し、MCPクライアントからのプロンプトによって自動化処理をリモートマシンで実行する手順について説明しました。
- 本記事ではMCPサーバーとは stdio で連携しましたが、Streamable HTTP でもAPIトリガーによるUnattendedロボット実行は可能と思われますので、それについては別記事にしたいと思います。個人的にはUiPath公式のMCPソリューションが提供されることを期待しています。