Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions .vscode/settings.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
{
"window.title": "${rootName}${separator}${activeEditorMedium}",
"files.exclude": {
"**/*.pyc": true,
"**/__pycache__": true,
".pytest_cache": true,
".mypy_cache": true,
".ruff_cache": true,
".venv": true
},
"search.exclude": {
"**/__pycache__": true,
"**/*.pyc": true,
".venv": true,
".pytest_cache": true,
".mypy_cache": true,
".ruff_cache": true
},
// Formatting
"editor.formatOnSave": true,
"[python]": {
"editor.defaultFormatter": "charliermarsh.ruff",
"editor.codeActionsOnSave": {
"source.organizeImports": "explicit"
}
},
"workbench.colorCustomizations": {
"titleBar.activeBackground": "#0099cc",
"titleBar.inactiveBackground": "#0099cc"
},
"python.testing.pytestArgs": [
"tests"
],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true
}
9 changes: 6 additions & 3 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
[project]
name = "uipath-mcp"
version = "0.0.106"
version = "0.0.107"
description = "UiPath MCP SDK"
readme = { file = "README.md", content-type = "text/markdown" }
requires-python = ">=3.11"
dependencies = [
"mcp==1.11.0",
"mcp==1.23.3",
"pysignalr==1.3.0",
"uipath>=2.1.108, <2.2.0",
"uipath>=2.2.16, <2.3.0",
]
classifiers = [
"Development Status :: 3 - Alpha",
Expand All @@ -25,6 +25,9 @@ maintainers = [
[project.entry-points."uipath.middlewares"]
register = "uipath_mcp.middlewares:register_middleware"

[project.entry-points."uipath.runtime.factories"]
mcp = "uipath_mcp._cli._runtime:register_runtime_factory"

[project.urls]
Homepage = "https://uipath.com"
Repository = "https://github.com/UiPath/uipath-mcp-python"
Expand Down
30 changes: 30 additions & 0 deletions src/uipath_mcp/_cli/_runtime/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
"""UiPath MCP Runtime package."""

from uipath.runtime import (
UiPathRuntimeContext,
UiPathRuntimeFactoryProtocol,
UiPathRuntimeFactoryRegistry,
)

from uipath_mcp._cli._runtime._factory import UiPathMcpRuntimeFactory
from uipath_mcp._cli._runtime._runtime import UiPathMcpRuntime


def register_runtime_factory() -> None:
"""Register the MCP factory. Called automatically via entry point."""

def create_factory(
context: UiPathRuntimeContext | None = None,
) -> UiPathRuntimeFactoryProtocol:
return UiPathMcpRuntimeFactory(
context=context if context else UiPathRuntimeContext(),
)

UiPathRuntimeFactoryRegistry.register("mcp", create_factory, "mcp.json")


__all__ = [
"register_runtime_factory",
"UiPathMcpRuntimeFactory",
"UiPathMcpRuntime",
]
41 changes: 0 additions & 41 deletions src/uipath_mcp/_cli/_runtime/_context.py
Original file line number Diff line number Diff line change
@@ -1,45 +1,4 @@
from enum import Enum
from typing import Optional

from uipath._cli._runtime._contracts import UiPathRuntimeContext

from .._utils._config import McpConfig


class UiPathMcpRuntimeContext(UiPathRuntimeContext):
"""Context information passed throughout the runtime execution."""

config: Optional[McpConfig] = None
folder_key: Optional[str] = None
server_id: Optional[str] = None
server_slug: Optional[str] = None

@classmethod
def from_config(
cls, config_path: str | None = None, **kwargs: object
) -> "UiPathMcpRuntimeContext":
"""Load configuration from uipath.json file with MCP-specific handling."""
# Use the parent's implementation
instance = super().from_config(config_path, **kwargs)

# Convert to our type (since parent returns UiPathRuntimeContext)
mcp_instance = cls(**instance.model_dump())

# Add AgentHub-specific configuration handling
import json
import os

path = config_path or "uipath.json"
if os.path.exists(path):
with open(path, "r") as f:
config = json.load(f)

config_runtime = config.get("runtime", {})
if "fpsContext" in config_runtime:
fps_context = config_runtime["fpsContext"]
mcp_instance.server_id = fps_context.get("Id")
mcp_instance.server_slug = fps_context.get("Slug")
return mcp_instance


class UiPathServerType(Enum):
Expand Down
2 changes: 1 addition & 1 deletion src/uipath_mcp/_cli/_runtime/_exception.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from enum import Enum
from typing import Optional, Union

from uipath._cli._runtime._contracts import (
from uipath.runtime.errors import (
UiPathBaseRuntimeError,
UiPathErrorCategory,
UiPathErrorCode,
Expand Down
145 changes: 145 additions & 0 deletions src/uipath_mcp/_cli/_runtime/_factory.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
"""Factory for creating MCP runtime instances."""

import json
import logging
import os
import uuid
from typing import Any

from uipath.runtime import (
UiPathRuntimeContext,
UiPathRuntimeProtocol,
)
from uipath.runtime.errors import UiPathErrorCategory

from uipath_mcp._cli._runtime._exception import McpErrorCode, UiPathMcpRuntimeError
from uipath_mcp._cli._runtime._runtime import UiPathMcpRuntime
from uipath_mcp._cli._utils._config import McpConfig

logger = logging.getLogger(__name__)


class UiPathMcpRuntimeFactory:
"""Factory for creating MCP runtimes from mcp.json configuration."""

def __init__(
self,
context: UiPathRuntimeContext,
):
"""Initialize the factory.

Args:
context: UiPathRuntimeContext to use for runtime creation.
"""
self.context = context
self._mcp_config: McpConfig | None = None
self._server_id: str | None = None
self._server_slug: str | None = None

# Load fps context from uipath.json if available
self._load_fps_context()

def _load_fps_context(self) -> None:
"""
Load fps context from uipath.json for server registration.
"""
config_path = self.context.config_path or "uipath.json"
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
config_path = self.context.config_path or "uipath.json"
config_path = self.context.config_path or UiPathConfig.config_file_path()

ref: https://github.com/UiPath/uipath-python/blob/b2eed6e5d8322855144db0d3b71d2c219d1bdb71/src/uipath/platform/common/_config.py#L27

if os.path.exists(config_path):
try:
with open(config_path, "r") as f:
config: dict[str, Any] = json.load(f)

config_runtime = config.get("runtime", {})
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should be a part of the UiPathRuntimeContext, let's do something similar to what we did for conversational agents: https://github.com/UiPath/uipath-runtime-python/blob/main/src/uipath/runtime/context.py#L319

if "fpsContext" in config_runtime:
fps_context = config_runtime["fpsContext"]
self._server_id = fps_context.get("Id")
self._server_slug = fps_context.get("Slug")
except Exception as e:
logger.warning(f"Failed to load fps context: {e}")

def _load_mcp_config(self) -> McpConfig:
"""Load mcp.json configuration."""
if self._mcp_config is None:
self._mcp_config = McpConfig()
return self._mcp_config

def discover_entrypoints(self) -> list[str]:
"""Discover all MCP server entrypoints.

Returns:
List of server names that can be used as entrypoints.
"""
mcp_config = self._load_mcp_config()
if not mcp_config.exists:
return []
return mcp_config.get_server_names()

async def discover_runtimes(self) -> list[UiPathRuntimeProtocol]:
"""Discover runtime instances for all entrypoints.
This is not running as part of a job, but is intended for the dev machine.

Returns:
List of UiPathMcpRuntime instances, one per entrypoint.
"""
entrypoints = self.discover_entrypoints()
runtimes: list[UiPathRuntimeProtocol] = []

for entrypoint in entrypoints:
runtime = await self.new_runtime(entrypoint, entrypoint)
runtimes.append(runtime)

return runtimes

async def new_runtime(
self, entrypoint: str, runtime_id: str
) -> UiPathRuntimeProtocol:
"""Create a new MCP runtime instance.

Args:
entrypoint: Server name from mcp.json.
runtime_id: Unique identifier for the runtime instance.

Returns:
Configured UiPathMcpRuntime instance.

Raises:
UiPathMcpRuntimeError: If configuration is invalid or server not found.
"""
mcp_config = self._load_mcp_config()

if not mcp_config.exists:
raise UiPathMcpRuntimeError(
McpErrorCode.CONFIGURATION_ERROR,
"Invalid configuration",
"mcp.json not found",
UiPathErrorCategory.DEPLOYMENT,
)

server = mcp_config.get_server(entrypoint)
if not server:
available = ", ".join(mcp_config.get_server_names())
raise UiPathMcpRuntimeError(
McpErrorCode.SERVER_NOT_FOUND,
"MCP server not found",
f"Server '{entrypoint}' not found. Available: {available}",
UiPathErrorCategory.DEPLOYMENT,
)

# Validate runtime_id is a valid UUID, generate new one if not
try:
uuid.UUID(runtime_id)
except ValueError:
runtime_id = str(uuid.uuid4())

return UiPathMcpRuntime(
server=server,
runtime_id=runtime_id,
entrypoint=entrypoint,
folder_key=self.context.folder_key,
server_id=self._server_id,
server_slug=self._server_slug,
)

async def dispose(self) -> None:
"""Cleanup factory resources."""
self._mcp_config = None
Loading