-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathcanvas.py
More file actions
164 lines (123 loc) · 4.91 KB
/
canvas.py
File metadata and controls
164 lines (123 loc) · 4.91 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""
Canvas declarations, provider callbacks, and host-side canvas RPC types.
The Copilot CLI runtime sends inbound canvas JSON-RPC requests to any session
that declares canvases. The SDK forwards every such request to a single
user-supplied :class:`CanvasHandler`; multiplexing across multiple declared
canvases is the implementor's responsibility (for example by switching on
``ctx.canvas_id``).
.. note::
**Experimental.** Canvas types are part of an experimental wire-protocol
surface and may change or be removed in future SDK or CLI releases.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
from .generated.rpc import (
CanvasAction,
CanvasHostContext,
CanvasHostContextCapabilities,
CanvasJsonSchema,
CanvasProviderCloseRequest,
CanvasProviderInvokeActionRequest,
CanvasProviderOpenRequest,
CanvasProviderOpenResult,
OpenCanvasInstance,
)
__all__ = [
"CanvasAction",
"CanvasDeclaration",
"CanvasError",
"CanvasHandler",
"CanvasHostContext",
"CanvasHostContextCapabilities",
"CanvasJsonSchema",
"ExtensionInfo",
"OpenCanvasInstance",
]
@dataclass
class ExtensionInfo:
"""Stable extension identity for session participants that provide canvases.
Serializes to ``{"source": ..., "name": ...}`` on the wire.
.. note::
**Experimental.** This type is part of an experimental wire-protocol
surface and may change or be removed in future SDK or CLI releases.
"""
source: str
"""Extension namespace/source, e.g. ``"github-app"``."""
name: str
"""Stable provider name within the source namespace."""
def to_dict(self) -> dict[str, Any]:
return {"source": self.source, "name": self.name}
@dataclass
class CanvasDeclaration:
"""Declarative metadata for a single canvas, sent on create/resume.
.. note::
**Experimental.** This type is part of an experimental wire-protocol
surface and may change or be removed in future SDK or CLI releases.
"""
id: str
"""Canvas identifier, unique within the declaring connection."""
display_name: str
"""Human-readable name shown in host UI and canvas pickers."""
description: str
"""Short description shown to the agent in canvas catalogs."""
input_schema: CanvasJsonSchema | None = None
"""JSON Schema for the ``input`` payload accepted by ``canvas.open``."""
actions: list[CanvasAction] | None = None
"""Agent-callable actions this canvas exposes."""
def to_dict(self) -> dict[str, Any]:
result: dict[str, Any] = {
"id": self.id,
"displayName": self.display_name,
"description": self.description,
}
if self.input_schema is not None:
result["inputSchema"] = self.input_schema
if self.actions is not None:
result["actions"] = [action.to_dict() for action in self.actions]
return result
class CanvasError(Exception):
"""Structured error returned from canvas handlers.
.. note::
**Experimental.** This type is part of an experimental wire-protocol
surface and may change or be removed in future SDK or CLI releases.
"""
def __init__(self, code: str, message: str) -> None:
self.code = code
self.message = message
super().__init__(f"{code}: {message}")
def to_envelope(self) -> dict[str, str]:
return {"code": self.code, "message": self.message}
@classmethod
def no_handler(cls) -> CanvasError:
"""Default error returned when a custom action has no handler."""
return cls(
"canvas_action_no_handler",
"No handler implemented for this canvas action",
)
@classmethod
def handler_unset(cls) -> CanvasError:
"""Error returned when a canvas RPC arrives but no handler is installed."""
return cls(
"canvas_handler_unset",
"No CanvasHandler installed on this session; "
"install one via SessionConfig.canvas_handler before creating the session.",
)
class CanvasHandler(ABC):
"""Provider-side canvas lifecycle handler.
.. note::
**Experimental.** This type is part of an experimental wire-protocol
surface and may change or be removed in future SDK or CLI releases.
"""
@abstractmethod
async def on_open(self, ctx: CanvasProviderOpenRequest) -> CanvasProviderOpenResult:
"""Open a new canvas instance.
May raise :class:`CanvasError` to surface a structured failure to
the host.
"""
async def on_close(self, ctx: CanvasProviderCloseRequest) -> None:
"""Canvas was closed by the user or agent. Default: no-op."""
async def on_action(self, ctx: CanvasProviderInvokeActionRequest) -> Any:
"""Handle a non-lifecycle action declared by the canvas."""
raise CanvasError.no_handler()