ProcessCube Robot Agent - Architecture

Technische Architektur und Systemdesign

🏛️ System-Übersicht

Das Diagramm zeigt:

ProcessCube Engine Integration (Blau) - BPMN Workflow und External Tasks
FastAPI REST Server (Grün) - Port 42042 mit allen Komponenten
External Task Handler (Lila) - Task-Routing und Orchestrierung
MultiRunnerFactoryCreator (Rot) - Zentrale Factory-Orchestrierung
RCC Robot Factory & Agent (Orange) - Für UI/Web Automation
UV Robot Factory & Agent (Cyan) - Für API/Data Processing
Project Watcher (Violett) - Hot Reload Funktionalität
Externe Dependencies (Grau) - RCC CLI, Work Items, UV Tool

📝 Editierbare Version: Das Diagramm kann als Excalidraw-Datei bearbeitet werden: architecture-diagram.excalidraw

Kernarchitektur

Die Architektur basiert auf einem Dual-Engine Ansatz mit zwei unterschiedlichen Robot-Typen:

RCC Robots (Robot Framework)

Optimal für UI/Web-Automation
Verwendet Robocorp RCC CLI
Conda-basiertes Dependency Management
Topic-Präfix: rcc.

UV Robots (Pure Python)

Optimal für API-Integration und Datenverarbeitung
Verwendet UV Package Manager
Schnelle Python Virtual Environments
Topic-Präfix: uv.

Gemeinsame Komponenten:

FastAPI REST Server (Port 42042)
External Task Handler für ProcessCube Integration
Work Items (JSON) für Input/Output
Factory Pattern für Robot Discovery
Hot Reload via File System Watcher

📦 Komponenten-Architektur

1. Entry Point: `processcube_robot_agent/main.py`

Verantwortung: CLI-Interface und Service-Start

# Kommandos:
# - serve: Startet REST API + External Task Handler
# - pack: Packt alle Robots
# - watch: Überwacht robot directory für Änderungen

Initialisierung:

Config laden
Logging einrichten
Python Debugger starten (optional)
CLI-Kommando ausführen

2. REST API: `rest_api_command.py` & `rest_api/`

Port: 42042 (konfigurierbar) Framework: FastAPI + uvicorn

Endpoints

# GET /robot_agents/robots
# Returns: {"topics": [{"name": "...", "topic": "..."}]}

Architektur:

