connectors

🇬🇧 English | 🇩🇪 DE | 🇪🇸 ES | 🇯🇵 JA | 🇷🇺 RU | 🇨🇳 ZH

Standalone, zero-dependency messaging connectors for autonomous AI agents — Telegram, Discord, Signal, WhatsApp, Home Assistant, Webhooks, Slack, and macOS iMessage.

Extracted and decoupled from BACH. No external framework required. Zero mandatory runtime dependencies (100% Python standard library).

Note

LLM & Agent-Native Architecture: connectors is engineered specifically for autonomous AI agents, multi-agent swarms, and cognitive runtime loops (such as BACH, USMC, and clutch). It provides a zero-dependency, standardized messaging contract (connect(), send_message(), poll_threaded()) allowing AI agents to interact with human operators across multi-channel chat platforms without framework lock-in. Machine-readable context available at llms.txt.

---

Quick Navigation

• Key Features
• Target Personas & Discoverability
• Comparative Matrix vs Alternatives
• System Architecture
• Interactive Messaging & Polling Lifecycle
• Supported Connectors & Status
• Governance & Safety Invariants
• Quick Start
• Slack & iMessage Connectors
• Secret Management & Zero Leakage
• Threaded Polling & Event Callbacks
• Interactive Setup Wizard & Templates
• BACH Framework Integration
• Sibling Ecosystem & Partner Repositories
• Smoke Testing & Verification
• Security Policy & Vulnerability Reporting
• Third-Party Licenses & Notices
• Contributing
• License

---

Key Features

• 100% Python Standard Library Core: Zero external runtime pip dependencies (urllib.request, json, threading, subprocess, sqlite3). No bloat, minimal attack surface, instant cold starts.
• Unified Abstract Connector Contract: Standardized BaseConnector ABC with uniform signatures across Telegram, Discord, Signal, WhatsApp, Home Assistant, Webhooks, Slack, and macOS iMessage.
• Zero Runtime Secret Leakage: Credentials stored in ConnectorConfig.auth_config are hidden via field(repr=False). All class __repr__() implementations mask secrets to prevent token exposure in logs or stack traces.
• Pluggable Credential Resolution: Direct environment access (os.environ), .env support, or pluggable SecretAdapter for vault and framework integration.
• Thread-Safe Decoupled Polling: Built-in poll_threaded() background worker with threading.Event stop triggers and isolated exception handling.
• Native macOS iMessage Support: Direct read-only querying of macOS chat.db (SQLite) and AppleScript osascript message dispatch with fail-closed safety on non-Darwin platforms.
• Multi-Platform Certified: Verified and tested across Ubuntu Linux, Windows, and macOS via GitHub Actions.
• Interactive Scaffolding CLI: Standalone setup wizard (python -m connectors.templates.setup_wizard) with YAML templates for rapid connector authoring.

---

Target Personas & Discoverability

Persona	Core Profile & Tech Stack	Architectural Friction & Pain Point	How connectors Solves It
Autonomous AI Agent Engineers	Building agent loops & swarms (BACH, USMC, LangChain, AutoGen, CrewAI).	Heavyweight bot frameworks (discord.py, python-telegram-bot) impose asynchronous event loops that conflict with agent runtime loops.	100% stdlib synchronous contract with non-blocking poll_threaded(), pluggable SecretAdapter, and zero pip dependencies.
Self-Hosted Homelab Automators	Automating smart home alerts across Home Assistant, Signal, Telegram & Webhooks.	Fragmented APIs, disparate config structures, and risk of token leakage in git commits or log files.	Uniform YAML template wizard, strict token masking in repr/logs (field(repr=False)), and instant multi-OS portability.
Privacy-Conscious SecOps	Enterprise security officers auditing software supply chain risks.	Third-party SDKs pulling dozens of transitive dependencies with unpredictable CVE surfaces and disk persistence.	Strict dependencies = [], unprivileged user-mode execution, shell injection immunity (shell=False), and 48h security SLA.
Multi-Platform Dispatcher Builders	DevOps teams routing alerting gateways across Slack, iMessage, WhatsApp & Webhooks.	Redundant SDK code, differing auth schemas, and fragile error handlers cascading crashes.	Single unified send_message() interface across all 8 channels with fail-closed error handling and isolated workers.

Discovery Keywords & Topic Tags: autonomous-agents, ai-messaging, zero-dependencies, stdlib-only, telegram-bot, discord-webhook, signal-cli, whatsapp-business-api, home-assistant-notify, slack-bot, apple-imessage, local-first, privacy-first, multi-agent-swarms, python-stdlib.

