How to Connect a Python Agent to Slack with Arcade (MCP)

AI agents need direct access to Slack workspaces to read messages, send notifications, and manage conversations. Arcade provides OAuth-backed authentication and a complete Slack toolkit through the Model Context Protocol (MCP), eliminating the need to build authentication infrastructure from scratch.

This guide covers the technical implementation of Python agents that interact with Slack through Arcade's authentication layer and pre-built tools.

Prerequisites

Python 3.10+
Arcade account with API key
Slack workspace admin access
Async Python knowledge

Installation

Install the Arcade Python SDK:

pip install arcadepy

For toolkit development and CLI access:

pip install arcade-ai

Set your API key:

export ARCADE_API_KEY="your_api_key_here"

Get your API key from the Arcade dashboard.

Slack App Configuration

Create Slack Application

Navigate to https://api.slack.com/apps
Click "Create New App" and select "From scratch"
Name your app and select your workspace
Go to "OAuth & Permissions"

Configure OAuth Scopes

Add these User Token Scopes:

channels:history
channels:read
chat:write
groups:read
groups:history
groups:write
im:history
im:read
im:write
mpim:history
mpim:read
mpim:write
users:read
users:read.email

These scopes enable the full Arcade Slack toolkit.

Set Redirect URL

For Arcade Cloud:

https://api.arcade.dev/v1/auth/callback

Self-hosted deployments require your custom domain.

Add Provider to Arcade

In the Arcade dashboard:

Navigate to OAuth > Providers
Click "Add OAuth Provider"
Select Slack from dropdown
Enter Client ID and Client Secret
Copy the generated redirect URL to your Slack app
Save configuration

Reference: Slack auth provider documentation

Basic Implementation

Send Slack Message

import asyncio
from arcadepy import Arcade

async def send_message():
    client = Arcade()
    user_id = "user@company.com"

    # Check authorization
    auth = await client.tools.authorize(
        tool_name="Slack.SendMessage",
        user_id=user_id
    )

    if auth.status != "completed":
        print(f"Authorize at: {auth.url}")
        await client.auth.wait_for_completion(auth)

    # Execute tool
    result = await client.tools.execute(
        tool_name="Slack.SendMessage",
        input={
            "channel_name": "general",
            "message": "Message from Python agent"
        },
        user_id=user_id
    )

    return result.output

asyncio.run(send_message())

Read Channel Messages

async def read_messages():
    client = Arcade()
    user_id = "user@company.com"

    result = await client.tools.execute(
        tool_name="Slack.GetMessages",
        input={
            "channel_name": "engineering",
            "limit": 20,
            "oldest_relative": "02:00:00"  # Last 2 hours
        },
        user_id=user_id
    )

    return result.output.get("messages", [])

List Workspace Users

async def list_users():
    client = Arcade()
    user_id = "user@company.com"

    result = await client.tools.execute(
        tool_name="Slack.ListUsers",
        input={
            "exclude_bots": True,
            "limit": 100
        },
        user_id=user_id
    )

    return result.output

Multi-User Agent Architecture

Production agents serve multiple users with isolated authentication:

from arcadepy import Arcade
from typing import Dict, Optional

class SlackAgent:
    def __init__(self):
        self.client = Arcade()
        self.sessions: Dict[str, bool] = {}

    async def authorize_user(self, user_id: str) -> Optional[str]:
        """Returns auth URL if authorization needed, None if complete"""
        if user_id in self.sessions:
            return None

        auth = await self.client.tools.authorize(
            tool_name="Slack.SendMessage",
            user_id=user_id
        )

        if auth.status != "completed":
            return auth.url

        self.sessions[user_id] = True
        return None

    async def send_message(
        self,
        user_id: str,
        channel: str,
        message: str
    ) -> dict:
        auth_url = await self.authorize_user(user_id)
        if auth_url:
            return {
                "error": "authorization_required",
                "url": auth_url
            }

        result = await self.client.tools.execute(
            tool_name="Slack.SendMessage",
            input={"channel_name": channel, "message": message},
            user_id=user_id
        )

        return {"success": True, "data": result.output}

    async def get_conversation_metadata(
        self,
        user_id: str,
        channel: str
    ) -> dict:
        result = await self.client.tools.execute(
            tool_name="Slack.GetConversationMetadata",
            input={"channel_name": channel},
            user_id=user_id
        )

        return result.output

Available Tools

The Slack toolkit provides these tools:

User Operations

