Add Python cloud session SDK API#1258
Closed
JasonEtco wants to merge 1 commit into
Closed
Conversation
Implement cloud session support in the Python SDK, matching the Node SDK behavior from PR #1256. This adds: - copilot.cloud sub-package with CloudSession, MissionControlClient, and all related types - CopilotClient.create_cloud_session() for creating sandbox-backed cloud sessions through Mission Control - CopilotClient.connect_cloud_session() for attaching to existing tasks - Event polling with deduplication and chronological sorting - Steering helpers: send, abort, respond_to_permission, respond_to_ask_user, respond_to_elicitation, respond_to_exit_plan_mode, switch_mode - MissionControlCommandType enum covering user_message, abort, permission_response, ask_user_response, plan_approval_response, elicitation_response, and mode_switch - 13 unit tests covering task creation, repo-less tasks, owner validation, steer API, event sorting/dedup, error handling, typed handlers, and context manager support Uses X-Copilot-Agent-Slug: copilot-developer-sandbox. Requires callers to pass explicit repository or owner (no Git inference). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| # ------------------------------------------------------------------ | ||
|
|
||
| @overload | ||
| def on(self, handler: CloudSessionEventHandler, /) -> Callable[[], None]: ... |
| def on(self, handler: CloudSessionEventHandler, /) -> Callable[[], None]: ... | ||
|
|
||
| @overload | ||
| def on(self, event_type: str, handler: CloudSessionEventHandler, /) -> Callable[[], None]: ... |
| await self._poll_events() | ||
| except Exception as exc: | ||
| self._report_poll_error(exc) | ||
| except asyncio.CancelledError: |
| for handler in list(typed_handlers): | ||
| try: | ||
| handler(event) | ||
| except Exception: |
| for handler in list(self._event_handlers): | ||
| try: | ||
| handler(event) | ||
| except Exception: |
| error_body = "" | ||
| try: | ||
| error_body = http_err.read().decode("utf-8") | ||
| except Exception: |
| msg = parsed.get("message") | ||
| if isinstance(msg, str) and msg: | ||
| return msg | ||
| except json.JSONDecodeError: |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR adds a Python implementation of the “cloud session” SDK API (Mission Control task creation/attachment, event polling, and steering), aligning the Python SDK with the behavior introduced in the Node SDK (per #1256).
Changes:
- Introduces a new
copilot.cloudsubpackage with cloud-session types, a Mission Control REST client, and aCloudSessionpolling/steering wrapper. - Extends
CopilotClientwithcreate_cloud_session(...)andconnect_cloud_session(...)entry points. - Adds a dedicated unit test suite covering creation/attachment, polling/dedup, steering, and lifecycle behaviors.
Show a summary per file
| File | Description |
|---|---|
| python/copilot/cloud/types.py | Defines cloud session/task/event types and option payloads for the Python API. |
| python/copilot/cloud/mission_control_client.py | Implements Mission Control REST calls (create task, list events, steer, get task, frontend URL). |
| python/copilot/cloud/cloud_session.py | Adds CloudSession lifecycle, polling/dedup logic, subscriptions, and steering helpers. |
| python/copilot/cloud/init.py | Exposes the cloud session API surface from the new subpackage. |
| python/copilot/client.py | Adds CopilotClient cloud-session creation/connection helpers and env-driven configuration. |
| python/copilot/init.py | Re-exports cloud session symbols from the package root. |
| python/test_cloud_session.py | Adds unit coverage for the new cloud session API and behaviors. |
Copilot's findings
- Files reviewed: 7/7 changed files
- Comments generated: 5
Comment on lines
+129
to
+134
| async def list_task_events(self, task_id: str) -> list[CloudSessionEvent]: | ||
| """Poll task events from Mission Control.""" | ||
| encoded_id = urllib.request.quote(task_id, safe="") | ||
| data = await self._request_json( | ||
| f"{self._base_url}/tasks/{encoded_id}/events", | ||
| method="GET", |
Comment on lines
+264
to
+269
| def _sync_request() -> str: | ||
| data = body.encode("utf-8") if body else None | ||
| req = urllib.request.Request(url, data=data, headers=headers, method=method) | ||
| try: | ||
| with urllib.request.urlopen(req) as resp: | ||
| return resp.read().decode("utf-8") |
Comment on lines
+2153
to
+2157
| owner=owner, | ||
| repository=repository, | ||
| created_at=datetime.fromisoformat(task.created_at), | ||
| updated_at=datetime.fromisoformat(task.updated_at), | ||
| state=task.state, |
Comment on lines
+408
to
+414
| typed_handlers = self._typed_event_handlers.get(event.type) | ||
| if typed_handlers: | ||
| for handler in list(typed_handlers): | ||
| try: | ||
| handler(event) | ||
| except Exception: | ||
| pass |
Comment on lines
+351
to
+355
| async def _wait_for_initial_events(self) -> list[CloudSessionEvent]: | ||
| deadline = asyncio.get_event_loop().time() + self._initial_event_timeout_s | ||
| while True: | ||
| events = await self._client.list_task_events(self.metadata.task_id) | ||
| if events: |
Contributor
Cross-SDK Consistency Review 🔍This PR adds the cloud session API to the Python SDK, mirroring the Node.js implementation from #1256. The implementation looks well-structured and follows Python conventions (snake_case, dataclasses, TypedDict, async/await). Summary of new API surface
Python and Node naming is consistent, properly applying each language's conventions. ✅
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements the cloud session SDK API for the Python SDK, matching the Node SDK behavior from #1256.
Changes
New
copilot.cloudsub-packagetypes.py— All cloud session types:CloudRepository,CloudSessionOptions,CloudConnectOptions,CloudSessionMetadata,MissionControlTask,MissionControlCommandTypeenum, steering payloads, and event types.mission_control_client.py— HTTP client for Mission Control REST API (create_cloud_task,list_task_events,steer_task,get_task,get_frontend_url). Usesurllib.request(stdlib). IncludesCloudSessionErrorwith typed failure reasons.cloud_session.py—CloudSessionclass with connect/disconnect lifecycle, event handlers, send/abort/steering helpers, background event polling with deduplication.CopilotClientadditionscreate_cloud_session(options)— Creates a sandbox-backed cloud session. Requires explicitrepositoryorowner.connect_cloud_session(task_id, options)— Attaches to an existing Mission Control task.Design decisions
urllib.requestin executor)X-Copilot-Agent-Slug: copilot-developer-sandboxTests
13 unit tests covering task creation, repo-less tasks, owner validation, steer API, event sorting/dedup, error handling, typed handlers, unsubscribe, and context manager. All 75 tests pass.