---

Comparative Matrix vs Alternatives

Architectural Criterion	connectors (ellmos-ai)	Individual SDKs (discord.py, python-telegram-bot)	All-in-One Frameworks (Errbot, OpsDroid)	Heavyweight Agent Tooling (LangChain community tools)
Runtime Dependencies	0 (100% Python Stdlib)	High (15–40+ transitive pip packages)	Very High (50+ packages, plugins)	Extreme (100+ transitive packages)
Cold-Start Overhead	< 1 ms	~200–500 ms	~800–2000 ms	~1500–4000 ms
Memory Footprint	~12–18 MB	~45–80 MB	~90–180 MB	~150–350 MB
Secret Masking Guarantee	Strict field(repr=False)	Inconsistent / Manual	Configuration dependent	Prone to prompt/log leakages
Unified Interface ABC	Yes (BaseConnector)	No (Platform-specific APIs)	Partial (Plugin-specific models)	Generic string I/O wrapper
Thread Polling vs Async Lock	Built-in poll_threaded()	Forces AsyncIO loop hijacking	Fixed threading/async runtime	Variable / External executor
Shell Injection Immunity	Strict shell=False lists	N/A (WebSockets/HTTP)	Shell commands possible	Dynamic subprocess risk
macOS Native iMessage	Yes (Direct SQLite + OSA)	No	No	No
Supply Chain Attack Surface	Zero (0 pip CVEs)	Broad attack surface	Broad attack surface	Very broad attack surface

---

System Architecture

flowchart TD
    subgraph AgentLayer ["Autonomous Agent / Client Layer"]
        A["Autonomous Agent / Multi-Agent Swarm (BACH / USMC)"]
        B["Cognitive Loop / Scheduler"]
    end

    subgraph CoreFactory ["Core Factory & Configuration"]
        CF["create_connector Factory"]
        CC["ConnectorConfig DataClass"]
        SA["SecretAdapter Hook"]
    end

    subgraph ConnectorsModule ["connectors Standalone Core (100% Stdlib)"]
        BC["BaseConnector (ABC)"]
        TC["TelegramConnector"]
        DC["DiscordConnector"]
        SC["SignalConnector"]
        WC["WhatsAppConnector"]
        HC["HomeAssistantConnector"]
        WH["WebhookConnector"]
        SLC["SlackConnector"]
        IMC["iMessageConnector"]
    end

    subgraph ExternalPlatforms ["External Messaging Channels & Protocols"]
        EP_TG["Telegram Bot API (Long-Polling & Send)"]
        EP_DC["Discord Gateway / Webhook REST API"]
        EP_SG["signal-cli IPC / Subprocess Daemon"]
        EP_WA["WhatsApp Cloud / On-Premises Business API"]
        EP_HA["Home Assistant REST API / Notify"]
        EP_WH["Custom Webhook Endpoint (HTTP POST)"]
        EP_SL["Slack Web API / Incoming Webhooks"]
        EP_IM["macOS chat.db (Read) & osascript (Send)"]
    end

    A -->|"Instantiates Config"| CC
    B -->|"Secret Resolution"| SA
    CC --> CF
    SA --> CF
    CF -->|"Instantiates"| BC

    BC --> TC
    BC --> DC
    BC --> SC
    BC --> WC
    BC --> HC
    BC --> WH
    BC --> SLC
    BC --> IMC

    TC -->|"HTTPS POST / getUpdates"| EP_TG
    DC -->|"HTTPS POST / Execute Webhook"| EP_DC
    SC -->|"CLI Arguments (No Shell)"| EP_SG
    WC -->|"HTTPS POST / Graph API"| EP_WA
    HC -->|"HTTPS POST / Services"| EP_HA
    WH -->|"JSON Payload"| EP_WH
    SLC -->|"HTTPS POST / chat.postMessage"| EP_SL
    IMC -->|"SQLite Read-Only / AppleScript"| EP_IM

Loading

---

Interactive Messaging & Polling Lifecycle

