Slack Plugin

Team notifications and collaboration with Slack messaging, threads, and file sharing. Built on slack-sdk.

Installation

pip install slack-sdk

Quick Start

Direct Usage (Scripts)

from daita.plugins import slack

# Direct usage in scripts
async with slack(token="xoxb-your-bot-token") as slack_client:
    await slack_client.send_message("#alerts", "System update complete")

Agent Integration (Recommended)

from daita import SubstrateAgent
from daita.plugins import slack

# Create plugin
slack_client = slack(token="xoxb-your-bot-token")

# Agent uses Slack tools autonomously
agent = SubstrateAgent(
    name="Notification Agent",
    prompt="You are a notification specialist. Help users send messages to Slack.",
    tools=[slack_client]
)

await agent.start()
result = await agent.run("Send a message to #alerts that the deployment is complete")

Connection Parameters

slack(
    token: str,
    bot_user_oauth_token: Optional[str] = None,
    app_token: Optional[str] = None,
    default_channel: Optional[str] = None,
    **kwargs
)

Parameters

• token (str): Slack bot token (xoxb-...) - required
• bot_user_oauth_token (str): Bot user OAuth token for extended permissions (optional)
• app_token (str): App-level token for Socket Mode (optional)
• default_channel (str): Default channel for messages (optional)
• **kwargs: Additional Slack client parameters

Connection Methods

# Basic bot token
async with slack(token="xoxb-your-bot-token") as slack:
    await slack.send_message("#general", "Hello!")

# With default channel
async with slack(
    token="xoxb-your-bot-token",
    default_channel="#alerts"
) as slack:
    await slack.send_message(None, "Using default channel")

# From environment variable (recommended)
import os
async with slack(token=os.getenv("SLACK_BOT_TOKEN")) as slack:
    await slack.send_message("#team", "Secure notification")

Sending Messages

from daita.agents import SubstrateAgent

async def main():
    agent = SubstrateAgent(name="Agent")

    async with slack(token="xoxb-token") as slack:
        # Simple text message
        result = await slack.send_message("#alerts", "Deployment complete")

        # Message with metadata
        result = await slack.send_message(
            channel="#team",
            text="Agent finished processing",
            thread_ts=None,
            reply_broadcast=False
        )

        print(f"Message sent: {result['ts']}")

Rich Formatting with Blocks

async with slack(token="xoxb-token") as slack:
    # Block Kit formatting
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "Deployment Status"}
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": "*Environment:* Production"},
                {"type": "mrkdwn", "text": "*Status:* ✅ Success"}
            ]
        }
    ]

    await slack.send_message("#deploys", text="Deployment", blocks=blocks)

Thread Management

async with slack(token="xoxb-token") as slack:
    # Send initial message
    result = await slack.send_message("#team", "Starting data processing")
    thread_ts = result['ts']

    # Reply in thread
    await slack.send_message(
        channel="#team",
        text="Step 1 complete",
        thread_ts=thread_ts
    )

    # Reply with broadcast to channel
    await slack.send_message(
        channel="#team",
        text="Processing complete!",
        thread_ts=thread_ts,
        reply_broadcast=True
    )

Agent-Specific Features

Send agent results:

async with slack(token="xoxb-token") as slack:
    agent_results = {
        "status": "success",
        "duration_ms": 1234,
        "output": {"records_processed": 1000},
        "start_time": "2024-01-15T10:00:00",
        "end_time": "2024-01-15T10:00:01"
    }

    await slack.send_agent_summary(
        channel="#data-team",
        agent_results=agent_results,
        title="Data Processing Agent"
    )

Create workflow thread:

async with slack(token="xoxb-token") as slack:
    workflow_results = {
        "agents": {
            "agent_a": {"status": "success", "output": {"count": 100}},
            "agent_b": {"status": "success", "output": {"count": 200}}
        }
    }

    result = await slack.create_thread_from_workflow(
        channel="#workflows",
        workflow_results=workflow_results,
        title="Daily Pipeline Results"
    )

    print(f"Thread created: {result['thread_ts']}")