Slack.WhoAmI - Get authenticated user profile
Slack.GetUsersInfo - Retrieve users by ID, username, or email
Slack.ListUsers - List workspace members
Slack.GetUsersInConversation - Get conversation participants

Messaging Operations

Slack.SendMessage - Send to channels or DMs
Slack.GetMessages - Retrieve conversation history with time filters

Conversation Operations

Slack.GetConversationMetadata - Get channel or DM metadata
Slack.ListConversations - List user's conversations

All tools support pagination via next_cursor parameter.

Framework Integration

LangChain

from langchain_arcade import ArcadeToolManager
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

manager = ArcadeToolManager(api_key="your_key")
slack_tools = manager.get_tools(toolkits=["Slack"])

llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You manage Slack communications"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

agent = create_openai_functions_agent(llm, slack_tools, prompt)
executor = AgentExecutor(agent=agent, tools=slack_tools)

result = executor.invoke({
    "input": "Send standup reminder to engineering channel"
})

Reference: LangChain integration guide

CrewAI

from crewai import Agent, Task, Crew
from crewai_arcade import ArcadeToolManager

manager = ArcadeToolManager(default_user_id="user@company.com")
tools = manager.get_tools(toolkits=["Slack"])

agent = Agent(
    role="Slack Manager",
    goal="Handle team notifications",
    backstory="Manages Slack communications",
    tools=tools
)

task = Task(
    description="Send deployment notification to engineering",
    expected_output="Message confirmation",
    agent=agent
)

crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()

Reference: CrewAI integration guide

Error Handling

Handle authorization and execution errors:

from arcadepy import Arcade

async def safe_execute(
    client: Arcade,
    tool_name: str,
    user_id: str,
    params: dict
):
    try:
        result = await client.tools.execute(
            tool_name=tool_name,
            input=params,
            user_id=user_id
        )
        return {"success": True, "data": result.output}

    except Exception as e:
        if hasattr(e, 'type'):
            if e.type == "authorization_required":
                return {
                    "error": "auth_required",
                    "url": e.url
                }
            elif e.type == "rate_limit":
                return {
                    "error": "rate_limit",
                    "retry_after": getattr(e, 'retry_after', 60)
                }

        return {"error": "execution_failed", "message": str(e)}

Production API Implementation

FastAPI endpoint with proper authorization flow:

from fastapi import FastAPI, HTTPException
from arcadepy import Arcade
from pydantic import BaseModel

app = FastAPI()
arcade = Arcade()

class MessageRequest(BaseModel):
    user_id: str
    channel: str
    message: str

@app.post("/slack/message")
async def send_message(req: MessageRequest):
    try:
        result = await arcade.tools.execute(
            tool_name="Slack.SendMessage",
            input={
                "channel_name": req.channel,
                "message": req.message
            },
            user_id=req.user_id
        )
        return {"success": True, "result": result.output}

    except Exception as e:
        if hasattr(e, 'type') and e.type == "authorization_required":
            return {
                "authorization_required": True,
                "auth_url": e.url
            }
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/slack/callback")
async def oauth_callback(user_id: str, auth_id: str):
    result = await arcade.auth.wait_for_completion(auth_id)

    if result.status == "completed":
        return {"success": True}

    return {"success": False, "error": "Authorization failed"}

Custom Tool Development

Build custom Slack tools with the Tool Development Kit:

Initialize Toolkit

arcade new slack_custom
cd slack_custom

Create Custom Tool

from typing import Annotated
from arcade_tdk import ToolContext, tool
from arcade_tdk.auth import Slack
from slack_sdk import WebClient

@tool(
    requires_auth=Slack(
        scopes=["chat:write", "users:read", "channels:read"]
    )
)
async def send_formatted_notification(
    context: ToolContext,
    channel: Annotated[str, "Channel name"],
    title: Annotated[str, "Notification title"],
    message: Annotated[str, "Message content"]
) -> dict:
    """Send formatted notification with blocks"""

    client = WebClient(token=context.authorization.token)

    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": title}
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": message}
        }
    ]

    response = client.chat_postMessage(
        channel=channel,
        blocks=blocks
    )

    return {
        "ts": response["ts"],
        "channel": response["channel"]
    }

Deploy Tool

# Local testing
arcade serve --port 8002

# Deploy to Arcade Cloud
arcade deploy

Reference: Tool development guide

Performance Optimization

Caching Implementation

from collections import OrderedDict
import time