sequenceDiagram
    autonumber
    actor Operator as Human / External User
    participant Platform as Messaging Platform (Telegram/Discord/Slack/iMessage)
    participant Worker as Background Polling Loop (poll_threaded)
    participant Conn as BaseConnector Instance
    participant Agent as Autonomous Agent Loop

    Agent->>Conn: connect()
    Conn->>Platform: Probe API / Verify Credentials / Check DB
    Platform-->>Conn: 200 OK / Authenticated / DB Accessible
    Conn-->>Agent: True (Connected)

    Agent->>Conn: poll_threaded(on_message=callback)
    activate Worker
    Conn-->>Agent: (WorkerThread, StopEvent)

    loop Polling Loop (interval=5.0s)
        Worker->>Conn: get_messages(since, limit=50)
        Conn->>Platform: Fetch pending updates / Query chat.db
        Platform-->>Conn: Return JSON updates / New message rows
        Conn->>Conn: Parse & sanitize into List[Message]
        Conn-->>Worker: messages
        alt New Messages Available
            Worker->>Agent: callback(message)
            Agent->>Agent: Process cognitive prompt
            Agent->>Conn: send_message(recipient_id, response_text)
            Conn->>Platform: Dispatch HTTP POST / osascript send
            Platform-->>Operator: Deliver message to user
        end
    end

    Agent->>Worker: StopEvent.set()
    deactivate Worker
    Agent->>Conn: disconnect()
    Conn-->>Agent: True (Disconnected)

Loading

---

Supported Connectors & Status

Connector	Protocol & Transport	Target Ecosystem	Status	Secret Keys Required
telegram	Telegram Bot API (HTTPS)	Telegram Groups & Direct Chats	Production	bot_token
discord	Discord Bot API / Webhook (HTTPS)	Discord Guilds & Channels	Production	bot_token or webhook_url
signal	signal-cli Process IPC	Encrypted Signal Messenger	Production	phone_number
whatsapp	WhatsApp Business REST API	Meta Cloud API / On-Premises	Production	api_token, phone_number_id
homeassistant	Home Assistant REST API	Smart Home Notifications	Production	access_token
webhook	Generic HTTP POST (JSON Payload)	Custom Webhooks & Ingestion	Baseline	Optional api_key / secret
slack	Slack Web API / Incoming Webhook (HTTPS)	Slack Channels & Workspaces	Production	bot_token or webhook_url
imessage	macOS chat.db SQLite & osascript IPC	Apple iMessage / macOS Desktop	Production	None (macOS Full Disk Access)

---

Governance & Safety Invariants

Invariant ID	Safety Invariant	Architectural Implementation	Validation & Guarantees
INV-LOCAL-01	Zero Runtime Dependencies	100% Python Standard Library (urllib.request, json, threading, subprocess, sqlite3).	Audited in pyproject.toml (dependencies = []) and regression tests.
INV-SECRET-02	Zero Secret Persistence	Credentials stored exclusively in-memory; tokens are never written to disk or logs.	Validated in tests/test_repository_hygiene.py and tests/test_behavior.py.
INV-MASK-03	Masked String Representation	ConnectorConfig.auth_config has field(repr=False); connector __repr__() masks secrets.	Strict assertion tests prevent credentials from leaking into debug prints.
INV-SHELL-04	Shell Injection Immune	Process calls in SignalConnector and iMessageConnector strictly use array parameter passing (shell=False).	Prevents arbitrary command execution on POSIX and Windows environments.
INV-ASYNC-05	Non-Blocking Execution	poll_threaded() manages daemon threads with cooperative cancellation (threading.Event).	Prevents event loops from freezing; clean shutdown guaranteed.
INV-FAIL-06	Fail-Closed Error Handling	Network anomalies and malformed responses return False or empty collections; errors to stderr.	Agent loops remain resilient without uncaught runtime crashes.
INV-PLAT-07	Local Platform Isolation	iMessageConnector queries macOS chat.db read-only; non-Darwin platforms fail closed safely.	Verified in cross-platform test fixtures and mock isolation suites.
INV-PRIV-08	Unprivileged Execution	Operates strictly within user-space without requiring root, sudo, or system elevation.	Hardened in security policy and audited against privileged syscalls.
INV-CI-09	Multi-Platform Support	Platform path separators and sub-process execution normalized across OS families.	Verified in CI matrix across Ubuntu Linux, Windows, and macOS.
INV-SLA-10	Bilingual Security Governance	48-hour response SLA and 5-day triage commitment in dual-language SECURITY.md.	Verified in tests/test_metadata.py and tests/test_repository_hygiene.py.

---

Quick Start

Installation

# Core package (100% stdlib - no external pip dependencies)
pip install git+https://github.com/ellmos-ai/connectors.git

# Editable local installation for development
git clone https://github.com/ellmos-ai/connectors.git
cd connectors
pip install -e ".[test,wizard]"

Basic Telegram Usage

import os
from connectors import create_connector, ConnectorConfig

# Configure Telegram bot via environment variables
config = ConnectorConfig(
    name="agent_assistant",
    connector_type="telegram",
    auth_config={"bot_token": os.environ["TELEGRAM_BOT_TOKEN"]},
    options={"owner_chat_id": os.environ.get("OWNER_CHAT_ID", "")},
)