File Upload

async with slack(token="xoxb-token") as slack:
    # Upload file
    result = await slack.upload_file(
        channel="#reports",
        file_path="/tmp/analysis.pdf",
        title="Monthly Analysis Report",
        initial_comment="Here's the latest analysis"
    )

    # Upload to thread
    result = await slack.upload_file(
        channel="#reports",
        file_path="/tmp/data.csv",
        title="Processed Data",
        thread_ts="1234567890.123456"
    )

Channel History

async with slack(token="xoxb-token") as slack:
    # Get recent messages
    messages = await slack.get_channel_history("#alerts", limit=50)

    for msg in messages:
        print(f"{msg['user']}: {msg['text']}")

    # Get messages with time filter
    messages = await slack.get_channel_history(
        channel="#alerts",
        limit=100,
        oldest="1609459200.000000",  # Unix timestamp
        latest="1609545600.000000"
    )

Channel Management

async with slack(token="xoxb-token") as slack:
    # List all channels
    channels = await slack.get_channels()

    for channel in channels:
        print(f"{channel['name']}: {channel['num_members']} members")

    # Filter to public channels only
    public_channels = await slack.get_channels(types="public_channel")

Using with Agents

Direct Messaging Operations (Scripts)

For scripts that don't need agent capabilities:

from daita.plugins import slack
import os

async with slack(token=os.getenv("SLACK_BOT_TOKEN")) as slack_client:
    # Notify start
    result = await slack_client.send_message("#processing", "Starting data processing")
    thread_ts = result['ts']

    # Process data
    processed_count = 100  # Your processing logic here

    # Notify completion in thread
    await slack_client.send_message(
        channel="#processing",
        text=f"Processed {processed_count} records",
        thread_ts=thread_ts
    )

    print(f"Notifications sent: {processed_count} records processed")

Tool-Based Integration (Recommended)

Slack plugin exposes messaging operations as tools that agents can use autonomously:

from daita import SubstrateAgent
from daita.plugins import slack
import os

# Create Slack plugin
slack_client = slack(token=os.getenv("SLACK_BOT_TOKEN"))

# Pass plugin to agent - agent can now use Slack tools autonomously
agent = SubstrateAgent(
    name="Notification Agent",
    prompt="You are a notification specialist. Help users send messages and updates to Slack.",
    llm_provider="openai",
    model="gpt-4",
    tools=[slack_client]
)

await agent.start()

# Agent autonomously uses Slack tools to answer questions
result = await agent.run("Send a summary of today's metrics to #analytics")

# The agent will autonomously:
# 1. Use send_slack_message or send_slack_summary tool
# 2. Format the message appropriately
# 3. Send to the correct channel

await agent.stop()

Available Tools

The Slack plugin exposes these tools to LLM agents:

Tool	Description	Parameters
send_slack_message	Send message to channel	channel (required), text (required)
send_slack_summary	Send formatted agent summary	channel (required), summary (required), results (object)
list_slack_channels	List available channels	None

Tool Categories: communication Tool Source: plugin Bot Token: Configured at plugin initialization

Tool Usage Example

from daita import SubstrateAgent
from daita.plugins import slack
import os

# Setup Slack with tool integration
slack_client = slack(token=os.getenv("SLACK_BOT_TOKEN"))

agent = SubstrateAgent(
    name="Alert System",
    prompt="You are an alert system. Help users send notifications to Slack channels.",
    llm_provider="openai",
    model="gpt-4",
    tools=[slack_client]
)

await agent.start()

# Natural language command - agent uses tools autonomously
result = await agent.run("""
Monitor system status and notify team:
1. Check all available channels
2. Send deployment status to #ops
3. Post error summary to #alerts if errors found
""")

# Agent orchestrates Slack tool calls to send notifications
print(result)
await agent.stop()

Error Handling

from daita.agents import SubstrateAgent

agent = SubstrateAgent(name="Agent")

