LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@hidechainUiPath

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

Last updated at Posted at 2025-05-08

はじめに

  • 本記事ではMCP(Model Context Protocol)クライアントからRPA自動化処理を実行する方法について説明します。
  • 基本編をまだご覧になっていない方は、まず↓こちらの記事をお読みください。

応用編のメイントピック

  • 基本編ではMCPサーバーにてUiPath自動化処理（プロセス）ごとにPython関数として一つ一つ定義しましたが、応用編ではこの作業を効率化します。

    • 具体的にはプロセスが一元管理されているOrchestratorのAPIを利用してプロセス一覧を取得し、動的に関数を定義してMCPクライアントから実行できるように構成を変更します。
    • また個々のプロセスの引数の数や形式が異なりますが、これらもAPIで取得して関数の定義に含めていきます。

  • 次に自動化処理を連結して実行する手順について説明します。つまり一つの自動化処理の実行結果を受けて、別の自動化処理を連続して実行します。

    • 筆者としてはMCPとRPAの連携でこの部分が一番面白い部分だと感じています。従来の方法（Function Callingなど）でも自動化処理を連結することはできましたが、そのロジックは呼び出し元のプログラムで事前に定義していました。
    • 今回のMCP x RPA連携では、タスクの分割や連結部分はLLMに考えさせるようにしています。 これを実現するにはワークフローでの出力形式も少し調整が必要となりますので、その点についても後ほど詳しく説明します。

  • 基本編と応用編の構成の違いを図示したものがこちらです。

    • 基本編は自動化プロセスをあらかじめMCPサーバーで事前に関数として定義していました。

    MCPxUiPath構成図01.png

    • 応用編ではOrchestrator APIを使用してプロセス一覧を取得し、MCPサーバーにて動的に関数を定義します。

    MCPxUiPath構成図02.png

応用編実装方法

全体構成

  • UiPath Automation Cloud: Orchestratorにてプロセス（自動化処理）を一元管理
  • UiPath Robot (Windows): Orchestratorに対話型サインインで接続し、Attended Robotとして構成
  • Claude Desktop (Windows): MCPクライアントとして利用
  • Python 3.11 + uv: MCPサーバー実行環境として利用、構築手順は 基本編 を参照

プロセス一覧取得

  • Orchestrator APIを利用して自動化プロセス一覧を取得し、JSONとして保存するPythonスクリプトを実行します。この処理はMCPサーバーに含めても良いですが、今回は独立したスクリプトとして事前に手動実行するものとします。
  • MCPサーバーではこのJSONファイルをロードし、プロセスごとに @mcp.tool() 配下のPython関数として動的に定義します。

  1. Orchestrator API利用のために 個人用アクセストークン (PAT: Personal Access Token)を取得します。UiPath Robotに対話型サインインでログインしているユーザーがPATを取得します。スコープはOrchestrator API > OR.Execution.Read を割り当てます。

    PAT.PNG

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

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

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

    # get_uipath_process.py
import os
import json
import requests
import sys
from dotenv import load_dotenv