connector = create_connector(config)

if connector.connect():
    # Send an outgoing message
    connector.send_message(recipient=os.environ["OWNER_CHAT_ID"], content="Agent online and ready.")

    # Start non-blocking background message receiver
    def handle_incoming(msg):
        print(f"Received from {msg.sender}: {msg.content}")

    thread, stop_event = connector.poll_threaded(on_message=handle_incoming, interval=3.0)

    # Stop polling when done
    # stop_event.set()
    # connector.disconnect()

Discord Webhook Usage

import os
from connectors import create_connector, ConnectorConfig

config = ConnectorConfig(
    name="discord_alerts",
    connector_type="discord",
    auth_config={"webhook_url": os.environ["DISCORD_WEBHOOK_URL"]},
)

connector = create_connector(config)
if connector.connect():
    connector.send_message(recipient="", content="Deployment pipeline completed successfully! :rocket:")

---

Slack & iMessage Connectors

Slack Bot & Webhook Usage

import os
from connectors import create_connector, ConnectorConfig

# Option A: Slack Bot API (Token-based)
config_bot = ConnectorConfig(
    name="slack_bot",
    connector_type="slack",
    auth_config={"bot_token": os.environ["SLACK_BOT_TOKEN"]},
    options={"channel": "#general"},
)
slack_conn = create_connector(config_bot)
if slack_conn.connect():
    slack_conn.send_message(recipient="#general", content="Hello from autonomous agent!")

# Option B: Slack Incoming Webhook
config_webhook = ConnectorConfig(
    name="slack_webhook",
    connector_type="slack",
    auth_config={"webhook_url": os.environ["SLACK_WEBHOOK_URL"]},
)
webhook_conn = create_connector(config_webhook)
if webhook_conn.connect():
    webhook_conn.send_message(recipient="", content="Alert: High latency detected.")

macOS iMessage Usage

from connectors import create_connector, ConnectorConfig

# Connect to local macOS iMessage (requires macOS and Full Disk Access for chat.db)
config_imessage = ConnectorConfig(
    name="imessage_local",
    connector_type="imessage",
    options={"service": "iMessage"},
)

imessage_conn = create_connector(config_imessage)
if imessage_conn.connect():
    # Query recent messages from SQLite chat.db
    messages = imessage_conn.get_messages(limit=5)
    for msg in messages:
        print(f"[{msg.timestamp}] {msg.sender}: {msg.content}")

    # Dispatch outgoing message via AppleScript osascript
    imessage_conn.send_message(recipient="+1234567890", content="Agent task completed.")

---

Secret Management & Zero Leakage

To prevent leaking sensitive bot tokens, API keys, or phone numbers, connectors provides three tiers of secret resolution:

# Tier 1: Direct environment variables (Recommended for CLI & container runtimes)
config = ConnectorConfig(
    name="telegram_bot",
    connector_type="telegram",
    auth_config={"bot_token": os.environ.get("TELEGRAM_BOT_TOKEN", "")}
)

# Tier 2: Decoupled SecretAdapter (Recommended for Enterprise Vaults & Frameworks)
from connectors.base import SecretAdapter

class CustomVaultAdapter(SecretAdapter):
    def __init__(self, vault_client):
        self.vault = vault_client

    def get_secret(self, key: str) -> str:
        return self.vault.read_secret(f"secret/connectors/{key}")

connector = create_connector(config, secret_adapter=CustomVaultAdapter(vault_client))

---

Threaded Polling & Event Callbacks

All connectors implement threaded polling for non-blocking asynchronous event handling:

import time
from connectors import create_connector, ConnectorConfig

config = ConnectorConfig(
    name="signal_listener",
    connector_type="signal",
    auth_config={"phone_number": "+1234567890"},
)

conn = create_connector(config)
if conn.connect():
    def on_user_input(message):
        print(f"Message from {message.sender} at {message.timestamp}: {message.content}")

    # Launch background thread
    worker_thread, stop_event = conn.poll_threaded(
        on_message=on_user_input,
        interval=5.0
    )

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Stopping listener...")
        stop_event.set()
        worker_thread.join(timeout=10.0)
        conn.disconnect()

---

Interactive Setup Wizard & Templates

Create new connectors effortlessly using the built-in scaffolding wizard:

# Run interactive CLI wizard
python -m connectors.templates.setup_wizard

The wizard guides you through selecting transport protocols, secret requirements, and generating production-ready connector skeletons matching the BaseConnector specification. Ready-to-use YAML configuration templates are available in templates/:

