Enhance Integration Services SDK for Full Connection and Connector Management

[uipreliga]

The uipath-python SDK's ConnectionsService currently provides minimal read-only access to Integration Service connections, severely limiting programmatic integration capabilities. This prevents developers from discovering connection capabilities, managing connection lifecycles, or building dynamic integrations. This issue proposes a comprehensive expansion to enable full connection management and connector schema discovery, transforming it into a powerful tool for Infrastructure-as-Code (IaC) and dynamic agent development.

2. Problem Statement & Business Impact

Current State

The ConnectionsService is one of the most limited services in the SDK, offering only three methods:

• retrieve(key): Get connection metadata (requires knowing the key).
• retrieve_token(key): Get an OAuth token for a connection.
• retrieve_event_payload(event_args): Get event data for triggers.

Critical Gaps & Business Impact

This minimal API surface creates significant gaps that hinder enterprise automation:

Gap	Business Impact
No Discovery	Blocks Dynamic Integrations: Agents and automations cannot adapt to the available tools in an environment.
No Lifecycle Management	Prevents Infrastructure-as-Code: Every connection must be configured manually in the UI, making environment setup slow, error-prone, and inconsistent.
No Schema Introspection	Increases Development Friction: Developers must constantly switch between code and the UI to find parameter names and data structures, leading to brittle, hardcoded integrations.
No Automated Administration	High Operational Overhead: Auditing, credential rotation, and health checks cannot be automated, requiring manual effort and increasing security risks.

3. Proposed Solution

We propose expanding the SDK by enhancing the ConnectionsService for instance management and introducing a new ConnectorsService for schema and template discovery.

---

Part 1: ConnectionsService Enhancements (Instance Management)

The existing ConnectionsService will be expanded to support the full lifecycle of connection instances.

API Design (sdk.connections)

class ConnectionsService(BaseService):
    # --- Existing Methods ---
    def retrieve(self, key: str) -> Connection: ...
    async def retrieve_async(self, key: str) -> Connection: ...
    def retrieve_token(self, key: str) -> ConnectionToken: ...
    async def retrieve_token_async(self, key: str) -> ConnectionToken: ...
    # ...

    # --- NEW Methods ---
    def list(
        self,
        *,
        folder_path: Optional[str] = None,
        folder_key: Optional[str] = None,
        connector_key: Optional[str] = None,
    ) -> List[Connection]:
        """Lists all connections, with optional filtering."""
        # Implementation: GET /connections_/api/v1/Connections with OData filters

    def create(
        self,
        name: str,
        connector_key: str,
        *,
        auth_properties: Dict[str, Any],
        folder_path: Optional[str] = None,
        folder_key: Optional[str] = None,
        description: Optional[str] = None,
    ) -> Connection:
        """Creates a new connection instance from a connector template."""
        # Implementation: POST /connections_/api/v1/Connections

    def update(
        self,
        key: str,
        *,
        name: Optional[str] = None,
        auth_properties: Optional[Dict[str, Any]] = None,
        description: Optional[str] = None,
    ) -> Connection:
        """Updates an existing connection's name or properties."""
        # Implementation: PUT /connections_/api/v1/Connections/{key}

    def delete(self, key: str) -> None:
        """Deletes a connection."""
        # Implementation: DELETE /connections_/api/v1/Connections/{key}

    def test(self, key: str) -> ConnectionTestResult:
        """Tests the health and validity of a connection."""
        # Implementation: POST /connections_/api/v1/Connections/{key}/test

---

Part 2: New ConnectorsService (Schema Discovery)

A new service, sdk.connectors, will be introduced for discovering available connector templates and their schemas.

API Design (sdk.connectors)

class ConnectorsService(BaseService):
    def list(self, category: Optional[str] = None) -> List[Connector]:
        """Lists all available connectors in the organization."""
        # Implementation: GET /connections_/api/v1/ConnectorTypes

    def retrieve(self, key: str) -> ConnectorSchema:
        """Retrieves the detailed schema for a specific connector."""
        # Implementation: GET /connections_/api/v1/ConnectorTypes/{key}/schema

