Plugged.in Python SDK

Official Python SDK for the Plugged.in Library API. Full support for both synchronous and asynchronous operations with comprehensive type hints.

PyPI: https://pypi.org/project/pluggedinkit/

Installation

pip install pluggedinkit

Quick Start

Synchronous Client

from pluggedinkit import PluggedInClient

# Initialize the client
client = PluggedInClient(
    api_key="your-api-key",
    # base_url defaults to https://plugged.in
)

# List documents
documents = client.documents.list()
print(f"Found {documents.total} documents")

# Search documents
results = client.documents.search("machine learning")
for result in results.results:
    print(f"{result.title} - Relevance: {result.relevance_score}")

# Query knowledge base
answer = client.rag.ask_question("What are the main features?")
print(answer)

Asynchronous Client

import asyncio
from pluggedinkit import AsyncPluggedInClient

async def main():
    # Initialize async client
    async with AsyncPluggedInClient(api_key="your-api-key") as client:
        # List documents
        documents = await client.documents.list()
        print(f"Found {documents.total} documents")

        # Query RAG
        answer = await client.rag.ask_question("What's new in the project?")
        print(answer)

asyncio.run(main())

Features

• 📄 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
• 🐍 Type Hints - Full typing support with Pydantic models
• 🔁 Retry Logic - Automatic retries with exponential backoff
• 📊 Rate Limiting - Built-in rate limit handling

Authentication

Get your API key from your Plugged.in profile settings:

import os
from pluggedinkit import PluggedInClient

client = PluggedInClient(
    api_key=os.environ["PLUGGEDIN_API_KEY"],
    # For local development:
    # base_url="http://localhost:12005"
)

Documentation

Document Operations

List Documents

from pluggedin.types import DocumentFilters, DocumentSource

filters = DocumentFilters(
    source=DocumentSource.AI_GENERATED,
    tags=["report", "analysis"],
    sort="date_desc",
    limit=20,
    offset=0
)

response = client.documents.list(filters)
for doc in response.documents:
    print(f"{doc.title} ({doc.file_size} bytes)")

Get Document

# Get document metadata
doc = client.documents.get("document-id")

# Get document with content
doc_with_content = client.documents.get(
    "document-id",
    include_content=True,
    include_versions=True
)

print(doc_with_content.content)

Search Documents

from pluggedin.types import SearchFilters

filters = SearchFilters(
    model_provider="anthropic",
    date_from="2024-01-01T00:00:00Z",
    tags=["finance", "q4"]
)

results = client.documents.search(
    "quarterly report",
    filters=filters,
    limit=10,
    offset=0
)

for result in results.results:
    print(f"{result.title}")
    print(f"  Relevance: {result.relevance_score}")
    print(f"  Snippet: {result.snippet}")

Update Document

from pluggedin.types import UpdateDocumentRequest, UpdateOperation

request = UpdateDocumentRequest(
    operation=UpdateOperation.APPEND,
    content="\n\n## New Section\n\nAdditional content here.",
    metadata={
        "changeSummary": "Added implementation details",
        "model": {
            "name": "claude-3-opus",
            "provider": "anthropic",
            "version": "20240229"
        }
    }
)

response = client.documents.update("document-id", request)
print(f"Updated to version {response.version}")

Create AI Document

metadata = {
    "format": "md",
    "category": "documentation",
    "tags": ["api", "guide"],
    "model": {
        "name": "gpt-4",
        "provider": "openai",
        "version": "0613"
    },
    "prompt": "Create an API integration guide",
    "visibility": "workspace"
}

doc = client.documents.create(
    title="API Integration Guide",
    content="# API Integration Guide\n\n## Introduction\n\n...",
    metadata=metadata
)

print(f"Created document: {doc.id}")

RAG Operations

Query Knowledge Base

# Simple query
answer = client.rag.ask_question("What are the deployment procedures?")
print(answer)

# Query with sources
result = client.rag.query_with_sources(
    "Explain the authentication flow",
    project_uuid="project-uuid"  # Optional
)

print(f"Answer: {result['answer']}")
print("Sources:")
for source in result["sources"]:
    print(f"- {source.name} (relevance: {source.relevance}%)")

Find Relevant Documents

documents = client.rag.find_relevant_documents(
    "user authentication",
    project_uuid="project-uuid",
    limit=5
)

for doc in documents:
    print(f"- {doc.name}")
    if doc.model:
        print(f"  By: {doc.model.provider}/{doc.model.name}")

Check RAG Status

# Check availability
status = client.rag.check_availability()
print(f"RAG Available: {status['available']}")

# Get storage stats
stats = client.rag.get_storage_stats()
print(f"Documents: {stats['document_count']}")
print(f"Total size: {stats['total_size']} bytes")

File Upload Operations

Upload Single File

from pathlib import Path

# Upload from file path
file_path = Path("./report.pdf")
metadata = {
    "name": "Q4 Report.pdf",
    "description": "Quarterly financial report",
    "tags": ["finance", "q4", "2024"],
    "purpose": "Financial documentation",
    "relatedTo": "PROJECT-123"
}