try:
    async with slack(token="xoxb-token") as slack:
        await slack.send_message("#alerts", "Test message")
except RuntimeError as e:
    if "slack-sdk not installed" in str(e):
        print("Install slack-sdk: pip install slack-sdk")
    elif "Invalid Slack token" in str(e):
        print("Check your bot token")
    elif "invalid_auth" in str(e):
        print("Authentication failed - verify token")

Best Practices

Security:

• Store bot tokens in environment variables, never hardcode
• Use bot tokens (xoxb-) instead of user tokens for automation
• Rotate tokens regularly
• Restrict bot permissions to minimum required scopes

Performance:

• Use threads to organize related messages
• Batch notifications when possible to avoid rate limits
• Cache channel IDs to reduce API calls
• Use Block Kit for rich formatting instead of attachments

Message Design:

• Keep messages concise and actionable
• Use threads for detailed information
• Include context (agent name, timestamp, etc.)
• Use emojis and formatting for better visibility

Error Handling:

• Always handle connection failures gracefully
• Validate channel names before sending
• Check file sizes before uploading (max 1GB)
• Implement retry logic for transient failures

Block Kit Examples

Status notification:

blocks = [
    {"type": "header", "text": {"type": "plain_text", "text": "🚀 Deployment Status"}},
    {
        "type": "section",
        "fields": [
            {"type": "mrkdwn", "text": "*Environment:* Production"},
            {"type": "mrkdwn", "text": "*Status:* ✅ Success"},
            {"type": "mrkdwn", "text": "*Duration:* 2m 34s"},
            {"type": "mrkdwn", "text": "*Deployed by:* CI/CD"}
        ]
    }
]
await slack.send_message("#deploys", "Deployment", blocks=blocks)

Alert with actions:

blocks = [
    {
        "type": "section",
        "text": {"type": "mrkdwn", "text": "⚠️ *High CPU Usage Detected*\nServer: web-01\nCPU: 95%"}
    },
    {
        "type": "actions",
        "elements": [
            {"type": "button", "text": {"type": "plain_text", "text": "View Logs"}, "value": "view_logs"},
            {"type": "button", "text": {"type": "plain_text", "text": "Restart"}, "value": "restart"}
        ]
    }
]
await slack.send_message("#ops", "Alert", blocks=blocks)

Getting Your Bot Token

1. Go to https://api.slack.com/apps
2. Create a new app or select existing app
3. Navigate to "OAuth & Permissions"
4. Add required bot token scopes:
   • chat:write - Send messages
   • files:write - Upload files
   • channels:history - Read channel history
   • channels:read - List channels
5. Install app to workspace
6. Copy the "Bot User OAuth Token" (starts with xoxb-)

Troubleshooting

Issue	Solution
slack-sdk not installed	pip install slack-sdk
Invalid Slack token	Verify bot token starts with xoxb-
invalid_auth	Check token is correct and app is installed
channel_not_found	Verify channel name/ID, ensure bot is invited
not_in_channel	Invite bot to channel: /invite @your-bot
rate_limited	Reduce message frequency, implement backoff
file_too_large	Compress file or split into multiple files

Common Patterns

Daily digest:

async with slack(token=os.getenv("SLACK_BOT_TOKEN")) as slack:
    summary = generate_daily_summary()
    await slack.send_message("#daily-digest", summary)

Error notifications:

async with slack(token=os.getenv("SLACK_BOT_TOKEN")) as slack:
    try:
        process_data()
    except Exception as e:
        await slack.send_message("#errors", f"⚠️ Error: {str(e)}")
        raise

Workflow progress:

async with slack(token=os.getenv("SLACK_BOT_TOKEN")) as slack:
    msg = await slack.send_message("#workflows", "Starting workflow")
    thread_ts = msg['ts']

    for step in workflow_steps:
        await slack.send_message(
            "#workflows",
            f"✓ Completed: {step}",
            thread_ts=thread_ts
        )

Next Steps

• Plugin Overview - All available plugins
• Redis Plugin - Distributed messaging
• REST Plugin - API integrations