def get_uipath_process():
    """
    UiPath Orchestrator APIからプロセス定義を取得し、
    .envファイルの環境変数を使用して、それらをuipath_process.jsonに保存します
    """
    # .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)
        return False
    
    if not api_token:
        print("Error: API_TOKEN environment variable is not set in .env file", file=sys.stderr)
        return False
    
    # Bearerトークンでヘッダーを設定
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    
    try:
        # APIリクエストを実行
        orch_process_api = f"{orch_base_url}/odata/Releases/UiPath.Server.Configuration.OData.ListReleases?$select=Id,Name,Description,ProcessKey,ProcessVersion,Arguments,ProcessType,Tags&$expand=EntryPoint"
        print(f"Calling API: {orch_process_api}", file=sys.stderr)
        response = requests.get(orch_process_api, headers=headers)
        
        # リクエストが成功したか確認
        response.raise_for_status()
        
        # レスポンスJSONを解析
        response_data = response.json()
        
        # レスポンスに'value'フィールドが存在するか確認
        if 'value' not in response_data:
            print("Error: 'value' field not found in API response", file=sys.stderr)
            print(f"Response: {json.dumps(response_data, indent=2)}", file=sys.stderr)
            return False
        
        # 'value'フィールドを抽出
        process_definitions = response_data['value']
        
        # JSONファイルに保存
        output_file = 'uipath_process.json'
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(process_definitions, f, ensure_ascii=False, indent=2)
        
        print(f"Successfully saved {len(process_definitions)} process definitions 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_process()

  4. python get_uipath_process.py にてスクリプトを実行し、プロセス一覧の定義がJSONファイル uipath_process.json に保存されていることを確認します。

    02.png

No module named 'requests' などのエラーが出る場合は次の手順を実行します。

  1. get-pip.py をダウンロードして python get-pip.py を実行します。
  2. pip install requests python-dotenv を実行してPythonライブラリーをインストールします。

MCPサーバー実装

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

    # exec_uipath_process.py
import asyncio
import json
import locale
import sys
import re
import urllib.parse
from typing import Dict, Any
from mcp.server.fastmcp import FastMCP

# UiPath Robot実行パス定義
UIPATH_EXECUTABLE = r"C:\Program Files\UiPath\Studio\UiRobot.exe"

# Systemよりエンコーディング取得
SYSTEM_ENCODING = locale.getpreferredencoding()

# FastMCP初期化
mcp = FastMCP("uipath_process_executor")

async def execute_uipath_process(process_name: str, args: Dict[str, Any]) -> str:
    """
    UiPathプロセスを実行する共通関数
    
    Args:
        process_name: 実行するUiPathプロセス名
        args: UiPathプロセスに渡す引数の辞書
    
    Returns:
        プロセスの実行結果
    """
    try:
        # 入力パラメータをJSON形式で構築
        input_json = json.dumps(args)
        
        print(f"Executing UiPath process: {process_name} with args: {input_json}", file=sys.stderr)
        
        # Windowsサブプロセス非同期起動
        process = await asyncio.create_subprocess_exec(
            UIPATH_EXECUTABLE,
            "execute",
            "--process",
            process_name,
            "--input",
            input_json,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        # 標準出力と標準エラーを取得
        stdout, stderr = await process.communicate()
        
        if process.returncode == 0:
            output = stdout.decode(SYSTEM_ENCODING).strip()
            return output or "Process completed successfully."
        else:
            error_msg = stderr.decode(SYSTEM_ENCODING).strip()
            return f"Error occurred (code {process.returncode}): {error_msg}"
    except Exception as e:
        # 例外出力
        return f"Failed to execute UiPath process: {type(e).__name__} - {str(e)}"

# UiPathプロセス定義ファイルを読み込む
def load_uipath_processes():
    try:
        with open('uipath_process.json', 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        print(f"Error loading UiPath processes: {e}", file=sys.stderr)
        return []

# 引数の型を文字列からPythonの型に変換する関数
def get_arg_type(type_str: str) -> type:
    if "System.Int32" in type_str:
        return int
    elif "System.Boolean" in type_str:
        return bool
    else:
        return str  # デフォルトは文字列型

# 関数名に使用できる安全な文字列を生成する関数
def safe_function_name(text):
    # 英数字とアンダースコア以外の文字を削除し、小文字に変換
    ascii_text = re.sub(r'[^\x00-\x7F]', '', text)
    safe_text = re.sub(r'[^a-zA-Z0-9_]', '_', ascii_text)
    if safe_text and safe_text[0].isdigit():
        safe_text = 'execute_' + safe_text
    # 空文字列になった場合のデフォルト値
    if not safe_text:
        safe_text = 'execute_uipath_process'
    return 'execute_' + safe_text.lower()

# アンパサンド区切りの文字列をJSONに変換する関数
def parse_ampersand_params(param_str):
    try:
        # アンパサンド区切りの文字列をパース
        parsed = urllib.parse.parse_qs(param_str)
        # 各キーの最初の値だけを取得
        result = {k: v[0] for k, v in parsed.items()}
        return result
    except Exception as e:
        print(f"Error parsing ampersand params: {e}", file=sys.stderr)
        return {}

# プロセスがmcp=falseのタグを持っているかチェックする関数
def has_mcp_false_tag(process):
    if "Tags" in process and process["Tags"]:
        for tag in process["Tags"]:
            if tag.get("Name", "").lower() == "mcp" and tag.get("Value", "").lower() == "false":
                return True
    return False

# UiPathプロセスを動的に定義する
def define_uipath_processes():
    processes = load_uipath_processes()
    
    for process in processes:
        # ProcessTypeがProcess
        process_type = process.get("ProcessType")
        
        if process_type != "Process":
            print(f"Skipping process {process.get('Name')}: ProcessType={process_type}", file=sys.stderr)
            continue
            
        # EntryPointがNullのプロセスはスキップ
        if process.get("EntryPoint") is None:
            print(f"Skipping process {process.get('Name')}: EntryPoint is None", file=sys.stderr)
            continue
            
        # mcp=falseのタグがついているプロセスはスキップ
        if has_mcp_false_tag(process):
            print(f"Skipping process {process.get('Name')}: mcp=false tag found", file=sys.stderr)
            continue
            
        process_key = process.get("ProcessKey")
        process_name = process.get("Name")
        description = process.get("Description") or f"Execute {process_name} UiPath process"
        process_version = process.get("ProcessVersion", "Unknown")
        
        # 入力引数の解析
        input_args = []
        if process.get("Arguments") and process["Arguments"].get("Input"):
            try:
                # JSON文字列の場合はパース
                if isinstance(process["Arguments"]["Input"], str):
                    input_args_data = json.loads(process["Arguments"]["Input"])
                else:
                    input_args_data = process["Arguments"]["Input"]
                
                # 入力引数がリストの場合
                if isinstance(input_args_data, list):
                    for arg in input_args_data:
                        if arg.get("name"):
                            input_args.append(arg["name"])
            except json.JSONDecodeError:
                pass
        
        # 各プロセスに対応する関数を定義
        def create_process_function(process_key, input_args, process_version):
            async def process_function(**kwargs):
                print(f"Received kwargs: {kwargs}", file=sys.stderr)
                
                # kwargsの内側を取り出す
                if 'kwargs' in kwargs:
                    try:
                        # 文字列の場合
                        if isinstance(kwargs['kwargs'], str):
                            # JSONとしてパースを試みる
                            try:
                                inner_kwargs = json.loads(kwargs['kwargs'])
                            except json.JSONDecodeError:
                                # JSONパースに失敗した場合、アンパサンド区切りかどうかチェック
                                if '&' in kwargs['kwargs'] and '=' in kwargs['kwargs']:
                                    inner_kwargs = parse_ampersand_params(kwargs['kwargs'])
                                else:
                                    inner_kwargs = kwargs
                        else:
                            inner_kwargs = kwargs['kwargs']
                    except Exception as e:
                        print(f"Error processing kwargs: {e}", file=sys.stderr)
                        inner_kwargs = kwargs
                else:
                    inner_kwargs = kwargs
                
                print(f"Processed inner_kwargs: {inner_kwargs}", file=sys.stderr)
                
                # 引数を辞書に変換
                args = {arg: inner_kwargs.get(arg) for arg in input_args if arg in inner_kwargs}
                return await execute_uipath_process(process_key, args)
            
            # 関数のドキュメント文字列を設定
            args_doc = ', '.join([f'{arg}' for arg in input_args]) if input_args else "なし"
            
            # 出力の説明を取得
            output_description = "なし"
            if process.get("Arguments") and process["Arguments"].get("Output"):
                try:
                    # JSON文字列の場合はパース
                    if isinstance(process["Arguments"]["Output"], str) and process["Arguments"]["Output"]:
                        output_args_data = json.loads(process["Arguments"]["Output"])
                        if isinstance(output_args_data, list) and output_args_data:
                            # 出力引数がある場合は最初の出力引数の名前を使用
                            output_description = f"{output_args_data[0].get('name', 'output')}"
                except json.JSONDecodeError:
                    pass
            
            doc_string = f"""
            {description}
            
            Input:
                {args_doc}
            
            Output:
                {output_description}
                
            [Version: {process_version}]
            """
            process_function.__doc__ = doc_string
            
            # 安全な関数名を生成
            safe_name = safe_function_name(process_key)
            process_function.__name__ = safe_name
            
            return process_function
        
        # 関数を生成してMCPツールとして登録
        func = create_process_function(process_key, input_args, process_version)
        mcp.tool()(func)
        print(f"Registered UiPath process: {process_name} ({process_key}) [Version: {process_version}] as {func.__name__}", file=sys.stderr)

# UiPathプロセスを動的に定義
define_uipath_processes()

if __name__ == "__main__":
    # FastMCP実行
    mcp.run(transport='stdio')

MCPクライアント設定・実行

  1. MCPクライアントを設定します。Claude Desktop for Windowsでは %AppData%\Claude\claude_desktop_config.json を次のように設定します。

    {
  "mcpServers": {
    "uipath_attended": {
      "command": "uv",
      "args": [
        "--directory",
        "C:\\mcp\\uipath_attended",
        "run",
        "exec_uipath_process.py"
      ]
    }
  }
}

  2. Claude Desktopを起動し、「検索とツール」アイコンをクリックして、UiPathプロセス一覧がMCPツールとして追加されていることを確認します。

    01a.png

  3. プロンプトにてUiPathプロセスが実行できることを確認します。

    • 試しに hello_uipath をプロンプトで実行してみましょう。
      「"MCPxRPA応用編"をhello_uipathに渡して実行して」

    03.PNG

    • exec_uipath_process.py のソースをご覧いただければ分かりますが、hello_uipathプロセスを実行するためのexecute_hello_uipath関数は静的には定義しておらず、uipath_process.jsonから動的にロードして定義していますが、引数の受け渡しも含めて問題なく実行できます。

  4. 他のUiPathプロセスもプロンプトから実行できるか確認してみてください。

プロセス識別の注意点など

  • exec_uipath_process.py ではUiPathプロセスの説明(Description)を元にPythonにて動的に関数の説明文を作成しています。Orchestratorでのプロセスの説明が「空のプロジェクト」「N/A」「<空白>」など判別しにくい値の場合にはLLMが適切なツールとして識別できない可能性があります。この問題を回避するためにプロセスの説明には分かりやすい説明文を付加してください。

    04.png

    • ※ 最近Claude DesktopのUIが変更され、MCPツールの個々の説明文は表示されなくなったようです。

  • OrchestratorプロセスのうちMCPクライアントに実行させたくないものは mcp=false のタグを付けてください。このタグが付いているプロセスはMCPクライアントでは関数として表示されないように実装しています。

    05.png

  • uipath_process.jsonを更新するために python get_uipath_process.py を再度実行してください。

Python関数名を動的に定義するときUiPathプロセス名に含まれる日本語文字は無視しています。今回実装したMCPサーバーでは日本語文字をPython関数名に指定すると正常に動作しなかったためです。なるべく英数記号文字のみでプロセス名を定義してください。

ツール同士をLLMに連結させる

  • 次に複数のツールを連鎖的に実行する手順について説明します。ここで "ツール" と敢えて述べているのはRPA同士の連結は勿論のこと、RPA以外のツール（たとえばSlackのMCPサーバーなど）との連結も可能だからです。LLMがユーザープロンプトの意図を汲み取って、MCPサーバーで登録されているツール（RPAもRPA以外も）を適切な順序で選択して実行してくれます。

  • ここでは分かりやすくするために例として次のRPA同士を連携してみます。

  1. 実行中のAzure仮想マシン一覧を出力
  2. 出力結果を特定のSlackチャンネルに投稿

ワークフロー実装での留意点

  • LLM連携するためにはワークフローの出力変数に実行結果をセットするようにします。単独の自動化処理においては出力変数はあまり重要ではありませんが、他のシステムなどと連携する際には出力変数を構造化されたデータ形式で出力することがあります。

  • LLM連携においては出力変数の内容に応じて次のアクションが決定されますので、次の観点で出力変数を自由な形式で実装すると良いでしょう。

    • 正常終了した場合にはそのことが分かるか？
    • 次のシステムに渡す情報がすべて出力されているか？
    • エラー終了した場合にはエラー内容が分かるように出力されているか？

  • 具体的な連携を実装してみましょう。

  1. 実行中のAzure仮想マシン一覧を出力

    • クロスプラットフォームのStudioプロジェクトを作成します。
    • Azureスコープのアクティビティ を使用して、指定したAzureサブスクリプションでRunningの仮想マシンと所属するリソースグループの一覧を取得します。
    • Azureに限らず他のシステムから取得した状態一覧などの情報でも良いですし、ワークフローの実装が面倒であれば単にダミーのマシン一覧が出力変数に渡されるだけのものでも構いません。

    06a.png

  2. 出力結果を特定のSlackチャンネルに投稿

    • クロスプラットフォームのStudioプロジェクトを作成します。
    • Slackアクティビティを使います。入力引数としてチャンネル名とメッセージを受け取って投稿します。正常完了した場合にはその旨、出力引数にセットします。
    • もしチャンネル名もしくはメッセージが引数として渡されなかった場合にはエラーメッセージを出力引数にセットします。

    07a.png

  • それぞれのワークフローをOrchestratorにパブリッシュし、プロセスとしてデプロイします。そして python get_uipath_process.py を実行してuipath_process.jsonを更新します。

ツールの連結を実行する

  • 準備が整いましたので、Claude Desktopで次のプロンプトを実行してみましょう！
    「実行中のAzure VMの一覧を作ってSlackのmessage-testチャンネルに投稿して」

  • これを実現するためには言うまでもなく2つのタスクに分割して実行結果を連携する必要があります。

    1. 実行中Azure VMの一覧を作成するプロセスを実行して結果を取得
    2. Slack投稿するプロセスにチャンネル名とVM一覧を渡して実行

  • 実際にプロンプト実行してみるとLLMが上記の意図を読み取って期待通りの順序で実行してくれることが分かります。

Agent Builderとの違い

  • ここまで複数の自動化を連鎖的に実行する動作を説明しました。勘の良い方ならお気づきかと思いますが、「Agent Builderで作るのと何が違うのか？」という疑問が生じますので、私なりにお答えします。

  • Agent Builder は2025年4月30日に一般公開された、UiPathが提供するAIエージェントの開発ツールです。Agent Builderでも今回MCPクライアントでデモしたように複数のツールとプロンプトを組み合わせて複雑なタスクを実行できるように設計されています。

  • ただし 現在のところAgent Builderで作成したエージェントは、Cloud上のロボットで実行され、手元のローカル端末で実行させることはできません。 このためMCPでは可能なローカル端末上のアプリを実行したり、ローカルドライブのファイル操作といったタスクはAgent Builderでは作成できません。

  • 今回実行したデモシナリオのような「Azure VMの一覧を作成」「Slackにメッセージを投稿」といった処理はローカル端末でなくても適切な認証情報を渡していればAgent Builderでも実装可能です。

    08.png

  • さらにAgent Builderではコンテキストグラウンディングや、人間の判断が必要な時にエスカレーションする機能も備えています。今回はAgent Builderがメインのトピックではないため、軽く言及するに留めておきますが、今後はAgent Builderを中心としたUiPathのAgentic Automationにもご期待ください。

おわりに

  • 今回はMCPによるRPA実行の応用編として、Orchestratorからプロセス一覧を取得して、MCPクライアントで連鎖的に実行する手順について説明しました。ぜひ皆さんも普段お使いの自動化処理をMCPクライアントに統合し、生産性向上に役立てていただければと思います。
  • 次回はUnattended RobotをMCPクライアントから実行する方法について説明する予定です。最後までお読みいただき有難うございました！
1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?