|
| 1 | +import importlib |
| 2 | +import inspect |
| 3 | +import re |
| 4 | +from typing import Any, Dict |
| 5 | + |
| 6 | +from mcp.server.fastmcp import FastMCP |
| 7 | + |
| 8 | +# FastMCP for UIPath Python SDK |
| 9 | +mcp = FastMCP("UIPath SDK Assistant") |
| 10 | + |
| 11 | +# Map of entity types to service classes with correct import paths |
| 12 | +SERVICE_MAP = { |
| 13 | + "buckets": "uipath._services.buckets_service.BucketsService", |
| 14 | + "assets": "uipath._services.assets_service.AssetsService", |
| 15 | + "jobs": "uipath._services.jobs_service.JobsService", |
| 16 | + "processes": "uipath._services.processes_service.ProcessesService", |
| 17 | + "queues": "uipath._services.queues_service.QueuesService", |
| 18 | + "actions": "uipath._services.actions_service.ActionsService", |
| 19 | + "context-grounding": "uipath._services.context_grounding_service.ContextGroundingService", |
| 20 | +} |
| 21 | + |
| 22 | + |
| 23 | +@mcp.tool() |
| 24 | +def get_entity_types() -> Dict[str, str]: |
| 25 | + """Returns all available entity types supported by the SDK. |
| 26 | +
|
| 27 | + This tool helps the agent understand what entity types are available |
| 28 | + for querying method information. |
| 29 | +
|
| 30 | + Returns: |
| 31 | + Dictionary mapping entity type names to their descriptions |
| 32 | + """ |
| 33 | + result = {} |
| 34 | + for entity_type, service_path in SERVICE_MAP.items(): |
| 35 | + # Try to extract descriptions from the class docstring |
| 36 | + try: |
| 37 | + module_path, class_name = service_path.rsplit(".", 1) |
| 38 | + module = importlib.import_module(module_path) |
| 39 | + service_class = getattr(module, class_name) |
| 40 | + docstring = inspect.getdoc(service_class) or "" |
| 41 | + |
| 42 | + # Clean up the docstring (remove extra whitespace and empty lines) |
| 43 | + cleaned_docstring = "\n".join( |
| 44 | + line |
| 45 | + for line in [line.strip() for line in docstring.split("\n")] |
| 46 | + if line |
| 47 | + ) |
| 48 | + |
| 49 | + # Use the entire docstring instead of just the first paragraph |
| 50 | + result[entity_type] = ( |
| 51 | + cleaned_docstring or f"{entity_type.capitalize()} related operations" |
| 52 | + ) |
| 53 | + except Exception: |
| 54 | + result[entity_type] = f"{entity_type.capitalize()} related operations" |
| 55 | + |
| 56 | + return result |
| 57 | + |
| 58 | + |
| 59 | +@mcp.tool() |
| 60 | +@mcp.tool() |
| 61 | +def get_sdk_methods(entity_type: str) -> Dict[str, Any]: |
| 62 | + """Returns detailed information about all methods available for a specific entity type. |
| 63 | +
|
| 64 | + This function only returns methods defined directly in the service class, |
| 65 | + excluding methods inherited from base classes. |
| 66 | +
|
| 67 | + Args: |
| 68 | + entity_type: The type of entity (e.g., 'buckets', 'assets', 'queues') |
| 69 | +
|
| 70 | + Returns: |
| 71 | + Dictionary containing all methods with their signatures, descriptions, and examples |
| 72 | +
|
| 73 | + Raises: |
| 74 | + ValueError: If the entity type is not found or cannot be imported |
| 75 | + """ |
| 76 | + if entity_type not in SERVICE_MAP: |
| 77 | + raise ValueError( |
| 78 | + f"Entity type '{entity_type}' not found. Use get_entity_types() to see available entity types." |
| 79 | + ) |
| 80 | + |
| 81 | + # Import the module dynamically |
| 82 | + module_path, class_name = SERVICE_MAP[entity_type].rsplit(".", 1) |
| 83 | + try: |
| 84 | + module = importlib.import_module(module_path) |
| 85 | + service_class = getattr(module, class_name) |
| 86 | + except (ImportError, AttributeError) as e: |
| 87 | + raise ValueError( |
| 88 | + f"Could not import {class_name} from {module_path}: {str(e)}" |
| 89 | + ) from e |
| 90 | + |
| 91 | + # Get class docstring for overall description |
| 92 | + class_description = inspect.getdoc(service_class) or "" |
| 93 | + |
| 94 | + methods = {} |
| 95 | + |
| 96 | + # Get the methods defined directly in this class (not in parent classes) |
| 97 | + class_methods = {} |
| 98 | + |
| 99 | + # Method 1: Use __dict__ to get only methods defined in the class itself |
| 100 | + for name, method in service_class.__dict__.items(): |
| 101 | + if callable(method) and not name.startswith("_"): |
| 102 | + # This gets methods defined directly as functions in the class |
| 103 | + class_methods[name] = method |
| 104 | + |
| 105 | + # Method 2: Check each method's __qualname__ to ensure it's defined in this class |
| 106 | + for name, method in inspect.getmembers(service_class, inspect.isfunction): |
| 107 | + if name.startswith("_"): |
| 108 | + continue |
| 109 | + |
| 110 | + # Check if method is defined in this class (not parent) |
| 111 | + method_class = method.__qualname__.split(".")[0] |
| 112 | + if method_class == class_name: |
| 113 | + class_methods[name] = method |
| 114 | + |
| 115 | + # Process each method found |
| 116 | + for name, method in class_methods.items(): |
| 117 | + # Get signature information |
| 118 | + try: |
| 119 | + sig = inspect.signature(method) |
| 120 | + except ValueError: |
| 121 | + # Skip methods that can't be inspected |
| 122 | + continue |
| 123 | + |
| 124 | + docstring = inspect.getdoc(method) or "" |
| 125 | + |
| 126 | + # Extract parameters info |
| 127 | + params = {} |
| 128 | + for param_name, param in sig.parameters.items(): |
| 129 | + if param_name == "self": |
| 130 | + continue |
| 131 | + |
| 132 | + # Try to get type hints |
| 133 | + param_type = "Any" # Default |
| 134 | + try: |
| 135 | + from typing import get_type_hints |
| 136 | + |
| 137 | + hints = get_type_hints(method) |
| 138 | + if param_name in hints: |
| 139 | + param_type = str(hints[param_name]) |
| 140 | + except Exception: |
| 141 | + pass |
| 142 | + |
| 143 | + # Extract parameter description |
| 144 | + param_desc = "" |
| 145 | + |
| 146 | + # Look for param in Args section |
| 147 | + args_pattern = r"Args:(.*?)(?:\n\n|\n[A-Z]|\Z)" |
| 148 | + args_match = re.search(args_pattern, docstring, re.DOTALL) |
| 149 | + |
| 150 | + if args_match: |
| 151 | + args_text = args_match.group(1) |
| 152 | + param_pattern = rf"\s+{param_name}\s*(?:\([^)]*\))?:\s*(.*?)(?:\n\s+\w+(?:\s*\(|\s*:)|\Z)" |
| 153 | + param_match = re.search(param_pattern, args_text, re.DOTALL) |
| 154 | + |
| 155 | + if param_match: |
| 156 | + param_desc = param_match.group(1).strip() |
| 157 | + |
| 158 | + # Store parameter info |
| 159 | + params[param_name] = { |
| 160 | + "type": param_type, |
| 161 | + "required": param.default == param.empty, |
| 162 | + "description": param_desc, |
| 163 | + "default": None if param.default == param.empty else str(param.default), |
| 164 | + } |
| 165 | + |
| 166 | + # Extract return type and description |
| 167 | + return_info = {"type": "None", "description": ""} |
| 168 | + |
| 169 | + try: |
| 170 | + from typing import get_type_hints |
| 171 | + |
| 172 | + hints = get_type_hints(method) |
| 173 | + if "return" in hints: |
| 174 | + return_info["type"] = str(hints["return"]) |
| 175 | + except Exception: |
| 176 | + pass |
| 177 | + |
| 178 | + # Look for Returns section |
| 179 | + returns_pattern = r"Returns:(.*?)(?:\n\n|\n[A-Z]|\Z)" |
| 180 | + returns_match = re.search(returns_pattern, docstring, re.DOTALL) |
| 181 | + |
| 182 | + if returns_match: |
| 183 | + return_info["description"] = returns_match.group(1).strip() |
| 184 | + |
| 185 | + # Extract examples from docstring |
| 186 | + examples = [] |
| 187 | + example_pattern = r"Examples:(.*?)(?:\n\n|\Z)" |
| 188 | + example_match = re.search(example_pattern, docstring, re.DOTALL) |
| 189 | + |
| 190 | + if example_match: |
| 191 | + example_text = example_match.group(1) |
| 192 | + code_blocks = re.findall( |
| 193 | + r"```(?:python)?\n(.*?)```", example_text, re.DOTALL |
| 194 | + ) |
| 195 | + |
| 196 | + for i, block in enumerate(code_blocks): |
| 197 | + examples.append({"title": f"Example {i + 1}", "code": block.strip()}) |
| 198 | + |
| 199 | + # If no examples found in docstring, generate a simple one based on correct SDK usage |
| 200 | + if not examples: |
| 201 | + param_strings = [] |
| 202 | + for param_name, info in params.items(): |
| 203 | + param_type = info.get("type", "") |
| 204 | + is_optional = "Optional" in param_type |
| 205 | + if not is_optional or param_name in ["name", "folder_path"]: |
| 206 | + param_strings.append(f'{param_name}="example_{param_name}"') |
| 207 | + |
| 208 | + # Check if method is async or not |
| 209 | + is_async = name.endswith("_async") or ( |
| 210 | + "async" in inspect.getsource(method).split("def")[0] |
| 211 | + if hasattr(method, "__code__") |
| 212 | + else False |
| 213 | + ) |
| 214 | + |
| 215 | + if is_async: |
| 216 | + example_code = f"""# Example usage of {name} |
| 217 | +from uipath import UiPath |
| 218 | +
|
| 219 | +# Initialize the SDK client |
| 220 | +uipath = UiPath() |
| 221 | +
|
| 222 | +# Call the async method |
| 223 | +result = await uipath.{entity_type}.{name}({", ".join(param_strings)}) |
| 224 | +print(f"Operation completed successfully: {{result}}")""" |
| 225 | + else: |
| 226 | + example_code = f"""# Example usage of {name} |
| 227 | +from uipath import UiPath |
| 228 | +
|
| 229 | +# Initialize the SDK client |
| 230 | +uipath = UiPath() |
| 231 | +
|
| 232 | +# Call the method |
| 233 | +result = uipath.{entity_type}.{name}({", ".join(param_strings)}) |
| 234 | +print(f"Operation completed successfully: {{result}}")""" |
| 235 | + |
| 236 | + examples.append({"title": f"Basic {name} example", "code": example_code}) |
| 237 | + |
| 238 | + # Extract exceptions |
| 239 | + exceptions = [] |
| 240 | + raises_pattern = r"Raises:(.*?)(?:\n\n|\n[A-Z]|\Z)" |
| 241 | + raises_match = re.search(raises_pattern, docstring, re.DOTALL) |
| 242 | + |
| 243 | + if raises_match: |
| 244 | + raises_text = raises_match.group(1) |
| 245 | + exception_pattern = r"\s+(\w+(?:\.\w+)*):\s*(.*?)(?:\n\s+\w+:|\Z)" |
| 246 | + exception_matches = re.findall(exception_pattern, raises_text, re.DOTALL) |
| 247 | + |
| 248 | + for exc_type, desc in exception_matches: |
| 249 | + exceptions.append({"type": exc_type, "description": desc.strip()}) |
| 250 | + |
| 251 | + # Get method description (first paragraph) |
| 252 | + description = docstring.split("\n\n")[0].strip() if docstring else "" |
| 253 | + |
| 254 | + # Check if there's an async version of this method |
| 255 | + has_async_version = False |
| 256 | + if not name.endswith("_async"): |
| 257 | + has_async_version = hasattr(service_class, f"{name}_async") |
| 258 | + |
| 259 | + # Store all method info |
| 260 | + methods[name] = { |
| 261 | + "description": description, |
| 262 | + "parameters": params, |
| 263 | + "return": return_info, |
| 264 | + "examples": examples, |
| 265 | + "exceptions": exceptions, |
| 266 | + "is_async": is_async or name.endswith("_async"), |
| 267 | + "has_async_version": has_async_version, |
| 268 | + "async_version": f"{name}_async" if has_async_version else None, |
| 269 | + } |
| 270 | + |
| 271 | + return { |
| 272 | + "entity_type": entity_type, |
| 273 | + "description": class_description.split("\n\n")[0].strip() |
| 274 | + if class_description |
| 275 | + else "", |
| 276 | + "methods": methods, |
| 277 | + "usage_pattern": f"uipath.{entity_type}.method_name()", |
| 278 | + } |
| 279 | + |
| 280 | + |
| 281 | +# Run the server when the script is executed |
| 282 | +if __name__ == "__main__": |
| 283 | + mcp.run() |
0 commit comments