Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds a comprehensive 1679-line runtime integration guide (GUIDE.md) that provides step-by-step instructions for creating UiPath runtime integrations for Python agentic frameworks. The guide covers the complete implementation process from project setup through testing, including schema inference, execution, streaming, human-in-the-loop support, and factory registration.
Changes:
- Added detailed architecture overview explaining the protocol hierarchy and wrapper/decorator pattern
- Provided 8-step implementation guide covering configuration, schema inference, execution, streaming, HITL, chat integration, factory setup, and testing
- Included code templates for all major components (config parser, loader, runtime, factory, storage)
- Added framework-specific implementation checklist and reference to existing integrations
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def _get_agent_input_type(agent): | ||
| """Detect the agent's input type. FRAMEWORK-SPECIFIC. | ||
|
|
||
| Examples by framework: | ||
| - OpenAI Agents: get_args(agent.__orig_class__)[0] if Agent[MyContext] | ||
| - LlamaIndex: workflow's StartEvent class | ||
| - LangGraph: graph.get_input_jsonschema() (already JSON schema) | ||
| - Pydantic AI: agent._deps_type | ||
| - Google ADK: agent.input_schema or session state type | ||
| """ | ||
| raise NotImplementedError("Implement for your framework") |
There was a problem hiding this comment.
The variable name _get_agent_input_type is defined but then the implementation raises NotImplementedError. While this is intentional as a template, the guide should clarify earlier that functions with NotImplementedError are meant to be implemented by the user for their specific framework, not left as-is.
| # Fallback: just execute and yield result | ||
| result = await self.execute(input, options) | ||
| yield result |
There was a problem hiding this comment.
The streaming example shows yielding result directly (line 667) but earlier in the same function, the execute method returns a UiPathRuntimeResult. The placeholder implementation should clarify whether this is intentionally showing a fallback or if it should construct a proper result object.
| # Fallback: just execute and yield result | |
| result = await self.execute(input, options) | |
| yield result | |
| # Fallback: run execute once and emit a single completion event | |
| result = await self.execute(input, options) | |
| yield UiPathRuntimeEvent( | |
| type="completed", | |
| result=result, | |
| ) |
| yield runtime_event | ||
|
|
||
| # Final result (MUST be last yield) | ||
| result = self._get_final_result() |
There was a problem hiding this comment.
Missing method definition: _get_final_result() is called at line 866 but is never defined in the runtime class. This method needs to be implemented or the guide should show how to track the final result during streaming (e.g., storing it in an instance variable during the stream loop).
| # In runtime.py - extend execute() and stream() | ||
|
|
||
| async def _handle_suspension(self, suspend_value, runtime_id): | ||
| """Handle agent suspension for HITL. | ||
|
|
||
| Returns UiPathRuntimeResult with SUSPENDED status and resume trigger info. | ||
|
|
||
| FRAMEWORK-SPECIFIC: | ||
| - LangGraph: detect state.interrupts, save checkpointed state | ||
| - LlamaIndex: detect InputRequiredEvent, save context via JsonPickleSerializer | ||
| """ | ||
| return UiPathRuntimeResult( | ||
| output=serialize_output(suspend_value), | ||
| status=UiPathRuntimeStatus.SUSPENDED, | ||
| # Triggers are managed by UiPathResumableRuntime wrapper | ||
| ) | ||
|
|
||
| async def _handle_resume(self, input, options): | ||
| """Resume a suspended execution. | ||
|
|
||
| FRAMEWORK-SPECIFIC: | ||
| - LangGraph: wrap input in Command(resume=input), graph resumes from checkpoint | ||
| - LlamaIndex: load saved context, send HumanResponseEvent | ||
| """ | ||
| raise NotImplementedError("Implement resume for your framework") |
There was a problem hiding this comment.
Missing method definitions: The HITL example at lines 1094-1116 shows methods _handle_suspension() and _handle_resume() but these are shown as standalone code snippets with comment "# In runtime.py - extend execute() and stream()". The guide should clarify where in the runtime class these methods should be added, and show them with proper indentation as class methods.
| # In runtime.py - extend execute() and stream() | |
| async def _handle_suspension(self, suspend_value, runtime_id): | |
| """Handle agent suspension for HITL. | |
| Returns UiPathRuntimeResult with SUSPENDED status and resume trigger info. | |
| FRAMEWORK-SPECIFIC: | |
| - LangGraph: detect state.interrupts, save checkpointed state | |
| - LlamaIndex: detect InputRequiredEvent, save context via JsonPickleSerializer | |
| """ | |
| return UiPathRuntimeResult( | |
| output=serialize_output(suspend_value), | |
| status=UiPathRuntimeStatus.SUSPENDED, | |
| # Triggers are managed by UiPathResumableRuntime wrapper | |
| ) | |
| async def _handle_resume(self, input, options): | |
| """Resume a suspended execution. | |
| FRAMEWORK-SPECIFIC: | |
| - LangGraph: wrap input in Command(resume=input), graph resumes from checkpoint | |
| - LlamaIndex: load saved context, send HumanResponseEvent | |
| """ | |
| raise NotImplementedError("Implement resume for your framework") | |
| # In runtime.py - inside your UiPath runtime class | |
| class UiPathFrameworkRuntime(UiPathRuntime): | |
| ... | |
| async def _handle_suspension(self, suspend_value, runtime_id): | |
| """Handle agent suspension for HITL. | |
| Returns UiPathRuntimeResult with SUSPENDED status and resume trigger info. | |
| FRAMEWORK-SPECIFIC: | |
| - LangGraph: detect state.interrupts, save checkpointed state | |
| - LlamaIndex: detect InputRequiredEvent, save context via JsonPickleSerializer | |
| """ | |
| return UiPathRuntimeResult( | |
| output=serialize_output(suspend_value), | |
| status=UiPathRuntimeStatus.SUSPENDED, | |
| # Triggers are managed by UiPathResumableRuntime wrapper | |
| ) | |
| async def _handle_resume(self, input, options): | |
| """Resume a suspended execution. | |
| FRAMEWORK-SPECIFIC: | |
| - LangGraph: wrap input in Command(resume=input), graph resumes from checkpoint | |
| - LlamaIndex: load saved context, send HumanResponseEvent | |
| """ | |
| raise NotImplementedError("Implement resume for your framework") |
| async def _resolve(self, obj): | ||
| """Resolve agent from various definition patterns.""" | ||
| # Direct instance - return as-is | ||
| if not callable(obj) or isinstance(obj, type): |
There was a problem hiding this comment.
The conditional check isinstance(obj, type) is incorrect logic. If obj is a type (class), it IS callable, so the or condition would make the entire check always pass when obj is a type. This would cause class types to fall through to the next section where obj() is called, which is correct for instantiating a class. However, the comment says "Direct instance - return as-is", which contradicts returning a class type as-is. The logic should likely be if not callable(obj) to handle direct instances, or the comment should be updated to clarify that classes are also instantiated.
| if not callable(obj) or isinstance(obj, type): | |
| if not callable(obj): |
| with pytest.raises(ValueError): | ||
| FrameworkConfig(str(config_file)).agents | ||
|
|
||
| def test_loader_rejects_path_outside_cwd(): |
There was a problem hiding this comment.
The validation section references test functions that check for path security ("test_loader_rejects_path_outside_cwd") but this uses await without being defined as an async function. The test function should be decorated with @pytest.mark.asyncio or defined as async def test_loader_rejects_path_outside_cwd():.
| def test_loader_rejects_path_outside_cwd(): | |
| async def test_loader_rejects_path_outside_cwd(): |
|
|
||
| | Framework | Input Model | Output Model | Graph Model | | ||
| |-----------|------------|--------------|-------------| | ||
| | **OpenAI Agents** | `messages: str\|list` + `Agent[Context]` generic param | `agent.output_type` (Pydantic model or string) | Agents + handoffs + tools (flat) | |
There was a problem hiding this comment.
The table structure shows OpenAI Agents with Agent[Context] generic parameter, but the input detection example later (line 470) uses get_args(agent.__orig_class__)[0] which only works for instances of generic classes, not all Agent definitions. This should clarify that this only applies when the agent is an instance of a generic Agent class.
| | **OpenAI Agents** | `messages: str\|list` + `Agent[Context]` generic param | `agent.output_type` (Pydantic model or string) | Agents + handoffs + tools (flat) | | |
| | **OpenAI Agents** | `messages: str\|list` + optional `Agent[Context]` generic param on generic Agent instances | `agent.output_type` (Pydantic model or string) | Agents + handoffs + tools (flat) | |
| async def initialize(self): | ||
| self._db = await aiosqlite.connect(self._db_path) | ||
| await self._db.execute("PRAGMA journal_mode=WAL") | ||
| await self._db.execute("PRAGMA synchronous=NORMAL") |
There was a problem hiding this comment.
Incomplete SQLite initialization: The storage class initializes with WAL mode and NORMAL synchronous mode, but doesn't handle the case where the database file's parent directory doesn't exist. This could cause a runtime error. Consider adding directory creation or documenting this requirement.
| storage = SqliteResumableStorage(str(tmp_path / "state.db")) | ||
| await storage.initialize() | ||
|
|
||
| trigger = UiPathResumeTrigger(interrupt_id="int-1", ...) |
There was a problem hiding this comment.
The test creates a UiPathResumeTrigger with incomplete parameters (line 1154 shows UiPathResumeTrigger(interrupt_id="int-1", ...)). The ellipsis should be replaced with the actual required fields, or the comment should note which fields are required. This makes the example non-functional as written.
| trigger = UiPathResumeTrigger(interrupt_id="int-1", ...) | |
| # NOTE: In your implementation, pass all required UiPathResumeTrigger | |
| # constructor arguments here (for example, any resume token, payload, | |
| # or metadata fields your runtime defines) in addition to interrupt_id. | |
| trigger = UiPathResumeTrigger(interrupt_id="int-1") |
| # In factory.py - new_runtime() wraps with UiPathResumableRuntime | ||
|
|
||
| from uipath.runtime import UiPathResumableRuntime | ||
|
|
||
| async def new_runtime(self, entrypoint, runtime_id, **kwargs): | ||
| agent = await self._resolve_agent(entrypoint) | ||
| base_runtime = UiPathFrameworkRuntime(agent, entrypoint, runtime_id) | ||
|
|
||
| # Only wrap if framework supports HITL | ||
| if self._supports_hitl(): | ||
| storage = await self.get_storage() | ||
| trigger_manager = self._create_trigger_manager() | ||
| return UiPathResumableRuntime( | ||
| delegate=base_runtime, | ||
| storage=storage, | ||
| trigger_manager=trigger_manager, | ||
| runtime_id=runtime_id, | ||
| ) | ||
| return base_runtime |
There was a problem hiding this comment.
Missing helper methods referenced: The code at line 1131 calls self._supports_hitl() and line 1133 calls self._create_trigger_manager(), but these methods are never defined or shown in the factory implementation. These should either be implemented in the complete factory example or documented as framework-specific helpers that need to be added.
| # In factory.py - new_runtime() wraps with UiPathResumableRuntime | |
| from uipath.runtime import UiPathResumableRuntime | |
| async def new_runtime(self, entrypoint, runtime_id, **kwargs): | |
| agent = await self._resolve_agent(entrypoint) | |
| base_runtime = UiPathFrameworkRuntime(agent, entrypoint, runtime_id) | |
| # Only wrap if framework supports HITL | |
| if self._supports_hitl(): | |
| storage = await self.get_storage() | |
| trigger_manager = self._create_trigger_manager() | |
| return UiPathResumableRuntime( | |
| delegate=base_runtime, | |
| storage=storage, | |
| trigger_manager=trigger_manager, | |
| runtime_id=runtime_id, | |
| ) | |
| return base_runtime | |
| # In factory.py - example factory wiring HITL into new_runtime() | |
| from uipath.runtime import UiPathResumableRuntime | |
| class UiPathRuntimeFactory: | |
| async def new_runtime(self, entrypoint, runtime_id, **kwargs): | |
| """ | |
| Create a new runtime for the given entrypoint. | |
| If the underlying framework supports HITL, wrap the base runtime | |
| with UiPathResumableRuntime so executions can be suspended/resumed. | |
| """ | |
| agent = await self._resolve_agent(entrypoint) | |
| base_runtime = UiPathFrameworkRuntime(agent, entrypoint, runtime_id) | |
| # Only wrap if framework supports HITL | |
| if self._supports_hitl(): | |
| storage = await self.get_storage() | |
| trigger_manager = self._create_trigger_manager() | |
| return UiPathResumableRuntime( | |
| delegate=base_runtime, | |
| storage=storage, | |
| trigger_manager=trigger_manager, | |
| runtime_id=runtime_id, | |
| ) | |
| return base_runtime | |
| def _supports_hitl(self) -> bool: | |
| """ | |
| FRAMEWORK-SPECIFIC: | |
| Return True if the target framework and this integration support | |
| suspend/resume (HITL) semantics. | |
| Override or extend this logic based on your framework capabilities. | |
| """ | |
| return False | |
| def _create_trigger_manager(self): | |
| """ | |
| FRAMEWORK-SPECIFIC: | |
| Create and return a trigger manager instance responsible for | |
| emitting and handling HITL-related triggers (e.g., approvals). | |
| Implement this method using your framework's event/trigger system. | |
| """ | |
| raise NotImplementedError("Implement trigger manager creation for your framework") |
No description provided.