def on_progress(percent):
    print(f"Upload progress: {percent}%")

response = client.uploads.upload_file(
    file_path,
    metadata,
    on_progress=on_progress
)

if response.success:
    print(f"Uploaded: {response.document_id}")

    # Track processing
    if response.upload_id:
        def on_update(status):
            print(f"Status: {status.status} - {status.message}")

        client.uploads.track_upload(
            response.upload_id,
            on_update
        )

Upload from Memory

# Upload bytes directly
content = b"File content here..."
metadata = {
    "name": "data.txt",
    "description": "Data file",
    "tags": ["data"]
}

response = client.uploads.upload_file(content, metadata)

Batch Upload

files = [
    {
        "file": Path("doc1.pdf"),
        "metadata": {"name": "doc1.pdf", "tags": ["batch"]}
    },
    {
        "file": Path("doc2.txt"),
        "metadata": {"name": "doc2.txt", "tags": ["batch"]}
    }
]

def on_batch_progress(current, total):
    print(f"Uploaded {current}/{total} files")

results = client.uploads.upload_batch(
    files,
    on_progress=on_batch_progress
)

for i, result in enumerate(results):
    if result.success:
        print(f"✓ {files[i]['metadata']['name']}")
    else:
        print(f"✗ {files[i]['metadata']['name']}: {result.error}")

Async Examples

Async Document Operations

import asyncio

async def document_operations():
    async with AsyncPluggedInClient(api_key="your-key") as client:
        # List documents
        docs = await client.documents.list()

        # Search concurrently
        search_tasks = [
            client.documents.search("api"),
            client.documents.search("guide"),
            client.documents.search("tutorial")
        ]
        results = await asyncio.gather(*search_tasks)

        for result in results:
            print(f"Found {result.total} matches")

asyncio.run(document_operations())

Async RAG Queries

async def rag_operations():
    async with AsyncPluggedInClient(api_key="your-key") as client:
        # Multiple queries concurrently
        questions = [
            "What is the authentication process?",
            "How do I deploy the application?",
            "What are the API rate limits?"
        ]

        tasks = [client.rag.ask_question(q) for q in questions]
        answers = await asyncio.gather(*tasks)

        for q, a in zip(questions, answers):
            print(f"Q: {q}")
            print(f"A: {a}\n")

asyncio.run(rag_operations())

Clipboard Operations

The clipboard provides persistent key-value storage for MCP tools and AI agents.

Set Named Entry

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

# 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

entry = client.clipboard.push(
    value="Processing step 1 result",
    content_type="text/plain"
)

print(f"Pushed to index: {entry.idx}")

Pop from Stack

entry = client.clipboard.pop()
if entry:
    print(f"Popped value: {entry.value}")

List and Delete

# 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

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

from pluggedinkit import (
    PluggedInError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError
)

try:
    doc = client.documents.get("invalid-id")
except AuthenticationError as e:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except NotFoundError as e:
    print("Document not found")
except ValidationError as e:
    print(f"Validation error: {e.details}")
except PluggedInError as e:
    print(f"API error: {e}")

Advanced Configuration

client = PluggedInClient(
    api_key="your-api-key",
    base_url="https://plugged.in",
    timeout=60.0,  # 60 seconds
    max_retries=5,
    debug=True  # Enable debug logging
)

# Update API key at runtime
client.set_api_key("new-api-key")

Type Safety

The SDK uses Pydantic for comprehensive type safety:

from pluggedin.types import (
    Document,
    DocumentFilters,
    DocumentSource,
    DocumentVisibility,
    UpdateOperation,
    ModelInfo
)

# All types are validated
filters = DocumentFilters(
    source=DocumentSource.AI_GENERATED,
    limit=10  # Validated: must be > 0 and <= 100
)

# IDE autocomplete and type checking
doc: Document = client.documents.get("id")
print(doc.title)  # Type-safe attribute access

Environment Variables

Store your API key securely:

# .env
PLUGGEDIN_API_KEY=your-api-key
PLUGGEDIN_BASE_URL=https://plugged.in

import os
from dotenv import load_dotenv
from pluggedin import PluggedInClient

load_dotenv()

client = PluggedInClient(
    api_key=os.environ["PLUGGEDIN_API_KEY"],
    base_url=os.environ.get("PLUGGEDIN_BASE_URL")
)

Rate Limiting

The SDK automatically handles rate limiting:

• API Endpoints: 60 requests per minute
• Document Search: 10 requests per hour for AI documents
• RAG Queries: Subject to plan limits

Examples

See the examples directory for complete working examples:

• Basic Usage
• Document Management
• RAG Queries
• File Upload
• Async Operations
• Error Handling

Contributing

Contributions are welcome! Please see our Contributing Guide for details.

License

MIT - see LICENSE for details.

Support

• PyPI Package: https://pypi.org/project/pluggedinkit/
• GitHub Repository: https://github.com/VeriTeknik/pluggedinkit-python
• Documentation: https://docs.plugged.in
• Issues: GitHub Issues
• Discord: Join our community

Changelog

See CHANGELOG.md for release history.