【UiPath】MCP x RPA連携手順 (Unattendedロボット編)

# はじめに
- 本記事ではMCP(Model Context Protocol)クライアントからRPA自動化処理を実行する方法について説明します。
- これまでAttended(有人)ロボットとstdioで連携する手順について説明してきました。

https://qiita.com/hidecha/items/26087fd2b9479100eb04

https://qiita.com/hidecha/items/ae96131d19da6e53f2d7

- 今回は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ロボット|
|---|---|
|人間の監督が必要で、ユーザーがログインしている状態で動作します。|人間の監督を必要とせず、自立的に動作します。|
|ユーザーが指示を出して特定のアクティビティを実行する必要があります。|スケジュール実行などのトリガーを元に実行され、エンドツーエンドで完全に自動化されたプロセスを実行します。<br>ユーザーの指示による実行も可能です。|
|**ローカルマシン** で実行されます。|あらかじめ設定された **リモートマシン** で実行されます。|

- つまりAttendedロボットではMCPクライアントと同じマシンで自動化処理が実行されますが、Unattendedロボットでは別のマシンで実行されます。このためローカルのアプリケーション操作やローカルファイルのI/O処理は実行できないことに注意してください。
- リモートマシンではあらかじめUiPath RobotがインストールされたWindowsマシンを準備するか、クロスプラットフォームで作成したワークフローが実行可能な **サーバレスロボット** を利用する方法があります。今回はサーバレスロボットを利用したいと思います。

### 構成の違い
MCP連携という文脈においてはAttendedロボットとUnattendedロボットを比較すると次のような違いになります。
- **Attendedロボット構成**: MCPクライアント・MCPサーバー・Attendedロボットはすべて同じマシンで実行されます。
    ![MCPxAR構成図.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/7dd72d05-2f40-404c-85ed-9245ed7cca67.png)

- **Unattendedロボット構成**: MCPクライアントとMCPサーバーは同じマシンで実行されますが、Unattendedロボット(サーバレスロボット)はリモートマシンで実行されます。
    ![MCPxUR構成図.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/77d79bb3-015f-4b98-80b5-ef20f2f1c218.png)

- Attendedロボット構成ではMCPサーバーから直接UiPath Robotにコマンドで実行命令を送っていましたが、Unattendedロボット構成ではMCPサーバーからAPIトリガーを実行し、Orchestrator経由でリモートのサーバレスロボットに実行命令を送信しています。

## MCPサーバー実行環境の準備

全体構成をご理解いただいたところで具体的な実装手順について説明したいと思います。
- 次のコマンドを実行し、MCPサーバーの実行環境を準備します。

  ```powershell
  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トリガーの詳細は [こちらの記事](https://qiita.com/hidecha/items/5b0733639dcd1420531a) を参考にしてください。

1. まずMCPクライアントから実行するプロセスに対してAPIトリガーを作成します。今回は [Attendedロボット応用編](https://qiita.com/hidecha/items/ae96131d19da6e53f2d7) で作成した 2つのプロセス(Azure VM一覧を取得する `cloud-resource-report` と Slackに投稿する `Slack-Post-Message`)に対してそれぞれAPIトリガーを設定します。

    ![03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/39c0990f-29fd-4518-b1e3-55852eb84238.png)
    
    ![05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/6c992ac4-ea13-4a05-9b83-a0280e1d2fef.png)

    - APIトリガー設定時の注意点として、MCPサーバーからAPIトリガーを同期実行して実行結果を受け取りたいため、 **呼び出しモードは「同期 (ロングポーリング)」を指定** するようにしてください。

2. Orchestrator API利用のために [個人用アクセストークン](https://qiita.com/hidecha/items/609dc6723dffb3bd81c6) (**PAT**: Personal Access Token)を取得します。次のスコープを選択してください。
    - OR.Folders.Read
    - OR.Execution.Read
    - OR.Jobs

    ![08.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/43348ff6-3527-469b-a0b1-d1f114c38149.png)

3. Postmanなどを使用して正常にAPIトリガーを実行できることを確認します。
    - エンドポイントはAPIトリガー設定時に指定したスラグを使用します。
    - 認証は取得したPATをBearerトークンとして指定します。

    ![06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/d4fe867a-ca4d-4d61-9199-2631d07e63b6.png)

4. 環境変数にてOrchestrator URLとPATを `.env` ファイルに設定します。

    ```bash
    ORCH_BASE_URL=https://cloud.uipath.com/{組織名}/{テナント名}/orchestrator_
    API_TOKEN={取得したPAT}
    ```

5. 次のPythonコードを基本編で作成したプロジェクトディレクトリ `C:\mcp\uipath_unattended` に配置します。ファイル名は `get_uipath_api_trigger.py` とします。このスクリプトによってAPI経由でプロセス一覧を取得し、JSONファイルに保存します。

      ```python
      # 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()
      ```

6. `python get_uipath_api_trigger.py` にてPythonスクリプトを実行し、`uipath_api_trigger.json` というファイルにAPIトリガーの定義が保存されることを確認します。

    ![07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/9df4dbf9-1529-41ba-b3cf-bd36af0363f4.png)

:::note info
No module named 'requests' などのエラーが出る場合は次の手順を実行します。
1. [get-pip.py](https://bootstrap.pypa.io/get-pip.py) をダウンロードして `python get-pip.py` を実行します。
2. `pip install requests python-dotenv` を実行してPythonライブラリーをインストールします。
:::

## MCPサーバー実装
- 次にAPIトリガーのプロセス一覧が定義されたJSONファイルを利用してMCPサーバーを実装します。次のPythonコードを `exec_uipath_api_trigger.py` として保存し、プロジェクトディレクトリ `C:\mcp\uipath_unattended` に配置します。

  ```python
  # 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クライアント設定・実行
1. MCPクライアントの設定を行います。Claude Desktopの設定ファイル(%AppData%\Claude\claude_desktop_config.json)を開きます。存在しない場合は新規作成します。次のように設定します。

      ```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}"
            }
          }
        }
      }
      ```

2. Claude Desktopを起動し、「検索とツール」のアイコンをクリックして、APIトリガーが設定されたUiPathプロセス一覧がMCPツールとして追加されていることを確認します。
    
    ![01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/88f09374-8ffa-4c3c-870d-d60a108d8c20.png)

3. プロンプトにてUiPathプロセスが実行できることを確認します。
「実行中のAzure VMの一覧を作ってSlackのmessage-testチャンネルに投稿して」

- このプロンプトではAttendedロボット応用編と同じものですが、Azure VM一覧を取得する `cloud-resource-report` と Slackに投稿する `Slack-Post-Message` を連鎖的にサーバレスロボットで実行します。

    ![02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/b8ec9e9b-23ed-428e-b2cb-263f3cc1d821.png)

# おわりに
- 今回はMCPによるUnattendedロボット実行の方法として、OrchestratorからAPIトリガーを取得し、MCPクライアントからのプロンプトによって自動化処理をリモートマシンで実行する手順について説明しました。
- 本記事ではMCPサーバーとは [stdio](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#stdio) で連携しましたが、[Streamable HTTP](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http) でもAPIトリガーによるUnattendedロボット実行は可能と思われますので、それについては別記事にしたいと思います。個人的にはUiPath公式のMCPソリューションが提供されることを期待しています。