Elsai

Appearance

Elsai Cloud Connectors ​

Package: elsai-cloud  v2.1.0

Utilities to interact with cloud storage, email, collaboration, and search services — AWS S3, Azure Blob, SharePoint, OneDrive, Outlook, Elasticsearch, Microsoft Graph webhooks, and a generic email service.

Installation ​

bash
pip install --extra-index-url https://core-packages.elsai.ai/root/elsai-cloud-connectors/ elsai-cloud==2.1.0

Requirements: Python >= 3.9

Available connectors ​

ServiceClassImport path
AWS S3AwsS3Connectorelsai_cloud.aws
Azure Blob StorageAzureBlobStorageelsai_cloud.azureblobstorage
Microsoft SharePointSharePointServiceelsai_cloud.sharepoint
Microsoft OneDriveOneDriveServiceelsai_cloud.onedrive
Microsoft OutlookOutlookServiceelsai_cloud.outlook
Microsoft Graph WebhooksMSGraphWebhookselsai_cloud.microsoft_webhooks
Microsoft Graph DeltaMSGraphDeltaServiceelsai_cloud.microsoft_delta
ElasticsearchElasticSearchConnectorelsai_cloud.elastic_search
Generic EmailEmailServiceelsai_cloud.email_services

AwsS3Connector ​

Upload, download, list, and delete objects in AWS S3 buckets. Supports optional session tokens for temporary credentials and an optional region parameter.

python
from elsai_cloud.aws import AwsS3Connector

s3_client = AwsS3Connector(
    access_key="your_access_key",
    secret_key="your_secret_key",
    session_token="your_session_token",
)

# With an explicit region
s3_client = AwsS3Connector(
    access_key="your_access_key",
    secret_key="your_secret_key",
    session_token="your_session_token",
    region="us-east-1",
)

Constructor parameters:

ParameterRequiredDescription
access_keyYesAWS access key ID
secret_keyYesAWS secret access key
session_tokenYesAWS session token (use None for long-term credentials)
regionNoAWS region (e.g. "us-east-1")

Upload a file:

python
s3_client.upload_file_to_s3(
    bucket_name="your_bucket_name",
    s3_key="your_s3_key",
    file_path="path/to/your/file.txt",
)

Download a file:

python
s3_client.download_file_from_s3(
    bucket_name="your_bucket_name",
    file_name="your_s3_key",
    download_path="path/to/download/file.txt",
)

Delete a file:

python
s3_client.delete_file_from_s3(
    bucket_name="your_bucket_name",
    s3_key="your_s3_key",
)

List objects in a folder:

python
objects = s3_client.list_objects_in_s3_folder(
    bucket_name="your_bucket_name",
    prefix="your_folder_path/",
)
print(objects)

Download an entire folder:

python
s3_client.download_folder_from_s3(
    bucket_name="your_bucket_name",
    prefix="your_folder_path/",
    download_dir="path/to/local/download/folder",
)

AzureBlobStorage ​

Download files from Azure Blob Storage containers using a connection string.

python
from elsai_cloud.azureblobstorage import AzureBlobStorage

azure_blob_client = AzureBlobStorage(
    connection_string="your_connection_string",
)

azure_blob_client.download_file(
    container_name="your_container_name",
    blob_name="your_blob_name",
    target_folder_path="path/to/download/folder",
)

Constructor parameters:

ParameterRequiredDescription
connection_stringYesAzure Storage account connection string

SharePointService ​

Upload, list, and download files from SharePoint document libraries using the Microsoft Graph API.

python
from elsai_cloud.sharepoint import SharePointService

sharepoint_service = SharePointService(
    tenant_id="your_tenant_id",
    client_id="your_client_id",
    client_secret="your_client_secret",
    site_hostname="your_site_hostname",
    site_path="your_site_path",
    drive_name="your_drive_name",
    drive_id="your_drive_id",
)

Constructor parameters:

