From 70c0147c878c2a53b472e26e8864f6e3f9d7f4b6 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Fri, 28 Nov 2025 10:41:32 -0500 Subject: [PATCH 1/8] feat: add clipboard services Add clipboard (memory) service with sync and async support: - ClipboardService (sync) and AsyncClipboardService - Named access: set, get (by name), delete - Indexed access: push, pop, get (by index) - List all entries, clear_all - Full Pydantic model validation Types: - ClipboardEntry, ClipboardSetRequest, ClipboardPushRequest - ClipboardGetFilters, ClipboardDeleteRequest - ClipboardResponse, ClipboardDeleteResponse, ClipboardListResponse - ClipboardEncoding and ClipboardVisibility enums --- src/pluggedinkit/__init__.py | 24 ++ src/pluggedinkit/client.py | 3 + src/pluggedinkit/services/__init__.py | 3 + src/pluggedinkit/services/clipboard.py | 334 +++++++++++++++++++++++++ src/pluggedinkit/types.py | 96 +++++++ 5 files changed, 460 insertions(+) create mode 100644 src/pluggedinkit/services/clipboard.py diff --git a/src/pluggedinkit/__init__.py b/src/pluggedinkit/__init__.py index 074966e..7c88733 100644 --- a/src/pluggedinkit/__init__.py +++ b/src/pluggedinkit/__init__.py @@ -15,6 +15,18 @@ ValidationError, ) from .types import ( + # Clipboard types + ClipboardEntry, + ClipboardListResponse, + ClipboardSetRequest, + ClipboardPushRequest, + ClipboardGetFilters, + ClipboardDeleteRequest, + ClipboardResponse, + ClipboardDeleteResponse, + ClipboardEncoding, + ClipboardVisibility, + # Document types Document, DocumentWithContent, DocumentListResponse, @@ -38,6 +50,18 @@ "RateLimitError", "NotFoundError", "ValidationError", + # Clipboard types + "ClipboardEntry", + "ClipboardListResponse", + "ClipboardSetRequest", + "ClipboardPushRequest", + "ClipboardGetFilters", + "ClipboardDeleteRequest", + "ClipboardResponse", + "ClipboardDeleteResponse", + "ClipboardEncoding", + "ClipboardVisibility", + # Document types "Document", "DocumentWithContent", "DocumentListResponse", diff --git a/src/pluggedinkit/client.py b/src/pluggedinkit/client.py index 549a72d..c99e11a 100644 --- a/src/pluggedinkit/client.py +++ b/src/pluggedinkit/client.py @@ -12,6 +12,7 @@ PluggedInError, RateLimitError, ) +from .services.clipboard import ClipboardService, AsyncClipboardService from .services.documents import DocumentService, AsyncDocumentService from .services.rag import RagService, AsyncRagService from .services.uploads import UploadService, AsyncUploadService @@ -96,6 +97,7 @@ def __init__( ) # Initialize services + self.clipboard = ClipboardService(self) self.documents = DocumentService(self) self.rag = RagService(self) self.uploads = UploadService(self) @@ -195,6 +197,7 @@ def __init__( ) # Initialize async services + self.clipboard = AsyncClipboardService(self) self.documents = AsyncDocumentService(self) self.rag = AsyncRagService(self) self.uploads = AsyncUploadService(self) diff --git a/src/pluggedinkit/services/__init__.py b/src/pluggedinkit/services/__init__.py index be59a48..5b85d8c 100644 --- a/src/pluggedinkit/services/__init__.py +++ b/src/pluggedinkit/services/__init__.py @@ -1,10 +1,13 @@ """Service modules for Plugged.in SDK""" +from .clipboard import AsyncClipboardService, ClipboardService from .documents import AsyncDocumentService, DocumentService from .rag import AsyncRagService, RagService from .uploads import AsyncUploadService, UploadService __all__ = [ + "ClipboardService", + "AsyncClipboardService", "DocumentService", "AsyncDocumentService", "RagService", diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py new file mode 100644 index 0000000..bf8d1ed --- /dev/null +++ b/src/pluggedinkit/services/clipboard.py @@ -0,0 +1,334 @@ +"""Clipboard service for Plugged.in SDK""" + +from typing import TYPE_CHECKING, List, Optional, Union + +from ..exceptions import PluggedInError +from ..types import ( + ClipboardDeleteRequest, + ClipboardDeleteResponse, + ClipboardEntry, + ClipboardGetFilters, + ClipboardListResponse, + ClipboardPushRequest, + ClipboardResponse, + ClipboardSetRequest, +) + +if TYPE_CHECKING: + from ..client import AsyncPluggedInClient, PluggedInClient + + +class ClipboardService: + """Synchronous clipboard service""" + + def __init__(self, client: "PluggedInClient"): + self.client = client + + def list(self) -> List[ClipboardEntry]: + """List all clipboard entries""" + response = self.client.request("GET", "/api/clipboard") + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to list clipboard entries")) + + list_response = ClipboardListResponse(**data) + return list_response.entries + + def get( + self, + name: Optional[str] = None, + idx: Optional[int] = None, + ) -> Optional[ClipboardEntry]: + """Get a clipboard entry by name or index""" + if name is None and idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + + params = {} + if name is not None: + params["name"] = name + if idx is not None: + params["idx"] = str(idx) + + response = self.client.request("GET", "/api/clipboard", params=params) + data = response.json() + + if not data.get("success"): + return None + + if "entry" in data and data["entry"]: + return ClipboardEntry(**data["entry"]) + return None + + def set( + self, + name: str, + value: str, + content_type: str = "text/plain", + encoding: str = "utf-8", + visibility: str = "private", + created_by_tool: Optional[str] = None, + created_by_model: Optional[str] = None, + ttl_seconds: Optional[int] = None, + ) -> ClipboardEntry: + """Set a named clipboard entry (upsert)""" + payload = { + "name": name, + "value": value, + "contentType": content_type, + "encoding": encoding, + "visibility": visibility, + } + + if created_by_tool: + payload["createdByTool"] = created_by_tool + if created_by_model: + payload["createdByModel"] = created_by_model + if ttl_seconds: + payload["ttlSeconds"] = ttl_seconds + + response = self.client.request("POST", "/api/clipboard", json=payload) + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to set clipboard entry")) + + return ClipboardEntry(**data["entry"]) + + def push( + self, + value: str, + content_type: str = "text/plain", + encoding: str = "utf-8", + visibility: str = "private", + created_by_tool: Optional[str] = None, + created_by_model: Optional[str] = None, + ttl_seconds: Optional[int] = None, + ) -> ClipboardEntry: + """Push a value to the indexed clipboard (auto-increment index)""" + payload = { + "value": value, + "contentType": content_type, + "encoding": encoding, + "visibility": visibility, + } + + if created_by_tool: + payload["createdByTool"] = created_by_tool + if created_by_model: + payload["createdByModel"] = created_by_model + if ttl_seconds: + payload["ttlSeconds"] = ttl_seconds + + response = self.client.request("POST", "/api/clipboard/push", json=payload) + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to push to clipboard")) + + return ClipboardEntry(**data["entry"]) + + def pop(self) -> Optional[ClipboardEntry]: + """Pop the last indexed entry from clipboard""" + response = self.client.request("POST", "/api/clipboard/pop") + data = response.json() + + if not data.get("success") or not data.get("entry"): + return None + + return ClipboardEntry(**data["entry"]) + + def delete( + self, + name: Optional[str] = None, + idx: Optional[int] = None, + ) -> bool: + """Delete a clipboard entry by name or index""" + if name is None and idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + + payload = {} + if name is not None: + payload["name"] = name + if idx is not None: + payload["idx"] = idx + + response = self.client.request("DELETE", "/api/clipboard", json=payload) + data = response.json() + + return data.get("success", False) and data.get("deleted", False) + + def clear_all(self) -> int: + """Clear all clipboard entries. Returns count of deleted entries.""" + entries = self.list() + deleted = 0 + + for entry in entries: + try: + if entry.name: + self.delete(name=entry.name) + elif entry.idx is not None: + self.delete(idx=entry.idx) + deleted += 1 + except PluggedInError: + pass + + return deleted + + +class AsyncClipboardService: + """Asynchronous clipboard service""" + + def __init__(self, client: "AsyncPluggedInClient"): + self.client = client + + async def list(self) -> List[ClipboardEntry]: + """List all clipboard entries""" + response = await self.client.request("GET", "/api/clipboard") + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to list clipboard entries")) + + list_response = ClipboardListResponse(**data) + return list_response.entries + + async def get( + self, + name: Optional[str] = None, + idx: Optional[int] = None, + ) -> Optional[ClipboardEntry]: + """Get a clipboard entry by name or index""" + if name is None and idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + + params = {} + if name is not None: + params["name"] = name + if idx is not None: + params["idx"] = str(idx) + + response = await self.client.request("GET", "/api/clipboard", params=params) + data = response.json() + + if not data.get("success"): + return None + + if "entry" in data and data["entry"]: + return ClipboardEntry(**data["entry"]) + return None + + async def set( + self, + name: str, + value: str, + content_type: str = "text/plain", + encoding: str = "utf-8", + visibility: str = "private", + created_by_tool: Optional[str] = None, + created_by_model: Optional[str] = None, + ttl_seconds: Optional[int] = None, + ) -> ClipboardEntry: + """Set a named clipboard entry (upsert)""" + payload = { + "name": name, + "value": value, + "contentType": content_type, + "encoding": encoding, + "visibility": visibility, + } + + if created_by_tool: + payload["createdByTool"] = created_by_tool + if created_by_model: + payload["createdByModel"] = created_by_model + if ttl_seconds: + payload["ttlSeconds"] = ttl_seconds + + response = await self.client.request("POST", "/api/clipboard", json=payload) + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to set clipboard entry")) + + return ClipboardEntry(**data["entry"]) + + async def push( + self, + value: str, + content_type: str = "text/plain", + encoding: str = "utf-8", + visibility: str = "private", + created_by_tool: Optional[str] = None, + created_by_model: Optional[str] = None, + ttl_seconds: Optional[int] = None, + ) -> ClipboardEntry: + """Push a value to the indexed clipboard (auto-increment index)""" + payload = { + "value": value, + "contentType": content_type, + "encoding": encoding, + "visibility": visibility, + } + + if created_by_tool: + payload["createdByTool"] = created_by_tool + if created_by_model: + payload["createdByModel"] = created_by_model + if ttl_seconds: + payload["ttlSeconds"] = ttl_seconds + + response = await self.client.request("POST", "/api/clipboard/push", json=payload) + data = response.json() + + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to push to clipboard")) + + return ClipboardEntry(**data["entry"]) + + async def pop(self) -> Optional[ClipboardEntry]: + """Pop the last indexed entry from clipboard""" + response = await self.client.request("POST", "/api/clipboard/pop") + data = response.json() + + if not data.get("success") or not data.get("entry"): + return None + + return ClipboardEntry(**data["entry"]) + + async def delete( + self, + name: Optional[str] = None, + idx: Optional[int] = None, + ) -> bool: + """Delete a clipboard entry by name or index""" + if name is None and idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + + payload = {} + if name is not None: + payload["name"] = name + if idx is not None: + payload["idx"] = idx + + response = await self.client.request("DELETE", "/api/clipboard", json=payload) + data = response.json() + + return data.get("success", False) and data.get("deleted", False) + + async def clear_all(self) -> int: + """Clear all clipboard entries. Returns count of deleted entries.""" + entries = await self.list() + deleted = 0 + + for entry in entries: + try: + if entry.name: + await self.delete(name=entry.name) + elif entry.idx is not None: + await self.delete(idx=entry.idx) + deleted += 1 + except PluggedInError: + pass + + return deleted diff --git a/src/pluggedinkit/types.py b/src/pluggedinkit/types.py index 8cbb42d..bff3765 100644 --- a/src/pluggedinkit/types.py +++ b/src/pluggedinkit/types.py @@ -266,3 +266,99 @@ class RagStorageStats(BaseModel): vectors_count: Optional[int] = None embedding_dimension: Optional[int] = None is_estimate: Optional[bool] = None + + +# Clipboard types +class ClipboardEncoding(str, Enum): + """Clipboard content encoding""" + UTF8 = "utf-8" + BASE64 = "base64" + HEX = "hex" + + +class ClipboardVisibility(str, Enum): + """Clipboard visibility levels""" + PRIVATE = "private" + WORKSPACE = "workspace" + PUBLIC = "public" + + +class ClipboardEntry(BaseModel): + """Clipboard entry model""" + uuid: str + name: Optional[str] = None + idx: Optional[int] = None + value: str + content_type: str = Field(alias="contentType", default="text/plain") + encoding: ClipboardEncoding = ClipboardEncoding.UTF8 + size_bytes: int = Field(alias="sizeBytes") + visibility: ClipboardVisibility = ClipboardVisibility.PRIVATE + created_by_tool: Optional[str] = Field(None, alias="createdByTool") + created_by_model: Optional[str] = Field(None, alias="createdByModel") + created_at: datetime = Field(alias="createdAt") + updated_at: datetime = Field(alias="updatedAt") + expires_at: Optional[datetime] = Field(None, alias="expiresAt") + + class Config: + populate_by_name = True + + +class ClipboardListResponse(BaseModel): + """Response for clipboard listing""" + success: bool + entries: List[ClipboardEntry] + + +class ClipboardSetRequest(BaseModel): + """Request for setting a named clipboard entry""" + name: str = Field(..., min_length=1, max_length=255) + value: str + content_type: Optional[str] = Field(None, alias="contentType", max_length=256) + encoding: Optional[ClipboardEncoding] = None + visibility: Optional[ClipboardVisibility] = None + created_by_tool: Optional[str] = Field(None, alias="createdByTool", max_length=255) + created_by_model: Optional[str] = Field(None, alias="createdByModel", max_length=255) + ttl_seconds: Optional[int] = Field(None, alias="ttlSeconds", gt=0) + + class Config: + populate_by_name = True + + +class ClipboardPushRequest(BaseModel): + """Request for pushing to indexed clipboard""" + value: str + content_type: Optional[str] = Field(None, alias="contentType", max_length=256) + encoding: Optional[ClipboardEncoding] = None + visibility: Optional[ClipboardVisibility] = None + created_by_tool: Optional[str] = Field(None, alias="createdByTool", max_length=255) + created_by_model: Optional[str] = Field(None, alias="createdByModel", max_length=255) + ttl_seconds: Optional[int] = Field(None, alias="ttlSeconds", gt=0) + + class Config: + populate_by_name = True + + +class ClipboardGetFilters(BaseModel): + """Filters for getting a clipboard entry""" + name: Optional[str] = None + idx: Optional[int] = None + + +class ClipboardDeleteRequest(BaseModel): + """Request for deleting a clipboard entry""" + name: Optional[str] = None + idx: Optional[int] = None + + +class ClipboardResponse(BaseModel): + """Response for clipboard operations""" + success: bool + entry: Optional[ClipboardEntry] = None + error: Optional[str] = None + + +class ClipboardDeleteResponse(BaseModel): + """Response for clipboard delete operations""" + success: bool + deleted: Optional[bool] = None + error: Optional[str] = None From f48b86bafe96cec5fc8139f3d873e0423ea12610 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 08:09:41 -0500 Subject: [PATCH 2/8] feat: Add clipboard source tracking - Add ClipboardSource enum (ui/sdk/mcp) - Add source field to ClipboardEntry model - Hardcode source as 'sdk' in sync/async set/push methods --- src/pluggedinkit/__init__.py | 2 ++ src/pluggedinkit/services/clipboard.py | 4 ++++ src/pluggedinkit/types.py | 8 ++++++++ 3 files changed, 14 insertions(+) diff --git a/src/pluggedinkit/__init__.py b/src/pluggedinkit/__init__.py index 7c88733..a9e4633 100644 --- a/src/pluggedinkit/__init__.py +++ b/src/pluggedinkit/__init__.py @@ -26,6 +26,7 @@ ClipboardDeleteResponse, ClipboardEncoding, ClipboardVisibility, + ClipboardSource, # Document types Document, DocumentWithContent, @@ -61,6 +62,7 @@ "ClipboardDeleteResponse", "ClipboardEncoding", "ClipboardVisibility", + "ClipboardSource", # Document types "Document", "DocumentWithContent", diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py index bf8d1ed..9c4306b 100644 --- a/src/pluggedinkit/services/clipboard.py +++ b/src/pluggedinkit/services/clipboard.py @@ -78,6 +78,7 @@ def set( "contentType": content_type, "encoding": encoding, "visibility": visibility, + "source": "sdk", # Hardcoded: SDK always uses 'sdk' source } if created_by_tool: @@ -111,6 +112,7 @@ def push( "contentType": content_type, "encoding": encoding, "visibility": visibility, + "source": "sdk", # Hardcoded: SDK always uses 'sdk' source } if created_by_tool: @@ -236,6 +238,7 @@ async def set( "contentType": content_type, "encoding": encoding, "visibility": visibility, + "source": "sdk", # Hardcoded: SDK always uses 'sdk' source } if created_by_tool: @@ -269,6 +272,7 @@ async def push( "contentType": content_type, "encoding": encoding, "visibility": visibility, + "source": "sdk", # Hardcoded: SDK always uses 'sdk' source } if created_by_tool: diff --git a/src/pluggedinkit/types.py b/src/pluggedinkit/types.py index bff3765..78fa184 100644 --- a/src/pluggedinkit/types.py +++ b/src/pluggedinkit/types.py @@ -283,6 +283,13 @@ class ClipboardVisibility(str, Enum): PUBLIC = "public" +class ClipboardSource(str, Enum): + """Clipboard data source""" + UI = "ui" + SDK = "sdk" + MCP = "mcp" + + class ClipboardEntry(BaseModel): """Clipboard entry model""" uuid: str @@ -295,6 +302,7 @@ class ClipboardEntry(BaseModel): visibility: ClipboardVisibility = ClipboardVisibility.PRIVATE created_by_tool: Optional[str] = Field(None, alias="createdByTool") created_by_model: Optional[str] = Field(None, alias="createdByModel") + source: ClipboardSource = ClipboardSource.UI created_at: datetime = Field(alias="createdAt") updated_at: datetime = Field(alias="updatedAt") expires_at: Optional[datetime] = Field(None, alias="expiresAt") From 783f12a944ec5a017a918292e7820d9d86e336e5 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 08:29:03 -0500 Subject: [PATCH 3/8] refactor: Make source field optional with DEFAULT_CLIPBOARD_SOURCE - Add DEFAULT_CLIPBOARD_SOURCE constant for backward compatibility - Make ClipboardEntry.source Optional to handle older API responses - Export constant from __init__.py Addresses code review feedback about breaking existing call sites. --- src/pluggedinkit/__init__.py | 2 ++ src/pluggedinkit/types.py | 7 ++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/pluggedinkit/__init__.py b/src/pluggedinkit/__init__.py index a9e4633..f4c0650 100644 --- a/src/pluggedinkit/__init__.py +++ b/src/pluggedinkit/__init__.py @@ -27,6 +27,7 @@ ClipboardEncoding, ClipboardVisibility, ClipboardSource, + DEFAULT_CLIPBOARD_SOURCE, # Document types Document, DocumentWithContent, @@ -63,6 +64,7 @@ "ClipboardEncoding", "ClipboardVisibility", "ClipboardSource", + "DEFAULT_CLIPBOARD_SOURCE", # Document types "Document", "DocumentWithContent", diff --git a/src/pluggedinkit/types.py b/src/pluggedinkit/types.py index 78fa184..f6bfed4 100644 --- a/src/pluggedinkit/types.py +++ b/src/pluggedinkit/types.py @@ -290,6 +290,10 @@ class ClipboardSource(str, Enum): MCP = "mcp" +# Default source for backward compatibility with older data +DEFAULT_CLIPBOARD_SOURCE = ClipboardSource.UI + + class ClipboardEntry(BaseModel): """Clipboard entry model""" uuid: str @@ -302,7 +306,8 @@ class ClipboardEntry(BaseModel): visibility: ClipboardVisibility = ClipboardVisibility.PRIVATE created_by_tool: Optional[str] = Field(None, alias="createdByTool") created_by_model: Optional[str] = Field(None, alias="createdByModel") - source: ClipboardSource = ClipboardSource.UI + # Optional for backward compatibility with older API responses + source: Optional[ClipboardSource] = DEFAULT_CLIPBOARD_SOURCE created_at: datetime = Field(alias="createdAt") updated_at: datetime = Field(alias="updatedAt") expires_at: Optional[datetime] = Field(None, alias="expiresAt") From 17c7b5f7343b83274d23cb439c15cc895bb6645a Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 08:40:10 -0500 Subject: [PATCH 4/8] docs: Add clipboard operations with source field documentation - Add Clipboard Operations section to README - Document set, get, push, pop, list, and delete operations - Add ClipboardEntry structure with source field - Note that source is auto-set to ClipboardSource.SDK when using this SDK - Add Clipboard/Memory to Features list --- README.md | 91 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/README.md b/README.md index 088858a..71ac9b7 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,7 @@ asyncio.run(main()) - 📄 **Document Management** - Full CRUD operations with type safety - 🔍 **Semantic Search** - AI-powered document search - 🤖 **RAG Integration** - Natural language queries to your knowledge base +- 📋 **Clipboard/Memory** - Persistent key-value storage for MCP tools and AI agents - 📤 **File Uploads** - Support for various file formats - 🔄 **Version Control** - Document versioning and history - ⚡ **Async Support** - Both sync and async clients @@ -377,6 +378,96 @@ async def rag_operations(): asyncio.run(rag_operations()) ``` +### Clipboard Operations + +The clipboard provides persistent key-value storage for MCP tools and AI agents. + +#### Set Named Entry + +```python +entry = client.clipboard.set( + name="user_preferences", + value='{"theme": "dark", "lang": "en"}', + content_type="application/json", + encoding="utf-8", + visibility="private", + ttl_seconds=86400 # 24 hours +) + +print(f"Created entry: {entry.uuid}") +print(f"Source: {entry.source}") # 'sdk' - automatically set +``` + +#### Get Entry + +```python +# By name +entry = client.clipboard.get(name="user_preferences") +print(entry.value) + +# By index (stack access) +latest = client.clipboard.get(idx=0) +``` + +#### Push to Stack + +```python +entry = client.clipboard.push( + value="Processing step 1 result", + content_type="text/plain" +) + +print(f"Pushed to index: {entry.idx}") +``` + +#### Pop from Stack + +```python +entry = client.clipboard.pop() +if entry: + print(f"Popped value: {entry.value}") +``` + +#### List and Delete + +```python +# List all entries +response = client.clipboard.list() +for entry in response.entries: + label = entry.name or f"idx:{entry.idx}" + print(f"{label} - source: {entry.source}") + +# Delete by name +client.clipboard.delete(name="old_entry") + +# Clear all +client.clipboard.clear_all() +``` + +#### Clipboard Entry Structure + +```python +from pluggedinkit import ClipboardEntry, ClipboardSource + +# ClipboardEntry fields: +# - uuid: str +# - name: Optional[str] # Semantic key +# - idx: Optional[int] # Stack index +# - value: str +# - content_type: str +# - encoding: "utf-8" | "base64" | "hex" +# - size_bytes: int +# - visibility: "private" | "workspace" | "public" +# - created_by_tool: Optional[str] +# - created_by_model: Optional[str] +# - source: Optional[ClipboardSource] # Auto-set: ui, sdk, or mcp +# - created_at: datetime +# - updated_at: datetime +# - expires_at: Optional[datetime] +``` + +> **Note**: The `source` field is automatically set to `ClipboardSource.SDK` when using this SDK. It indicates how the entry was created (UI, SDK, or MCP proxy). + ### Error Handling ```python From 029ad0489a0244a5c27e8ccb1429d5115a23ac5e Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 09:38:00 -0500 Subject: [PATCH 5/8] fix: Improve ttl_seconds validation and clear_all counter logic - Use 'is not None' check for ttl_seconds instead of truthiness - Add explicit validation that ttl_seconds > 0 when provided - Fix clear_all to only increment counter when delete() returns True - Applied to both sync and async service classes (4 locations each) --- src/pluggedinkit/services/clipboard.py | 32 ++++++++++++++++++-------- 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py index 9c4306b..10f9119 100644 --- a/src/pluggedinkit/services/clipboard.py +++ b/src/pluggedinkit/services/clipboard.py @@ -85,7 +85,9 @@ def set( payload["createdByTool"] = created_by_tool if created_by_model: payload["createdByModel"] = created_by_model - if ttl_seconds: + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be greater than 0 when provided") payload["ttlSeconds"] = ttl_seconds response = self.client.request("POST", "/api/clipboard", json=payload) @@ -119,7 +121,9 @@ def push( payload["createdByTool"] = created_by_tool if created_by_model: payload["createdByModel"] = created_by_model - if ttl_seconds: + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be greater than 0 when provided") payload["ttlSeconds"] = ttl_seconds response = self.client.request("POST", "/api/clipboard/push", json=payload) @@ -167,11 +171,13 @@ def clear_all(self) -> int: for entry in entries: try: + success = False if entry.name: - self.delete(name=entry.name) + success = self.delete(name=entry.name) elif entry.idx is not None: - self.delete(idx=entry.idx) - deleted += 1 + success = self.delete(idx=entry.idx) + if success: + deleted += 1 except PluggedInError: pass @@ -245,7 +251,9 @@ async def set( payload["createdByTool"] = created_by_tool if created_by_model: payload["createdByModel"] = created_by_model - if ttl_seconds: + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be greater than 0 when provided") payload["ttlSeconds"] = ttl_seconds response = await self.client.request("POST", "/api/clipboard", json=payload) @@ -279,7 +287,9 @@ async def push( payload["createdByTool"] = created_by_tool if created_by_model: payload["createdByModel"] = created_by_model - if ttl_seconds: + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be greater than 0 when provided") payload["ttlSeconds"] = ttl_seconds response = await self.client.request("POST", "/api/clipboard/push", json=payload) @@ -327,11 +337,13 @@ async def clear_all(self) -> int: for entry in entries: try: + success = False if entry.name: - await self.delete(name=entry.name) + success = await self.delete(name=entry.name) elif entry.idx is not None: - await self.delete(idx=entry.idx) - deleted += 1 + success = await self.delete(idx=entry.idx) + if success: + deleted += 1 except PluggedInError: pass From e996090e01203637b60447185aaff1a7d388dc35 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 10:02:34 -0500 Subject: [PATCH 6/8] refactor: Use ClipboardSource enum instead of string literal - Import ClipboardSource from types - Replace hardcoded "sdk" string with ClipboardSource.SDK.value - Consistent with enum usage across the SDK --- src/pluggedinkit/services/clipboard.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py index 10f9119..610d3cd 100644 --- a/src/pluggedinkit/services/clipboard.py +++ b/src/pluggedinkit/services/clipboard.py @@ -12,6 +12,7 @@ ClipboardPushRequest, ClipboardResponse, ClipboardSetRequest, + ClipboardSource, ) if TYPE_CHECKING: @@ -78,7 +79,7 @@ def set( "contentType": content_type, "encoding": encoding, "visibility": visibility, - "source": "sdk", # Hardcoded: SDK always uses 'sdk' source + "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source } if created_by_tool: @@ -114,7 +115,7 @@ def push( "contentType": content_type, "encoding": encoding, "visibility": visibility, - "source": "sdk", # Hardcoded: SDK always uses 'sdk' source + "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source } if created_by_tool: @@ -244,7 +245,7 @@ async def set( "contentType": content_type, "encoding": encoding, "visibility": visibility, - "source": "sdk", # Hardcoded: SDK always uses 'sdk' source + "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source } if created_by_tool: @@ -280,7 +281,7 @@ async def push( "contentType": content_type, "encoding": encoding, "visibility": visibility, - "source": "sdk", # Hardcoded: SDK always uses 'sdk' source + "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source } if created_by_tool: From 7406cdd7009c3f56bdd64354f8d8c76cb00714d5 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 10:29:24 -0500 Subject: [PATCH 7/8] fix: Apply code review feedback for clipboard service - Add model_validator to ClipboardGetFilters for name/idx constraint - Add model_validator to ClipboardDeleteRequest for name/idx constraint - Use ClipboardEncoding enum for encoding parameter - Use ClipboardVisibility enum for visibility parameter - Extract _build_clipboard_payload helper to reduce sync/async duplication --- src/pluggedinkit/services/clipboard.py | 155 +++++++++++++------------ src/pluggedinkit/types.py | 14 ++- 2 files changed, 94 insertions(+), 75 deletions(-) diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py index 610d3cd..f28b2c5 100644 --- a/src/pluggedinkit/services/clipboard.py +++ b/src/pluggedinkit/services/clipboard.py @@ -6,6 +6,7 @@ from ..types import ( ClipboardDeleteRequest, ClipboardDeleteResponse, + ClipboardEncoding, ClipboardEntry, ClipboardGetFilters, ClipboardListResponse, @@ -13,12 +14,46 @@ ClipboardResponse, ClipboardSetRequest, ClipboardSource, + ClipboardVisibility, ) if TYPE_CHECKING: from ..client import AsyncPluggedInClient, PluggedInClient +def _build_clipboard_payload( + value: str, + name: Optional[str] = None, + content_type: str = "text/plain", + encoding: ClipboardEncoding = ClipboardEncoding.UTF8, + visibility: ClipboardVisibility = ClipboardVisibility.PRIVATE, + created_by_tool: Optional[str] = None, + created_by_model: Optional[str] = None, + ttl_seconds: Optional[int] = None, +) -> dict: + """Build a clipboard request payload with common fields.""" + payload = { + "value": value, + "contentType": content_type, + "encoding": encoding.value if isinstance(encoding, ClipboardEncoding) else encoding, + "visibility": visibility.value if isinstance(visibility, ClipboardVisibility) else visibility, + "source": ClipboardSource.SDK.value, + } + + if name: + payload["name"] = name + if created_by_tool: + payload["createdByTool"] = created_by_tool + if created_by_model: + payload["createdByModel"] = created_by_model + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be greater than 0 when provided") + payload["ttlSeconds"] = ttl_seconds + + return payload + + class ClipboardService: """Synchronous clipboard service""" @@ -66,30 +101,23 @@ def set( name: str, value: str, content_type: str = "text/plain", - encoding: str = "utf-8", - visibility: str = "private", + encoding: Union[ClipboardEncoding, str] = ClipboardEncoding.UTF8, + visibility: Union[ClipboardVisibility, str] = ClipboardVisibility.PRIVATE, created_by_tool: Optional[str] = None, created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: """Set a named clipboard entry (upsert)""" - payload = { - "name": name, - "value": value, - "contentType": content_type, - "encoding": encoding, - "visibility": visibility, - "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source - } - - if created_by_tool: - payload["createdByTool"] = created_by_tool - if created_by_model: - payload["createdByModel"] = created_by_model - if ttl_seconds is not None: - if ttl_seconds <= 0: - raise ValueError("ttl_seconds must be greater than 0 when provided") - payload["ttlSeconds"] = ttl_seconds + payload = _build_clipboard_payload( + value=value, + name=name, + content_type=content_type, + encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), + visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + created_by_tool=created_by_tool, + created_by_model=created_by_model, + ttl_seconds=ttl_seconds, + ) response = self.client.request("POST", "/api/clipboard", json=payload) data = response.json() @@ -103,29 +131,22 @@ def push( self, value: str, content_type: str = "text/plain", - encoding: str = "utf-8", - visibility: str = "private", + encoding: Union[ClipboardEncoding, str] = ClipboardEncoding.UTF8, + visibility: Union[ClipboardVisibility, str] = ClipboardVisibility.PRIVATE, created_by_tool: Optional[str] = None, created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: """Push a value to the indexed clipboard (auto-increment index)""" - payload = { - "value": value, - "contentType": content_type, - "encoding": encoding, - "visibility": visibility, - "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source - } - - if created_by_tool: - payload["createdByTool"] = created_by_tool - if created_by_model: - payload["createdByModel"] = created_by_model - if ttl_seconds is not None: - if ttl_seconds <= 0: - raise ValueError("ttl_seconds must be greater than 0 when provided") - payload["ttlSeconds"] = ttl_seconds + payload = _build_clipboard_payload( + value=value, + content_type=content_type, + encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), + visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + created_by_tool=created_by_tool, + created_by_model=created_by_model, + ttl_seconds=ttl_seconds, + ) response = self.client.request("POST", "/api/clipboard/push", json=payload) data = response.json() @@ -232,30 +253,23 @@ async def set( name: str, value: str, content_type: str = "text/plain", - encoding: str = "utf-8", - visibility: str = "private", + encoding: Union[ClipboardEncoding, str] = ClipboardEncoding.UTF8, + visibility: Union[ClipboardVisibility, str] = ClipboardVisibility.PRIVATE, created_by_tool: Optional[str] = None, created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: """Set a named clipboard entry (upsert)""" - payload = { - "name": name, - "value": value, - "contentType": content_type, - "encoding": encoding, - "visibility": visibility, - "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source - } - - if created_by_tool: - payload["createdByTool"] = created_by_tool - if created_by_model: - payload["createdByModel"] = created_by_model - if ttl_seconds is not None: - if ttl_seconds <= 0: - raise ValueError("ttl_seconds must be greater than 0 when provided") - payload["ttlSeconds"] = ttl_seconds + payload = _build_clipboard_payload( + value=value, + name=name, + content_type=content_type, + encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), + visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + created_by_tool=created_by_tool, + created_by_model=created_by_model, + ttl_seconds=ttl_seconds, + ) response = await self.client.request("POST", "/api/clipboard", json=payload) data = response.json() @@ -269,29 +283,22 @@ async def push( self, value: str, content_type: str = "text/plain", - encoding: str = "utf-8", - visibility: str = "private", + encoding: Union[ClipboardEncoding, str] = ClipboardEncoding.UTF8, + visibility: Union[ClipboardVisibility, str] = ClipboardVisibility.PRIVATE, created_by_tool: Optional[str] = None, created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: """Push a value to the indexed clipboard (auto-increment index)""" - payload = { - "value": value, - "contentType": content_type, - "encoding": encoding, - "visibility": visibility, - "source": ClipboardSource.SDK.value, # SDK always uses 'sdk' source - } - - if created_by_tool: - payload["createdByTool"] = created_by_tool - if created_by_model: - payload["createdByModel"] = created_by_model - if ttl_seconds is not None: - if ttl_seconds <= 0: - raise ValueError("ttl_seconds must be greater than 0 when provided") - payload["ttlSeconds"] = ttl_seconds + payload = _build_clipboard_payload( + value=value, + content_type=content_type, + encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), + visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + created_by_tool=created_by_tool, + created_by_model=created_by_model, + ttl_seconds=ttl_seconds, + ) response = await self.client.request("POST", "/api/clipboard/push", json=payload) data = response.json() diff --git a/src/pluggedinkit/types.py b/src/pluggedinkit/types.py index f6bfed4..aaaf352 100644 --- a/src/pluggedinkit/types.py +++ b/src/pluggedinkit/types.py @@ -4,7 +4,7 @@ from enum import Enum from typing import Any, Dict, List, Literal, Optional, Union -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator class DocumentSource(str, Enum): @@ -356,12 +356,24 @@ class ClipboardGetFilters(BaseModel): name: Optional[str] = None idx: Optional[int] = None + @model_validator(mode='after') + def check_name_or_idx(self) -> 'ClipboardGetFilters': + if self.name is None and self.idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + return self + class ClipboardDeleteRequest(BaseModel): """Request for deleting a clipboard entry""" name: Optional[str] = None idx: Optional[int] = None + @model_validator(mode='after') + def check_name_or_idx(self) -> 'ClipboardDeleteRequest': + if self.name is None and self.idx is None: + raise ValueError("Either 'name' or 'idx' must be provided") + return self + class ClipboardResponse(BaseModel): """Response for clipboard operations""" From 6f2ef98342196a51abe3ce2837c737777561d125 Mon Sep 17 00:00:00 2001 From: Cem Karaca Date: Sat, 29 Nov 2025 15:53:00 -0500 Subject: [PATCH 8/8] refactor: Address code review feedback for clipboard service - Consolidate enum normalization to _build_clipboard_payload only - Extract shared helper functions (_parse_list_response, _parse_entry_response, etc.) - Use Pydantic models (ClipboardGetFilters, ClipboardDeleteRequest) for validation - Fix clear_all to use bulk delete API with clearAll=true instead of one-by-one - Add ClearAllResult dataclass for structured feedback on bulk operations - Add explicit type annotations (Dict[str, Any]) to fix mypy errors - Change delete() return type from bool to int (deleted count) - Add consistent error handling with raise_on_failure parameter Key improvements: - Single place for enum normalization (no more redundant checks) - Shared helpers reduce duplication between sync/async services - Pydantic validators handle "at least one of name/idx" checks - Bulk delete is efficient for large clipboards - ClearAllResult provides deleted/failed counts and success property --- src/pluggedinkit/__init__.py | 2 + src/pluggedinkit/services/__init__.py | 3 +- src/pluggedinkit/services/clipboard.py | 575 ++++++++++++++++++------- 3 files changed, 415 insertions(+), 165 deletions(-) diff --git a/src/pluggedinkit/__init__.py b/src/pluggedinkit/__init__.py index f4c0650..6e5e6c8 100644 --- a/src/pluggedinkit/__init__.py +++ b/src/pluggedinkit/__init__.py @@ -14,6 +14,7 @@ NotFoundError, ValidationError, ) +from .services.clipboard import ClearAllResult from .types import ( # Clipboard types ClipboardEntry, @@ -64,6 +65,7 @@ "ClipboardEncoding", "ClipboardVisibility", "ClipboardSource", + "ClearAllResult", "DEFAULT_CLIPBOARD_SOURCE", # Document types "Document", diff --git a/src/pluggedinkit/services/__init__.py b/src/pluggedinkit/services/__init__.py index 5b85d8c..cff8617 100644 --- a/src/pluggedinkit/services/__init__.py +++ b/src/pluggedinkit/services/__init__.py @@ -1,6 +1,6 @@ """Service modules for Plugged.in SDK""" -from .clipboard import AsyncClipboardService, ClipboardService +from .clipboard import AsyncClipboardService, ClearAllResult, ClipboardService from .documents import AsyncDocumentService, DocumentService from .rag import AsyncRagService, RagService from .uploads import AsyncUploadService, UploadService @@ -8,6 +8,7 @@ __all__ = [ "ClipboardService", "AsyncClipboardService", + "ClearAllResult", "DocumentService", "AsyncDocumentService", "RagService", diff --git a/src/pluggedinkit/services/clipboard.py b/src/pluggedinkit/services/clipboard.py index f28b2c5..134a89a 100644 --- a/src/pluggedinkit/services/clipboard.py +++ b/src/pluggedinkit/services/clipboard.py @@ -1,18 +1,19 @@ -"""Clipboard service for Plugged.in SDK""" +"""Clipboard service for Plugged.in SDK -from typing import TYPE_CHECKING, List, Optional, Union +This module provides both synchronous and asynchronous clipboard services +for persistent key-value and stack-based storage in AI workflows. +""" + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union from ..exceptions import PluggedInError from ..types import ( ClipboardDeleteRequest, - ClipboardDeleteResponse, ClipboardEncoding, ClipboardEntry, ClipboardGetFilters, ClipboardListResponse, - ClipboardPushRequest, - ClipboardResponse, - ClipboardSetRequest, ClipboardSource, ClipboardVisibility, ) @@ -21,22 +22,52 @@ from ..client import AsyncPluggedInClient, PluggedInClient +# ----------------------------------------------------------------------------- +# Shared Helper Functions +# ----------------------------------------------------------------------------- + + +def _normalize_encoding( + encoding: Union[ClipboardEncoding, str] +) -> ClipboardEncoding: + """Normalize encoding to ClipboardEncoding enum.""" + if isinstance(encoding, ClipboardEncoding): + return encoding + return ClipboardEncoding(encoding) + + +def _normalize_visibility( + visibility: Union[ClipboardVisibility, str] +) -> ClipboardVisibility: + """Normalize visibility to ClipboardVisibility enum.""" + if isinstance(visibility, ClipboardVisibility): + return visibility + return ClipboardVisibility(visibility) + + def _build_clipboard_payload( value: str, name: Optional[str] = None, content_type: str = "text/plain", - encoding: ClipboardEncoding = ClipboardEncoding.UTF8, - visibility: ClipboardVisibility = ClipboardVisibility.PRIVATE, + encoding: Union[ClipboardEncoding, str] = ClipboardEncoding.UTF8, + visibility: Union[ClipboardVisibility, str] = ClipboardVisibility.PRIVATE, created_by_tool: Optional[str] = None, created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, -) -> dict: - """Build a clipboard request payload with common fields.""" - payload = { +) -> Dict[str, Any]: + """Build a clipboard request payload with common fields. + + Normalizes encoding and visibility enums in one place. + """ + # Normalize enums here - callers don't need to + normalized_encoding = _normalize_encoding(encoding) + normalized_visibility = _normalize_visibility(visibility) + + payload: Dict[str, Any] = { "value": value, "contentType": content_type, - "encoding": encoding.value if isinstance(encoding, ClipboardEncoding) else encoding, - "visibility": visibility.value if isinstance(visibility, ClipboardVisibility) else visibility, + "encoding": normalized_encoding.value, + "visibility": normalized_visibility.value, "source": ClipboardSource.SDK.value, } @@ -54,47 +85,158 @@ def _build_clipboard_payload( return payload +def _parse_list_response(data: Dict[str, Any]) -> List[ClipboardEntry]: + """Parse list response and raise on failure.""" + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to list clipboard entries")) + return ClipboardListResponse(**data).entries + + +def _parse_entry_response( + data: Dict[str, Any], + error_msg: str = "Operation failed", + raise_on_failure: bool = True, +) -> Optional[ClipboardEntry]: + """Parse entry response with consistent error handling. + + Args: + data: API response data + error_msg: Error message to use if raising + raise_on_failure: If True, raises PluggedInError on failure. + If False, returns None on failure. + + Returns: + ClipboardEntry on success, None on failure (if not raising) + + Raises: + PluggedInError: If raise_on_failure is True and operation failed + """ + if not data.get("success"): + if raise_on_failure: + raise PluggedInError(data.get("error", error_msg)) + return None + + if "entry" in data and data["entry"]: + return ClipboardEntry(**data["entry"]) + + if raise_on_failure: + raise PluggedInError(data.get("error", error_msg)) + return None + + +def _parse_delete_response(data: Dict[str, Any]) -> int: + """Parse delete response and return deleted count. + + Returns: + Number of deleted entries (0 if failed) + """ + if not data.get("success"): + return 0 + deleted = data.get("deleted", 0) + return int(deleted) if deleted is not None else 0 + + +def _build_get_params(filters: ClipboardGetFilters) -> Dict[str, str]: + """Build query params from validated filters.""" + params: Dict[str, str] = {} + if filters.name is not None: + params["name"] = filters.name + if filters.idx is not None: + params["idx"] = str(filters.idx) + return params + + +def _build_delete_payload(request: ClipboardDeleteRequest) -> Dict[str, Any]: + """Build delete payload from validated request.""" + payload: Dict[str, Any] = {} + if request.name is not None: + payload["name"] = request.name + if request.idx is not None: + payload["idx"] = request.idx + return payload + + +# ----------------------------------------------------------------------------- +# Result Types +# ----------------------------------------------------------------------------- + + +@dataclass +class ClearAllResult: + """Result of clear_all operation with structured feedback. + + Attributes: + deleted: Number of successfully deleted entries + failed: Number of entries that failed to delete + total: Total entries attempted (deleted + failed) + """ + deleted: int + failed: int + + @property + def total(self) -> int: + """Total entries attempted.""" + return self.deleted + self.failed + + @property + def success(self) -> bool: + """True if all entries were deleted.""" + return self.failed == 0 + + +# ----------------------------------------------------------------------------- +# Synchronous Clipboard Service +# ----------------------------------------------------------------------------- + + class ClipboardService: - """Synchronous clipboard service""" + """Synchronous clipboard service for Plugged.in. + + Provides persistent key-value and stack-based storage for AI workflows. + """ def __init__(self, client: "PluggedInClient"): self.client = client def list(self) -> List[ClipboardEntry]: - """List all clipboard entries""" - response = self.client.request("GET", "/api/clipboard") - data = response.json() + """List all clipboard entries. - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to list clipboard entries")) + Returns: + List of ClipboardEntry objects - list_response = ClipboardListResponse(**data) - return list_response.entries + Raises: + PluggedInError: If the API request fails + """ + response = self.client.request("GET", "/api/clipboard") + return _parse_list_response(response.json()) def get( self, name: Optional[str] = None, idx: Optional[int] = None, ) -> Optional[ClipboardEntry]: - """Get a clipboard entry by name or index""" - if name is None and idx is None: - raise ValueError("Either 'name' or 'idx' must be provided") + """Get a clipboard entry by name or index. - params = {} - if name is not None: - params["name"] = name - if idx is not None: - params["idx"] = str(idx) + Args: + name: Entry name for key-value lookup + idx: Entry index for stack-based lookup - response = self.client.request("GET", "/api/clipboard", params=params) - data = response.json() + Returns: + ClipboardEntry if found, None if not found - if not data.get("success"): - return None + Raises: + ValueError: If neither name nor idx is provided + """ + # Use Pydantic model for validation + filters = ClipboardGetFilters(name=name, idx=idx) + params = _build_get_params(filters) - if "entry" in data and data["entry"]: - return ClipboardEntry(**data["entry"]) - return None + response = self.client.request("GET", "/api/clipboard", params=params) + return _parse_entry_response( + response.json(), + error_msg="Failed to get clipboard entry", + raise_on_failure=False, + ) def set( self, @@ -107,25 +249,45 @@ def set( created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: - """Set a named clipboard entry (upsert)""" + """Set a named clipboard entry (upsert). + + Args: + name: Entry name + value: Value to store + content_type: MIME type (default: text/plain) + encoding: Content encoding (default: utf-8) + visibility: Visibility level (default: private) + created_by_tool: Tool name for attribution + created_by_model: Model name for attribution + ttl_seconds: Time-to-live in seconds + + Returns: + The created/updated ClipboardEntry + + Raises: + PluggedInError: If the API request fails + ValueError: If ttl_seconds is <= 0 + """ payload = _build_clipboard_payload( value=value, name=name, content_type=content_type, - encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), - visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + encoding=encoding, + visibility=visibility, created_by_tool=created_by_tool, created_by_model=created_by_model, ttl_seconds=ttl_seconds, ) response = self.client.request("POST", "/api/clipboard", json=payload) - data = response.json() - - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to set clipboard entry")) - - return ClipboardEntry(**data["entry"]) + result = _parse_entry_response( + response.json(), + error_msg="Failed to set clipboard entry", + raise_on_failure=True, + ) + # result is guaranteed non-None when raise_on_failure=True + assert result is not None + return result def push( self, @@ -137,116 +299,155 @@ def push( created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: - """Push a value to the indexed clipboard (auto-increment index)""" + """Push a value to the indexed clipboard (auto-increment index). + + Args: + value: Value to store + content_type: MIME type (default: text/plain) + encoding: Content encoding (default: utf-8) + visibility: Visibility level (default: private) + created_by_tool: Tool name for attribution + created_by_model: Model name for attribution + ttl_seconds: Time-to-live in seconds + + Returns: + The created ClipboardEntry with assigned index + + Raises: + PluggedInError: If the API request fails + ValueError: If ttl_seconds is <= 0 + """ payload = _build_clipboard_payload( value=value, content_type=content_type, - encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), - visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + encoding=encoding, + visibility=visibility, created_by_tool=created_by_tool, created_by_model=created_by_model, ttl_seconds=ttl_seconds, ) response = self.client.request("POST", "/api/clipboard/push", json=payload) - data = response.json() - - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to push to clipboard")) - - return ClipboardEntry(**data["entry"]) + result = _parse_entry_response( + response.json(), + error_msg="Failed to push to clipboard", + raise_on_failure=True, + ) + assert result is not None + return result def pop(self) -> Optional[ClipboardEntry]: - """Pop the last indexed entry from clipboard""" - response = self.client.request("POST", "/api/clipboard/pop") - data = response.json() + """Pop the last indexed entry from clipboard (LIFO). - if not data.get("success") or not data.get("entry"): - return None - - return ClipboardEntry(**data["entry"]) + Returns: + The popped ClipboardEntry, or None if clipboard is empty + """ + response = self.client.request("POST", "/api/clipboard/pop") + return _parse_entry_response( + response.json(), + error_msg="Failed to pop from clipboard", + raise_on_failure=False, + ) def delete( self, name: Optional[str] = None, idx: Optional[int] = None, - ) -> bool: - """Delete a clipboard entry by name or index""" - if name is None and idx is None: - raise ValueError("Either 'name' or 'idx' must be provided") + ) -> int: + """Delete a clipboard entry by name or index. - payload = {} - if name is not None: - payload["name"] = name - if idx is not None: - payload["idx"] = idx + Args: + name: Entry name to delete + idx: Entry index to delete + + Returns: + Number of deleted entries (0 or 1) + + Raises: + ValueError: If neither name nor idx is provided + """ + # Use Pydantic model for validation + request = ClipboardDeleteRequest(name=name, idx=idx) + payload = _build_delete_payload(request) response = self.client.request("DELETE", "/api/clipboard", json=payload) + return _parse_delete_response(response.json()) + + def clear_all(self) -> ClearAllResult: + """Clear all clipboard entries using bulk delete API. + + Returns: + ClearAllResult with deleted count and status + + Raises: + PluggedInError: If the API request fails + """ + response = self.client.request( + "DELETE", + "/api/clipboard", + json={"clearAll": True} + ) data = response.json() - return data.get("success", False) and data.get("deleted", False) + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to clear clipboard")) - def clear_all(self) -> int: - """Clear all clipboard entries. Returns count of deleted entries.""" - entries = self.list() - deleted = 0 + deleted = data.get("deleted", 0) + return ClearAllResult(deleted=deleted, failed=0) - for entry in entries: - try: - success = False - if entry.name: - success = self.delete(name=entry.name) - elif entry.idx is not None: - success = self.delete(idx=entry.idx) - if success: - deleted += 1 - except PluggedInError: - pass - return deleted +# ----------------------------------------------------------------------------- +# Asynchronous Clipboard Service +# ----------------------------------------------------------------------------- class AsyncClipboardService: - """Asynchronous clipboard service""" + """Asynchronous clipboard service for Plugged.in. + + Provides persistent key-value and stack-based storage for AI workflows. + """ def __init__(self, client: "AsyncPluggedInClient"): self.client = client async def list(self) -> List[ClipboardEntry]: - """List all clipboard entries""" - response = await self.client.request("GET", "/api/clipboard") - data = response.json() + """List all clipboard entries. - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to list clipboard entries")) + Returns: + List of ClipboardEntry objects - list_response = ClipboardListResponse(**data) - return list_response.entries + Raises: + PluggedInError: If the API request fails + """ + response = await self.client.request("GET", "/api/clipboard") + return _parse_list_response(response.json()) async def get( self, name: Optional[str] = None, idx: Optional[int] = None, ) -> Optional[ClipboardEntry]: - """Get a clipboard entry by name or index""" - if name is None and idx is None: - raise ValueError("Either 'name' or 'idx' must be provided") + """Get a clipboard entry by name or index. - params = {} - if name is not None: - params["name"] = name - if idx is not None: - params["idx"] = str(idx) + Args: + name: Entry name for key-value lookup + idx: Entry index for stack-based lookup - response = await self.client.request("GET", "/api/clipboard", params=params) - data = response.json() + Returns: + ClipboardEntry if found, None if not found - if not data.get("success"): - return None + Raises: + ValueError: If neither name nor idx is provided + """ + filters = ClipboardGetFilters(name=name, idx=idx) + params = _build_get_params(filters) - if "entry" in data and data["entry"]: - return ClipboardEntry(**data["entry"]) - return None + response = await self.client.request("GET", "/api/clipboard", params=params) + return _parse_entry_response( + response.json(), + error_msg="Failed to get clipboard entry", + raise_on_failure=False, + ) async def set( self, @@ -259,25 +460,44 @@ async def set( created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: - """Set a named clipboard entry (upsert)""" + """Set a named clipboard entry (upsert). + + Args: + name: Entry name + value: Value to store + content_type: MIME type (default: text/plain) + encoding: Content encoding (default: utf-8) + visibility: Visibility level (default: private) + created_by_tool: Tool name for attribution + created_by_model: Model name for attribution + ttl_seconds: Time-to-live in seconds + + Returns: + The created/updated ClipboardEntry + + Raises: + PluggedInError: If the API request fails + ValueError: If ttl_seconds is <= 0 + """ payload = _build_clipboard_payload( value=value, name=name, content_type=content_type, - encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), - visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + encoding=encoding, + visibility=visibility, created_by_tool=created_by_tool, created_by_model=created_by_model, ttl_seconds=ttl_seconds, ) response = await self.client.request("POST", "/api/clipboard", json=payload) - data = response.json() - - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to set clipboard entry")) - - return ClipboardEntry(**data["entry"]) + result = _parse_entry_response( + response.json(), + error_msg="Failed to set clipboard entry", + raise_on_failure=True, + ) + assert result is not None + return result async def push( self, @@ -289,70 +509,97 @@ async def push( created_by_model: Optional[str] = None, ttl_seconds: Optional[int] = None, ) -> ClipboardEntry: - """Push a value to the indexed clipboard (auto-increment index)""" + """Push a value to the indexed clipboard (auto-increment index). + + Args: + value: Value to store + content_type: MIME type (default: text/plain) + encoding: Content encoding (default: utf-8) + visibility: Visibility level (default: private) + created_by_tool: Tool name for attribution + created_by_model: Model name for attribution + ttl_seconds: Time-to-live in seconds + + Returns: + The created ClipboardEntry with assigned index + + Raises: + PluggedInError: If the API request fails + ValueError: If ttl_seconds is <= 0 + """ payload = _build_clipboard_payload( value=value, content_type=content_type, - encoding=encoding if isinstance(encoding, ClipboardEncoding) else ClipboardEncoding(encoding), - visibility=visibility if isinstance(visibility, ClipboardVisibility) else ClipboardVisibility(visibility), + encoding=encoding, + visibility=visibility, created_by_tool=created_by_tool, created_by_model=created_by_model, ttl_seconds=ttl_seconds, ) response = await self.client.request("POST", "/api/clipboard/push", json=payload) - data = response.json() - - if not data.get("success"): - raise PluggedInError(data.get("error", "Failed to push to clipboard")) - - return ClipboardEntry(**data["entry"]) + result = _parse_entry_response( + response.json(), + error_msg="Failed to push to clipboard", + raise_on_failure=True, + ) + assert result is not None + return result async def pop(self) -> Optional[ClipboardEntry]: - """Pop the last indexed entry from clipboard""" - response = await self.client.request("POST", "/api/clipboard/pop") - data = response.json() - - if not data.get("success") or not data.get("entry"): - return None + """Pop the last indexed entry from clipboard (LIFO). - return ClipboardEntry(**data["entry"]) + Returns: + The popped ClipboardEntry, or None if clipboard is empty + """ + response = await self.client.request("POST", "/api/clipboard/pop") + return _parse_entry_response( + response.json(), + error_msg="Failed to pop from clipboard", + raise_on_failure=False, + ) async def delete( self, name: Optional[str] = None, idx: Optional[int] = None, - ) -> bool: - """Delete a clipboard entry by name or index""" - if name is None and idx is None: - raise ValueError("Either 'name' or 'idx' must be provided") + ) -> int: + """Delete a clipboard entry by name or index. + + Args: + name: Entry name to delete + idx: Entry index to delete + + Returns: + Number of deleted entries (0 or 1) - payload = {} - if name is not None: - payload["name"] = name - if idx is not None: - payload["idx"] = idx + Raises: + ValueError: If neither name nor idx is provided + """ + request = ClipboardDeleteRequest(name=name, idx=idx) + payload = _build_delete_payload(request) response = await self.client.request("DELETE", "/api/clipboard", json=payload) + return _parse_delete_response(response.json()) + + async def clear_all(self) -> ClearAllResult: + """Clear all clipboard entries using bulk delete API. + + Returns: + ClearAllResult with deleted count and status + + Raises: + PluggedInError: If the API request fails + """ + response = await self.client.request( + "DELETE", + "/api/clipboard", + json={"clearAll": True} + ) data = response.json() - return data.get("success", False) and data.get("deleted", False) - - async def clear_all(self) -> int: - """Clear all clipboard entries. Returns count of deleted entries.""" - entries = await self.list() - deleted = 0 - - for entry in entries: - try: - success = False - if entry.name: - success = await self.delete(name=entry.name) - elif entry.idx is not None: - success = await self.delete(idx=entry.idx) - if success: - deleted += 1 - except PluggedInError: - pass - - return deleted + if not data.get("success"): + raise PluggedInError(data.get("error", "Failed to clear clipboard")) + + deleted = data.get("deleted", 0) + return ClearAllResult(deleted=deleted, failed=0)