4. Proposed Data Models

To support these new capabilities, the following Pydantic models will be introduced.

# In src/uipath/models/connectors.py (new file)
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional

class ParameterDefinition(BaseModel):
    name: str
    display_name: str
    type: str  # "string", "integer", "boolean", "object", "array"
    description: Optional[str] = None
    required: bool
    sensitive: bool = False

class ActivityDefinition(BaseModel):
    key: str
    display_name: str
    description: Optional[str] = None
    input_schema: Dict[str, ParameterDefinition]
    output_schema: Dict[str, ParameterDefinition]

class EventDefinition(BaseModel):
    key: str
    display_name: str
    description: Optional[str] = None
    payload_schema: Dict[str, Any]

class ConnectorSchema(BaseModel):
    """Complete schema for a connector type."""
    key: str
    version: str
    auth_schema: Dict[str, ParameterDefinition]
    activities: List[ActivityDefinition]
    events: List[EventDefinition]

class Connector(BaseModel):
    """Represents a type of connector available in the platform."""
    key: str
    name: str
    version: str
    category: str
    description: Optional[str] = None
    auth_types: List[str] = Field(alias="authenticationTypes")

class ConnectionTestResult(BaseModel):
    success: bool
    latency_ms: Optional[int] = None
    error: Optional[str] = None
    diagnostics: Optional[Dict[str, Any]] = None

5. Use Cases Unlocked

• Infrastructure as Code (IaC):

  # connections.yaml
  # connections:
  #   - name: "ProductionSlack"
  #     connector_key: "uipath-slack"
  #     folder: "Production"
  #     auth_properties:
  #       client_id: ${SLACK_CLIENT_ID}
  
  # deploy.py
  for conn_spec in load_config("connections.yaml"):
      sdk.connections.create(**conn_spec)

• Dynamic Agent Capabilities:

  def get_weather(location: str):
      # Check if a weather connection exists
      weather_conns = sdk.connections.list(connector_key="uipath-weather")
      if not weather_conns:
          # Guide user to create one
          schema = sdk.connectors.retrieve("uipath-weather")
          # ... prompt user for fields in schema.auth_schema ...
      # ... use connection ...

• Automated Auditing & Health Checks:

  for conn in sdk.connections.list():
      test_result = sdk.connections.test(conn.key)
      if not test_result.success:
          alert_ops_team(f"Connection '{conn.name}' is failing: {test_result.error}")

6. Implementation Considerations & Risks

• API Discovery: The underlying Integration Service management APIs may not be fully documented. Initial implementation will require discovery via browser developer tools.
• Authentication: The auth_properties dictionary will need to be flexible to support various mechanisms (OAuth2, API Key, etc.). The ConnectorsService will provide the necessary schema for validation.
• Error Handling: Introduce specific exceptions like ConnectionNotFoundError, ConnectorNotAvailableError, and InsufficientPermissionsError.
• Security: Documentation must emphasize the use of secret management systems for auth_properties and never hardcoding credentials.
• Phased Rollout (Optional):
  • Phase 1: Implement discovery and read methods (list, retrieve for both services).
  • Phase 2: Implement lifecycle management methods (create, update, delete, test).

7. Acceptance Criteria

• sdk.connections.list() returns all connections in a folder.
• sdk.connections.create() successfully provisions a new connection.
• sdk.connections.delete() successfully removes a connection.
• sdk.connectors.list() returns all available connector types.
• sdk.connectors.retrieve() returns a valid and complete schema for a given connector key.
• All new methods are covered by unit and integration tests.
• SDK documentation is updated with examples for the new services and methods.

[uipreliga]

Please contact me & @adrianpetre before starting implementation. We have code snippets in other projects that will help in the implementation.