ParameterRequiredDescription
tenant_idYesAzure AD tenant ID
client_idYesAzure AD app client ID
client_secretYesAzure AD app client secret
site_hostnameYesSharePoint site hostname (e.g. yourorg.sharepoint.com)
site_pathYesSharePoint site path (e.g. /sites/MyTeam)
drive_nameYesDocument library name
drive_idYesDocument library drive ID

Upload a file:

python
sharepoint_service.upload_file_to_sharepoint(
    file_path="path/to/your/file.txt",
    target_folder="Documents/FolderName",
)

List files in a folder:

python
files = sharepoint_service.retrieve_sharepoint_files_from_folder(
    folder_name="folderName",
)

Download a file:

python
sharepoint_service.download_file_from_sharepoint(
    file_id="your_file_id",
    target_folder="path/to/download/folder",
)

OneDriveService ​

Upload, list, and download files from a user's OneDrive via the Microsoft Graph API.

python
from elsai_cloud.onedrive import OneDriveService

one_drive_service = OneDriveService(
    tenant_id="your_tenant_id",
    client_id="your_client_id",
    client_secret="your_client_secret",
)

Constructor parameters:

ParameterRequiredDescription
tenant_idYesAzure AD tenant ID
client_idYesAzure AD app client ID
client_secretYesAzure AD app client secret

Get user ID from email:

python
user_id = one_drive_service.get_user_id(email="your_mail_address")

Upload a file:

python
one_drive_service.upload_file_to_onedrive(
    email="your_mail_address",
    local_file_path="path/to/your/file.txt",
    folder_path="path/to/folder/in/onedrive",
)

List files in a folder:

python
files = one_drive_service.retrieve_onedrive_files_from_folder(
    email="your_mail_address",
    folder_path="path/to/folder/in/onedrive",
)

Download a file:

python
one_drive_service.download_file_from_onedrive(
    email="your_mail_address",
    file_id="your_file_id",
    target_folder="path/to/download/folder",
)

OutlookService ​

Send emails, list messages, retrieve message details, and download attachments via the Microsoft Graph API.

python
from elsai_cloud.outlook import OutlookService

outlook_service = OutlookService(
    tenant_id="your_tenant_id",
    client_id="your_client_id",
    client_secret="your_client_secret",
)

Constructor parameters:

ParameterRequiredDescription
tenant_idYesAzure AD tenant ID
client_idYesAzure AD app client ID
client_secretYesAzure AD app client secret

Send a plain-text email:

python
result = outlook_service.send_email(
    sender_email="sender@example.com",
    to_recipients=["recipient@example.com"],
    subject="Test Email",
    body="This is a test email.",
    cc_recipients=["cc@example.com"],
    is_html=False,
)
print("Result:", result)

Send an email with attachments:

python
result = outlook_service.send_email(
    sender_email="sender@example.com",
    to_recipients=["recipient@example.com"],
    subject="Email with Attachments",
    body="Please find the attached files.",
    attachments=["/path/to/file.pdf"],
    is_html=False,
)

Send an HTML email:

python
html_body = """
<html><body>
    <h1>Welcome!</h1>
    <p>This is an <strong>HTML formatted</strong> email.</p>
</body></html>
"""

result = outlook_service.send_email(
    sender_email="sender@example.com",
    to_recipients=["recipient@example.com"],
    subject="HTML Email",
    body=html_body,
    is_html=True,
)

send_email parameters:

ParameterRequiredDescription
sender_emailYesSender's email address
to_recipientsYesList of recipient email addresses
subjectYesEmail subject
bodyYesEmail body (plain text or HTML)
cc_recipientsNoList of CC email addresses
attachmentsNoList of local file paths to attach
is_htmlNoSet True to send body as HTML (default: False)

List messages:

python
emails = outlook_service.list_messages(
    user_email="user@example.com",
    top=25,
    query="Invoices",   # optional filter keyword
)