class ToolCache:
    def __init__(self, ttl: int = 300, size: int = 100):
        self.cache = OrderedDict()
        self.ttl = ttl
        self.size = size

    def get(self, key: str):
        if key not in self.cache:
            return None

        data, ts = self.cache[key]
        if time.time() - ts > self.ttl:
            del self.cache[key]
            return None

        self.cache.move_to_end(key)
        return data

    def set(self, key: str, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = (value, time.time())

        if len(self.cache) > self.size:
            self.cache.popitem(last=False)

cache = ToolCache()

async def get_user_cached(client, user_id, username):
    key = f"user:{username}"
    cached = cache.get(key)

    if cached:
        return cached

    result = await client.tools.execute(
        tool_name="Slack.GetUsersInfo",
        input={"usernames": [username]},
        user_id=user_id
    )

    cache.set(key, result.output)
    return result.output

Batch Operations

import asyncio

async def batch_send(
    client: Arcade,
    user_id: str,
    messages: list[dict]
):
    tasks = [
        client.tools.execute(
            tool_name="Slack.SendMessage",
            input=msg,
            user_id=user_id
        )
        for msg in messages
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    return {
        "sent": sum(1 for r in results if not isinstance(r, Exception)),
        "failed": sum(1 for r in results if isinstance(r, Exception)),
        "results": [r for r in results if not isinstance(r, Exception)]
    }

Rate Limit Handling

Slack enforces rate limits on API calls:

async def execute_with_retry(
    client: Arcade,
    tool_name: str,
    params: dict,
    user_id: str,
    max_retries: int = 3
):
    for attempt in range(max_retries):
        try:
            return await client.tools.execute(
                tool_name=tool_name,
                input=params,
                user_id=user_id
            )
        except Exception as e:
            if hasattr(e, 'type') and e.type == "rate_limit":
                if attempt == max_retries - 1:
                    raise
                wait_time = getattr(e, 'retry_after', 60)
                await asyncio.sleep(wait_time)
            else:
                raise

Testing

Unit Tests

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_send_message():
    with patch('arcadepy.Arcade') as mock:
        client = mock.return_value
        client.tools.execute = AsyncMock(
            return_value=type('R', (), {
                'output': {'success': True}
            })()
        )

        result = await client.tools.execute(
            tool_name="Slack.SendMessage",
            input={"channel_name": "test", "message": "test"},
            user_id="test@test.com"
        )

        assert result.output['success']

CLI Testing

# Interactive testing
arcade chat

# Inspect tools
arcade show Slack.SendMessage

Reference: Testing documentation

Monitoring

Track agent performance:

import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MonitoredAgent:
    def __init__(self):
        self.client = Arcade()
        self.metrics = {
            "calls": 0,
            "success": 0,
            "failed": 0
        }

    async def execute_monitored(
        self,
        tool_name: str,
        user_id: str,
        params: dict
    ):
        start = datetime.now()
        self.metrics["calls"] += 1

        try:
            result = await self.client.tools.execute(
                tool_name=tool_name,
                input=params,
                user_id=user_id
            )

            duration = (datetime.now() - start).total_seconds()
            self.metrics["success"] += 1

            logger.info(f"{tool_name}: {duration}s")
            return result

        except Exception as e:
            self.metrics["failed"] += 1
            logger.error(f"{tool_name} failed: {e}")
            raise

    def get_metrics(self):
        total = self.metrics["calls"]
        rate = (self.metrics["success"] / total * 100) if total > 0 else 0
        return {**self.metrics, "success_rate": f"{rate:.1f}%"}

Production Checklist

Configure custom Slack OAuth app
Implement user verifier for production
Add error handling and logging
Implement rate limit handling
Set up monitoring and alerts
Test multi-user scenarios
Document required scopes
Configure environment URLs
Add token refresh recovery
Test authorization flows

Troubleshooting

Authorization Issues

Verify redirect URL matches exactly in Slack app and Arcade configuration. Ensure proper scopes are configured.

Rate Limits

Implement exponential backoff and respect retry_after values in rate limit errors.

Missing User Context

Always provide user_id parameter:

# Correct
await client.tools.execute(
    tool_name="Slack.SendMessage",
    input={"channel_name": "general", "message": "test"},
    user_id="user@company.com"  # Required
)

Resources

This implementation provides a production-ready foundation for Python agents that interact with Slack through Arcade's OAuth-backed authentication and MCP protocol support.

How to Connect Python Agent to Slack with Arcade (MCP)