• templates/signal_template.yaml
• templates/discord_template.yaml
• templates/telegram_template.yaml
• templates/whatsapp_template.yaml

---

BACH Framework Integration

To integrate connectors with the BACH Autonomous Cognitive Hub, bind BACH's internal secret management system via SecretAdapter:

from connectors.base import SecretAdapter
from connectors import create_connector, ConnectorConfig

class BachSecretAdapter(SecretAdapter):
    def get_secret(self, key: str) -> str:
        try:
            from hub.secrets_handler import SecretsHandler
            return SecretsHandler().get_secret(key) or ""
        except ImportError:
            return ""

# Wire BACH runtime with external connectors
config = ConnectorConfig(name="bach_telegram", connector_type="telegram")
connector = create_connector(config, secret_adapter=BachSecretAdapter())

---

Sibling Ecosystem & Partner Repositories

connectors is an integral pillar of the autonomous agent and desktop tooling ecosystem stewarded by ellmos-ai and open-bricks:

Repository	Organization	Architectural Role in Autonomous Agent Stack
ellmos-ai/bach	ellmos-ai	Autonomous cognitive agent core & multi-agent supervisor.
ellmos-ai/usmc	ellmos-ai	Universal Shared Memory Coordinator (Agent-to-Agent state).
ellmos-ai/clutch	ellmos-ai	Dynamic LLM routing, inference fallback & cost management.
ellmos-ai/companion-for-agy	ellmos-ai	PTY stdout capture, interactive telemetry & runtime wrapper.
ellmos-ai/system-gap-master	ellmos-ai	Multi-host filesystem reconciliation & conflict management.
dev-bricks/lock-master	dev-bricks	Multi-agent distributed concurrency & cooperative locking.
dev-bricks/ticket-master	dev-bricks	Autonomous ticket dispatch, work queuing & review handoffs.
dev-bricks/automation-master	dev-bricks	Multi-agent fleet orchestration & headless automation controller.
dev-bricks/safe-start-for-codex	dev-bricks	Secure agent sandbox initialization & process management.
file-bricks/CloudLockFixer	file-bricks	Resilient cloud-synced IO engine (cldflt.sys lock mitigation).
file-bricks/ProSync	file-bricks	High-performance folder mirroring & cross-device sync.
file-bricks/ExplorerPro	file-bricks	Advanced Windows Explorer shell extensions & inspector tools.
doc-bricks/FormularErstellen	doc-bricks	Automated document drafting & sovereign form generation.
ellmos-ai/n8n-manager-mcp	ellmos-ai	Autonomous n8n workflow management & staging via MCP.
ellmos-ai/ellmos-homebase-mcp	ellmos-ai	Smart home telemetry & local bridge protocol over MCP.
open-bricks/.github	open-bricks	Global open-source umbrella governance & shared CI templates.

---

Smoke Testing & Verification

Run the comprehensive test suite and validation scripts locally:

# 1. Run all Pytest regression and contract suites
pytest -v

# 2. Run standalone import smoke test
python tests/test_imports.py

# 3. Verify Python bytecode compilation across all modules
python -m compileall -q -x "(^|[\\/])(build|templates[\\/]connector_template\.py)" .

# 4. Run automated code style and linter inspection
ruff check .

---

Security Policy & Vulnerability Reporting

Security and privacy are fundamental design requirements for connectors. We maintain a strict security response policy:

• Supported Versions: Security updates and patches are actively provided for 1.2.x and 1.1.x.
• 48-Hour Response SLA: All security disclosures receive an acknowledgment within 48 hours and an initial triage assessment within 5 business days.
• Reporting Channel: Disclose vulnerabilities privately via GitHub Security Advisories or directly via maintainer contacts listed in SECURITY.md.

---

Third-Party Licenses & Notices

connectors is built on a clean-room, zero-dependency architecture. All core messaging modules rely exclusively on the Python standard library under the PSF License.

Attribution notices and licensing terms for conceptual inspirations (OpenClaw, Hermes Agent) and development tooling (PyYAML, pytest, signal-cli) are documented in THIRD_PARTY_LICENSES.md.

---

Contributing

Contributions are welcome! Please ensure:

1. All changes adhere to the zero-dependency core principle (dependencies = [] in pyproject.toml).
2. Secrets are masked in string representations and never written to disk or logs (field(repr=False)).
3. Subprocess calls avoid shell invocation (shell=False).
4. Automated test suites and linter checks pass cleanly (pytest -v and ruff check .).

---

License

This project is licensed under the terms of the MIT License.