if emails.get("value"):
    print(f"Found {len(emails['value'])} emails")

Get a specific message:

python
message = outlook_service.get_message(
    user_email="user@example.com",
    message_id="message_id_here",
)
print(f"Subject: {message.get('subject')}")

Get a message with attachment status:

python
msg = outlook_service.get_message_with_attachments(
    user_email="user@example.com",
    message_id="message_id_here",
)
print(f"Has attachments: {msg['has_attachments']}")

List and download attachments:

python
attachments = outlook_service.list_attachments(
    user_email="user@example.com",
    message_id="message_id_here",
)

if attachments.get("value"):
    att = attachments["value"][0]
    saved_path = outlook_service.download_attachment(
        user_email="user@example.com",
        message_id="message_id_here",
        attachment_id=att["id"],
        attachment_name=att["name"],
        download_dir="./downloads",
    )
    print(f"Saved to: {saved_path}")

MSGraphWebhooks ​

Create and manage Microsoft Graph change notification subscriptions to receive webhook events for resources like mail, calendar, and OneDrive.

python
from elsai_cloud.microsoft_webhooks import MSGraphWebhooks

ms_graph = MSGraphWebhooks(
    client_id="your_client_id",
    tenant_id="your_tenant_id",
    client_secret="your_client_secret",
)

Create a subscription:

python
ms_graph.create_subscription(
    change_type="change_type",
    notification_url="https://webhook.example.com/notifications",
    resource="me/mailFolders('Inbox')/messages",
    expiration_date_time="2016-11-20T18:23:45.9356913Z",
    client_state="custom_state_123",
)

Get a subscription:

python
subscription = ms_graph.get_subscription(subscription_id="your_subscription_id")

List all subscriptions:

python
subs_list = ms_graph.list_subscriptions()

Update a subscription:

python
ms_graph.update_subscription(
    subscription_id="subscription_id",
    notification_url="https://webhook.example.com/notifications",
    expiration_date_time="2016-11-20T18:23:45.9356913Z",
)

Delete a subscription:

python
ms_graph.delete_subscription(subscription_id="subscription_id")

MSGraphDeltaService ​

Track incremental changes in Outlook mailboxes using Microsoft Graph delta queries. Establish a delta link on first sync, then use it to fetch only new or changed messages on subsequent calls.

python
from elsai_cloud.microsoft_delta import MSGraphDeltaService, DeltaConfig, DeltaStrategy
from datetime import datetime, timezone, timedelta
import asyncio

async def sync_emails():
    config = DeltaConfig(
        timeout=45.0,
        max_retries=3,
        page_size=50,
        establishment_strategies=[
            DeltaStrategy.CURRENT_TIME,
            DeltaStrategy.FUTURE_FILTER,
            DeltaStrategy.SKIP_TOKEN,
            DeltaStrategy.STANDARD,
        ],
    )

    service = MSGraphDeltaService(
        config=config,
        client_id="your_client_id",
        client_secret="your_client_secret",
        tenant_id="your_tenant_id",
    )

    user_id = "user@example.com"
    folder = "inbox"

    # Establish a delta link (first sync)
    delta_link = await service.establish_delta_link(user_id, folder)

    # Validate the delta link
    is_valid = await service.validate_delta_link(delta_link)
    print(f"Delta link valid: {is_valid}")

    # Get all messages (initial sync)
    initial_result = await service.get_initial_delta(user_id, folder)
    print(f"Total messages: {initial_result.total_count}")

    # Filter to last 7 days
    start_date = datetime.now(timezone.utc) - timedelta(days=7)
    filtered = await service.get_initial_delta(user_id, folder, start_date)
    print(f"Messages in last 7 days: {filtered.total_count}")

    # Get incremental changes since last sync
    if initial_result.delta_link:
        changes = await service.get_delta_changes(initial_result.delta_link)
        print(f"New changes: {changes.total_count}")

    # Get delta statistics
    if initial_result.delta_link:
        stats = await service.get_delta_statistics(initial_result.delta_link)
        print(f"Stats: {stats}")

    # Update config at runtime
    new_config = DeltaConfig(timeout=60.0, max_retries=5, page_size=100)
    service.update_config(new_config)