FastAPI App
    ├── /robot_agents/robots (GET)
    │   └── RobotTaskHandlerFactoryCreator.get_all_robots()
    │       └── Scans robots/installed/rcc/*.zip
    │           └── Returns registered topics
    │
    └── Health Checks
        └── Service status

Flow:

HTTP Request
    ↓
FastAPI Route Handler
    ↓
Builder Factory
    ↓
Scan zip packages
    ↓
Return topic mappings
    ↓
JSON Response

3. External Task Handler: `external_task/`

Verantwortung: ProcessCube Integration

Flow:

ProcessCube Task Event
    ↓
External Task Handler
    ├── Lookup Robot (by topic)
    ├── Load Robot Package
    └── Execute via RobotAgent
        ├── Prepare Input
        ├── Run Robot
        └── Extract Output
    ↓
Return Result to ProcessCube

Implementation:

class RobotTaskHandler(BaseHandler):
    def execute(self, payload, task):
        # 1. Get robot from topic
        robot_agent = RobotAgent(topic=task['topic'])

        # 2. Execute robot
        result = robot_agent.execute(payload, task)

        # 3. Return result
        return result

4. Robot Execution Engine: `robot_agent/rcc/`

Zentrale Komponente: RobotAgent

Execution Flow

RobotAgent.execute(payload, task)
    │
    ├─► Step 1: create_input_data(payload)
    │   └─ Convert dict → JSON work items
    │
    ├─► Step 2: unwrap(robot_zip)
    │   └─ Extract zip to temp directory
    │       └─ Validate robot.yaml exists
    │
    ├─► Step 3: run_robot(unwrapped_dir)
    │   └─ Execute: rcc run -c <dir>
    │       └─ Wait for completion
    │       └─ Capture stdout/stderr
    │
    ├─► Step 4: read_output_data(output_dir)
    │   └─ Parse output.xml
    │   └─ Extract work item results
    │
    └─► Return result dict
        └─ Success or error

Key Methods

class RobotAgent(RobotAgentBase):
    def execute(payload, task) -> Dict
        """Main entry point"""

    def create_input_data(payload) -> Path
        """Prepare JSON work items"""

    def unwrap(robot_path) -> Path
        """Unpack robot.zip using RCC"""

    def run_robot(unwrapped_path) -> Process
        """Execute via 'rcc run'"""

    def read_output_data(output_dir) -> Dict
        """Extract results from output.xml"""

5. Robot Discovery & Registration: `robot_agent/rcc/robot_task_handler_factory.py`

Verantwortung: Mapping Topics → Robot Packages

Factory Pattern Implementierung

RobotTaskHandlerFactoryCreator
    │
    ├─► Scan robots/installed/rcc/*.zip
    │   └─ For each zip:
    │       └─ Create Factory instance
    │
    ├─► Build topic names
    │   └─ "rcc/webui"
    │   └─ "rcc/windows/ui"
    │
    └─► Register with external task client
        └─ Topics → Handler mapping

Topic Building:

# Example path: robots/installed/rcc/webui.zip
# Topic prefix: "rcc"
# Result topic: "rcc/webui"

# Example path: robots/installed/rcc/windows/ui.zip
# Result topic: "rcc/windows/ui"

Class Diagram

RobotTaskHandlerFactoryCreator
    ├─ _factory_builder: FactoryBuilder
    ├─ _factories: Dict[str, Factory]
    │
    ├─ __init__(wrap_dir, topic_prefix)
    ├─ get_factories() → List[Factory]
    ├─ get_all_topics() → List[str]
    └─ build_external_task_handlers()

Factory
    ├─ robot_path: Path
    ├─ topic: str
    └─ create_handler() → RobotTaskHandler

FactoryBuilder
    └─ build(robot_path) → Factory

6. Project Packing: `robot_agent/rcc/project_packer.py`

Verantwortung: Robot-Projektierung und Packaging

Input: robots/src/rcc/<project>/
    ├─ robot.yaml
    ├─ tasks.robot
    ├─ conda.yaml
    └─ locators.json

    ↓ (rcc robot wrap)

Output: robots/installed/rcc/<project>.zip
    └─ Complete packaged robot
       ready for execution

Execution

ProjectPacker:
    ├─ pack_folder(robot_path)
    │   └─ Find robot.yaml
    │   └─ Execute: rcc robot wrap
    │   └─ Output to wrap_dir
    │
    └─ pack_all(source_dir)
        └─ Recursively find all robot.yaml
        └─ Pack each one

7. Hot-Reload Watcher: `robot_agent/rcc/project_watcher.py`

Verantwortung: Automatische Neu-Paketierung

File System Event (watchdog)
    │
    ├─ File changed in robots/src/rcc/
    │
    ├─► RobotsFileSystemEventHandler.on_any_event()
    │   │
    │   ├─ Find parent robot.yaml
    │   ├─ Pack robot
    │   └─ Register with factory creator
    │
    └─► Factory discovers new/updated robot
        └─ Topics updated automatically

Implementation:

class RobotsFileSystemEventHandler(FileSystemEventHandler):
    def on_any_event(self, event):
        changed_path = Path(event.src_path)
        robot_yaml_path = find_robot_yaml(changed_path)
        if robot_yaml_path:
            pack_folder(robot_yaml_path)
            factory_creator.reload()  # Re-register topics

Watch Process:

1. Service start mit watch_dir monitoring
2. Datei ändert sich → Watchdog erkennt Event
3. Event → auf robot.yaml prüfen
4. Gefunden → mit RCC packen
5. Neu gepackt → Factory aktualisieren
6. Topics re-registrieren

8. Configuration Management

Dateiformat: JSON

Struktur:

{
  "debugging": {
    "enabled": boolean,
    "hostname": string,
    "port": integer
  },
  "engine": {
    "url": string
  },
  "rcc": {
    "topic_prefix": string,
    "wrap_dir": string,
    "unwrap_dir": string,
    "start_watch_project_dir": boolean,
    "project_dir": string
  },
  "rest_api": {
    "port": integer,
    "host": string
  }
}

Laden:

import json
import os

config_file = os.getenv('CONFIG_FILE')
with open(config_file) as f:
    config = json.load(f)

🎨 Data Structures

Work Items (Input/Output)

Robot Framework RPA Work Items:

# Input (from ProcessCube)
{
    "order_id": "ORD-12345",
    "customer": "Acme Corp",
    "items": [
        {"sku": "ITEM-001", "qty": 5}
    ]
}

# In Robot Framework Access:
${order_id}=    Get Work Item Variable    order_id
${customer}=    Get Work Item Variable    customer

# Robot Output:
Set Work Item Variable    status    completed
Set Work Item Variable    result    ${result}

# Output (back to ProcessCube)
{
    "status": "completed",
    "result": {...},
    "error": null
}

Topic Format

<topic_prefix>/<robot_path>

Examples:
- rcc/webui
- rcc/windows/ui
- rcc/windows-example-calculator

🔄 Request Flow Example

Scenario: User startet Robot Task in ProcessCube

1. USER ACTION
   └─ Klick "Start Task" in ProcessCube UI

2. PROCESSCUBE
   └─ Event: External Task mit topic="rcc/webui"
       Payload: {"url": "https://example.com"}

3. ROBOT AGENT - External Task Handler
   └─ Listen auf topic="rcc/webui"
   └─ RobotTaskHandler.execute() aufgerufen
       ├─ Payload: {"url": "..."}
       └─ Task: {"id": "task-123", "topic": "rcc/webui"}

4. ROBOT AGENT - RobotAgent
   └─ execute(payload, task)
       ├─ create_input_data(payload)
       │   └─ JSON: {url: "..."}
       ├─ Find robot_path for topic "rcc/webui"
       │   └─ robots/installed/rcc/webui.zip
       ├─ unwrap(webui.zip)
       │   └─ temp/robots/rcc/unwrapped/webui/
       ├─ run_robot(unwrapped_dir)
       │   └─ Execute: rcc run -c temp/robots/rcc/unwrapped/webui/
       │   └─ Robot Framework runs tasks.robot
       │   └─ Output: temp/robots/rcc/unwrapped/webui/output/
       └─ read_output_data(output_dir)
           └─ Parse output.xml
           └─ Extract results

5. RESULT
   └─ {
        "status": "completed",
        "result": {"screenshot": "..."},
        "error": null
      }

6. PROCESSCUBE
   └─ Update Task status
   └─ Continue BPMN process

📊 Design Patterns Used

1. Factory Pattern

Location: robot_task_handler_factory.py

# FactoryBuilder creates Factory instances
# RobotTaskHandlerFactoryCreator manages all factories
factory = factory_builder.build(robot_path)  # Create handler for specific robot

2. Observer Pattern

Location: project_watcher.py

# watchdog monitors file system
# FileSystemEventHandler observes changes
# Auto-triggers robot re-packing
observer = Observer()
observer.schedule(event_handler, path)

3. Adapter Pattern

Location: external_task/robot_task_handler.py

# Adapts RobotAgent to ProcessCube's BaseHandler interface
class RobotTaskHandler(BaseHandler):
    def execute(self, payload, task):
        return self._robot_agent.execute(payload, task)

4. Strategy Pattern

Location: robot_agent/base_agent.py

# Different agent strategies (RCC, direct execution, etc)
class BaseAgent(ABC):
    @abstractmethod
    def execute(self, payload, task): ...

class RobotAgent(BaseAgent):
    def execute(self, payload, task): ...

🔌 Integration Points

1. ProcessCube SDK

from processcube_sdk.external_task import ExternalTaskClient
from processcube_sdk.external_task import BaseHandler

# Connects to ProcessCube engine
external_task_client = ExternalTaskClient(
    base_url="http://localhost:56100",
    auth_type="NONE"
)

# Subscribe to topics
external_task_client.subscribe(
    topic="rcc/webui",
    handler=RobotTaskHandler()
)

2. RCC CLI

import subprocess

# Unwrap robot
subprocess.run([
    "rcc", "robot", "unwrap",
    "-z", robot_path,
    "-d", unwrap_dir,
    "--force"
])

# Run robot
subprocess.run([
    "rcc", "run",
    "-c", unwrapped_dir
])

3. FastAPI Server

from fastapi import FastAPI
app = FastAPI()

@app.get("/robot_agents/robots")
async def get_robots():
    return factory_creator.get_all_robots()

# Run with uvicorn
uvicorn.run(app, host="0.0.0.0", port=42042)

🧵 Threading & Async

Async Architecture

Main Event Loop (uvicorn/asyncio)
    │
    ├─► HTTP Requests (FastAPI)
    │   └─ Async handlers
    │
    ├─► External Task Processing
    │   └─ Background threads
    │   └─ RobotAgent.execute() blocking
    │
    └─► File Watcher (watchdog)
        └─ Separate observer thread
        └─ Non-blocking events

Note: RobotAgent.execute() ist blockierend (wartet auf rcc run). Dies ist in Ordnung, da External Task Handler im Thread-Pool läuft.

🔒 Security Architecture

Current State

⚠️ Shell injection vulnerability in subprocess.run() with shell=True
⚠️ No input validation on work items
✅ Robot execution isolated in temp directories
✅ File permissions restricted by OS

Recommended Improvements

Use subprocess with argument list (no shell=True)
Validate work item schema
Sandbox robot execution (containers)
Secure config management (.env)
Input sanitization

📈 Scalability Considerations

Current Design

Single-threaded robot execution (sequential)
In-memory robot registry
Local file system for robots

Scaling Options

Multiple Agent Instances

# Run multiple agents on different ports
CONFIG_FILE=config1.json npm run processcube_robot_agent  # Port 42042
CONFIG_FILE=config2.json npm run processcube_robot_agent  # Port 42043
CONFIG_FILE=config3.json npm run processcube_robot_agent  # Port 42044

# Load balancer in front

Distributed Robot Registry
- Central database for robot metadata
- Redis/memcached for caching topics
- API-based discovery
Containerization
- Docker container per agent
- Kubernetes orchestration
- Resource limits per robot

🏗️ Extension Points

Adding New Features

1. Custom Robot Types

# Currently: RCC-based robots
# Could add: Direct Python execution, Docker containers, etc

class CustomAgent(BaseAgent):
    def execute(self, payload, task):
        # Custom logic
        pass

2. Alternative Work Item Formats

# Currently: JSON
# Could add: XML, CSV, custom formats

class WorkItemAdapter:
    def convert_from(self, payload): ...
    def convert_to(self, result): ...

3. Custom Topic Builders

# Currently: File path based
# Could add: Database lookup, API-based, regex patterns

class TopicBuilder:
    def build_topic(self, robot_path) -> str:
        # Custom logic
        pass

🧪 Testing Architecture

Current State

❌ No unit tests
❌ No integration tests
❌ No e2e tests

Recommended Structure

tests/
├── unit/
│   ├── test_robot_agent.py
│   ├── test_project_packer.py
│   ├── test_robot_task_handler_factory.py
│   └── test_external_task_handler.py
│
├── integration/
│   ├── test_rcc_integration.py
│   └── test_processcube_integration.py
│
└── e2e/
    ├── test_full_flow.py
    └── test_webui_robot.py

📝 API Contracts

Input Schema (ProcessCube → Agent)

{
    "taskId": str,
    "topic": str,
    "workItemId": str,
    "variables": {
        # Dynamic payload from ProcessCube
        "key": "value"
    }
}

Output Schema (Agent → ProcessCube)

{
    "success": bool,
    "data": {
        # Robot output
    },
    "error": Optional[str]
}

REST API Schema

# GET /robot_agents/robots
{
    "topics": [
        {
            "name": str,      # Short name
            "topic": str      # Full topic
        }
    ]
}

🎯 Key Architectural Decisions

Decision	Rationale	Trade-off
RCC for packaging	Robocorp standard, multi-OS	Dependency on RCC
Sequential execution	Simplicity	No parallelization
JSON config	Human readable	No schema validation
File watcher	Hot reload	Complexity
FastAPI	Async capable	Extra dependency
Subprocess for RCC	Direct execution	Shell injection risk

Dokumentation Version: 1.0 Zuletzt aktualisiert: November 2025

Uh oh!

FilesExpand file tree

ARCHITECTURE.md

Latest commit

History

ARCHITECTURE.md

File metadata and controls

ProcessCube Robot Agent - Architecture

🏛️ System-Übersicht

Kernarchitektur

📦 Komponenten-Architektur

1. Entry Point: processcube_robot_agent/__main__.py

2. REST API: rest_api_command.py & rest_api/

Endpoints

3. External Task Handler: external_task/

4. Robot Execution Engine: robot_agent/rcc/

Execution Flow

Key Methods

5. Robot Discovery & Registration: robot_agent/rcc/robot_task_handler_factory.py

Factory Pattern Implementierung

Class Diagram

6. Project Packing: robot_agent/rcc/project_packer.py

Execution

7. Hot-Reload Watcher: robot_agent/rcc/project_watcher.py

8. Configuration Management

🎨 Data Structures

Work Items (Input/Output)

Topic Format

🔄 Request Flow Example

Scenario: User startet Robot Task in ProcessCube

📊 Design Patterns Used

1. Factory Pattern

2. Observer Pattern

3. Adapter Pattern

4. Strategy Pattern

🔌 Integration Points

1. ProcessCube SDK

2. RCC CLI

3. FastAPI Server

🧵 Threading & Async

Async Architecture

🔒 Security Architecture

Current State

Recommended Improvements

📈 Scalability Considerations

Current Design

Scaling Options

🏗️ Extension Points

Adding New Features

1. Custom Robot Types

2. Alternative Work Item Formats

3. Custom Topic Builders

🧪 Testing Architecture

Current State

Recommended Structure

📝 API Contracts

Input Schema (ProcessCube → Agent)

Output Schema (Agent → ProcessCube)

REST API Schema

🎯 Key Architectural Decisions

1. Entry Point: `processcube_robot_agent/main.py`

2. REST API: `rest_api_command.py` & `rest_api/`

3. External Task Handler: `external_task/`

4. Robot Execution Engine: `robot_agent/rcc/`

5. Robot Discovery & Registration: `robot_agent/rcc/robot_task_handler_factory.py`

6. Project Packing: `robot_agent/rcc/project_packer.py`

7. Hot-Reload Watcher: `robot_agent/rcc/project_watcher.py`