asyncio.run(sync_emails())

DeltaConfig parameters:

ParameterDescription
timeoutRequest timeout in seconds
max_retriesMaximum retry attempts on failure
page_sizeNumber of messages per page
establishment_strategiesOrdered list of DeltaStrategy values to try when establishing a delta link

DeltaStrategy options:

StrategyDescription
CURRENT_TIMEStart delta from the current timestamp
FUTURE_FILTERUse a filter for future messages
SKIP_TOKENUse skip token pagination
STANDARDStandard delta query

ElasticSearchConnector ​

Index, retrieve, and search documents in Elasticsearch using cloud URL and API key authentication.

python
from elsai_cloud.elastic_search import ElasticSearchConnector

es_connector = ElasticSearchConnector(
    cloud_url="your_cloud_url",
    api_key="your_api_key",
)

Add a document:

python
doc = {
    "text": "This is a sample document to be indexed in Elasticsearch.",
}

es_connector.add_document(
    index_name="sampleindex",
    document=doc,
    doc_id="1",
)

Get a document:

python
retrieved_doc = es_connector.get_document(
    index_name="sampleindex",
    doc_id="1",
)

Search documents:

python
results = es_connector.search_documents(
    index_name="sampleindex",
    query={"match": {"text": "sample"}},
)

EmailService ​

A generic IMAP/SMTP email service supporting OAuth2 authentication. Supports sending, listing, reading, replying, forwarding, and downloading attachments.

python
from elsai_cloud.email_services import EmailService, AuthType
import os

service = EmailService(
    email_address=os.getenv("EMAIL_ADDRESS"),
    auth_type=AuthType.OAUTH2,
)

Send a plain-text email:

python
service.send_email(
    to_email="recipient@example.com",
    subject="Test Email",
    body="This is a test email from the Elsai Cloud Connector.",
)

Send with CC, BCC, and attachments:

python
service.send_email(
    to_email="recipient@example.com",
    subject="Email with Attachments",
    body="This email contains multiple attachments.",
    cc=["cc1@example.com"],
    bcc=["bcc1@example.com"],
    attachments=["/path/to/attachment1.pdf", "/path/to/image.png"],
)

List recent emails:

python
emails = service.list_emails(limit=5)
for mail in emails:
    print(f"[{mail['index']}] {mail['subject']} | {mail['from']} | UID={mail['uid']}")

Read a specific email and download attachments:

python
email_details = service.get_email(uid=emails[0]["uid"], download_dir="./downloads")
print(f"Subject: {email_details['subject']}")
print(f"Body: {email_details['body']}")

if email_details.get("attachments"):
    for att in email_details["attachments"]:
        print(f"Attachment saved to: {att.get('saved_to')}")

Reply to an email:

python
service.reply(
    uid=emails[0]["uid"],
    reply_text="Thank you for your email.",
)

# With attachment
service.reply(
    uid=emails[0]["uid"],
    reply_text="Please find the document attached.",
    attachments=["/path/to/document.pdf"],
)

Forward an email:

python
service.forward(
    uid=emails[0]["uid"],
    to_email="forwarded@example.com",
)

# With CC, BCC, and attachment
service.forward(
    uid=emails[0]["uid"],
    to_email="forwarded@example.com",
    cc=["manager@example.com"],
    bcc=["archive@example.com"],
    attachments=["/path/to/document.pdf"],
)

Version history ​

VersionChanges
2.1.0Current stable release
1.0.0OutlookService, MSGraphDeltaService, MSGraphWebhooks, EmailService added

Copyright © 2026 Elsai Foundry.

Built with VitePress