How to Manage Remote Teams with Slack 2026.0 and Asana 2026.0 for Engineering Leads

import os
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.response import SocketModeResponse
from asana import Client as AsanaClient
from asana.errors import AsanaError
import json
from datetime import datetime
from typing import Dict, Optional

# Configure logging for production debugging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SlackThreadResolver:
    '''Listens for resolved Slack threads and creates Asana tasks with full context'''

    def __init__(self):
        # Initialize Slack clients with 2026.0 Socket Mode support
        self.slack_web = WebClient(token=os.environ['SLACK_BOT_TOKEN'])
        self.slack_socket = SocketModeClient(
            app_token=os.environ['SLACK_APP_TOKEN'],
            web_client=self.slack_web
        )
        # Initialize Asana 2026.0 client with OAuth2 refresh
        self.asana_client = AsanaClient.access_token(os.environ['ASANA_ACCESS_TOKEN'])
        self.asana_project_id = os.environ['ASANA_PROJECT_ID']

        # Register Slack event handler for thread resolution (new in 2026.0)
        self.slack_socket.socket_mode_request_listeners.append(self.handle_thread_resolved)

    def handle_thread_resolved(self, client: SocketModeClient, req: SocketModeResponse):
        '''Process Slack 2026.0 thread_resolved event'''
        if req.type != 'events_api':
            return

        event = req.payload.get('event', {})
        if event.get('type') != 'thread_resolved':
            return

        try:
            thread_ts = event['thread_ts']
            channel_id = event['channel']
            resolver_id = event['user']

            # Fetch full thread context from Slack
            thread_messages = self.fetch_thread_messages(channel_id, thread_ts)
            # Create Asana task with thread context
            asana_task = self.create_asana_task(thread_messages, resolver_id)

            # Post confirmation to Slack thread
            self.slack_web.chat_postMessage(
                channel=channel_id,
                thread_ts=thread_ts,
                text=f'✅ Thread resolved! Asana task created: {asana_task["permalink_url"]}'
            )

            # Acknowledge Slack event to avoid retry
            req.context.ack()
            logger.info(f'Created Asana task {asana_task["gid"]} from Slack thread {thread_ts}')

        except SlackApiError as e:
            logger.error(f'Slack API error: {e.response["error"]}')
            req.context.ack()
        except AsanaError as e:
            logger.error(f'Asana API error: {e.message}')
            req.context.ack()
        except Exception as e:
            logger.error(f'Unexpected error: {str(e)}')
            req.context.ack()

    def fetch_thread_messages(self, channel_id: str, thread_ts: str) -> list:
        '''Retrieve all messages in a Slack thread with 2026.0 pagination support'''
        messages = []
        cursor = None

        while True:
            try:
                response = self.slack_web.conversations_replies(
                    channel=channel_id,
                    ts=thread_ts,
                    cursor=cursor,
                    limit=200  # Max per 2026.0 API docs
                )
                messages.extend(response['messages'])

                if not response.get('has_more'):
                    break
                cursor = response['response_metadata']['next_cursor']

            except SlackApiError as e:
                if e.response['error'] == 'thread_not_found':
                    logger.warning(f'Thread {thread_ts} not found in channel {channel_id}')
                    break
                raise

        return sorted(messages, key=lambda x: float(x['ts']))

    def create_asana_task(self, thread_messages: list, resolver_id: str) -> Dict:
        '''Create Asana 2026.0 task with full Slack thread context'''
        # Extract thread summary from first message
        thread_summary = thread_messages[0]['text'][:200] if thread_messages else 'Resolved Slack Thread'
        # Format full thread context as Asana task description
        thread_context = '\n\n'.join([
            f'**{msg.get("user", "unknown")}** ({datetime.fromtimestamp(float(msg["ts"])).strftime("%Y-%m-%d %H:%M")}):\n{msg["text"]}'
            for msg in thread_messages
        ])

        task_data = {
            'name': f'Slack Thread: {thread_summary}',
            'projects': [self.asana_project_id],
            'notes': f'Auto-generated from resolved Slack thread.\n\nFull Context:\n{thread_context}',
            'assignee': self.get_asana_user_id(resolver_id),
            'tags': ['slack-auto', '2026-orchestration']
        }

        try:
            return self.asana_client.tasks.create_task(task_data)
        except AsanaError as e:
            logger.error(f'Failed to create Asana task: {e.message}')
            raise

    def get_asana_user_id(self, slack_user_id: str) -> Optional[str]:
        '''Map Slack user ID to Asana user ID via 2026.0 identity API'''
        try:
            # Slack 2026.0 identity API returns linked Asana user
            response = self.slack_web.users_identity(
                user=slack_user_id,
                service='asana'
            )
            return response.get('asana_user_id')
        except SlackApiError:
            logger.warning(f'No Asana identity linked for Slack user {slack_user_id}')
            return None

    def start(self):
        '''Start listening for Slack events'''
        logger.info('Starting Slack Thread Resolver...')
        self.slack_socket.connect()
        # Keep socket open indefinitely
        import time
        while True:
            time.sleep(1)

if __name__ == '__main__':
    # Validate required environment variables
    required_vars = ['SLACK_BOT_TOKEN', 'SLACK_APP_TOKEN', 'ASANA_ACCESS_TOKEN', 'ASANA_PROJECT_ID']
    missing = [var for var in required_vars if not os.environ.get(var)]
    if missing:
        raise ValueError(f'Missing required environment variables: {missing}')

    resolver = SlackThreadResolver()
    resolver.start()

import os
import logging
from asana import Client as AsanaClient
from asana.errors import AsanaError
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from datetime import datetime, timedelta
from typing import List, Dict
import json

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

class AsanaSprintSync:
    '''Syncs Asana 2026.0 sprint progress to Slack channels daily'''

    def __init__(self):
        self.asana_client = AsanaClient.access_token(os.environ['ASANA_ACCESS_TOKEN'])
        self.slack_client = WebClient(token=os.environ['SLACK_BOT_TOKEN'])
        self.asana_workspace_id = os.environ['ASANA_WORKSPACE_ID']
        self.slack_channel_id = os.environ['SLACK_STANDUP_CHANNEL_ID']
        # Asana 2026.0 sprint custom field GID (replace with your own)
        self.sprint_custom_field_gid = os.environ.get('ASANA_SPRINT_FIELD_GID', '123456789')

    def fetch_active_sprints(self) -> List[Dict]:
        '''Retrieve active sprints from Asana 2026.0 using custom fields'''
        try:
            # Asana 2026.0 supports filtering by custom field values
            sprints = self.asana_client.tasks.get_tasks(
                workspace=self.asana_workspace_id,
                completed_since='now',
                opt_fields=['name', 'assignee', 'custom_fields', 'permalink_url']
            )

            active_sprints = []
            for task in sprints:
                # Check if task has sprint custom field set to active
                for field in task.get('custom_fields', []):
                    if field['gid'] == self.sprint_custom_field_gid and field.get('enum_value', {}).get('name') == 'Active':
                        active_sprints.append(task)

            logger.info(f'Found {len(active_sprints)} active sprints')
            return active_sprints

        except AsanaError as e:
            logger.error(f'Asana API error fetching sprints: {e.message}')
            raise

    def calculate_sprint_progress(self, sprint_task: Dict) -> Dict:
        '''Calculate completion percentage for a sprint using Asana 2026.0 subtask API'''
        try:
            subtasks = self.asana_client.tasks.get_subtasks(sprint_task['gid'])
            total = len(subtasks)
            if total == 0:
                return {'total': 0, 'completed': 0, 'percentage': 0}

            completed = sum(1 for sub in subtasks if sub.get('completed', False))
            percentage = round((completed / total) * 100, 1)

            return {
                'total': total,
                'completed': completed,
                'percentage': percentage,
                'name': sprint_task['name'],
                'url': sprint_task['permalink_url']
            }

        except AsanaError as e:
            logger.error(f'Error calculating progress for sprint {sprint_task["gid"]}: {e.message}')
            return {'total': 0, 'completed': 0, 'percentage': 0}

    def post_sprint_update_to_slack(self, progress_reports: List[Dict]):
        '''Post formatted sprint progress to Slack 2026.0 channel'''
        if not progress_reports:
            message = '📊 No active sprints found today.'
        else:
            # Build Slack 2026.0 block kit message
            blocks = [
                {
                    'type': 'header',
                    'text': {
                        'type': 'plain_text',
                        'text': f'📊 Sprint Progress Update: {datetime.now().strftime("%Y-%m-%d")}'
                    }
                },
                {'type': 'divider'}
            ]

            for report in progress_reports:
                if report['total'] == 0:
                    continue

                # Progress bar using Slack 2026.0 emoji scale
                progress_bar = '🟩' * int(report['percentage'] / 10) + '⬜' * (10 - int(report['percentage'] / 10))

                blocks.append({
                    'type': 'section',
                    'text': {
                        'type': 'mrkdwn',
                        'text': f'*{report["name"]}*\n{progress_bar} {report["percentage"]}%\n{report["completed"]}/{report["total"]} tasks completed\n<{report["url"]}|View in Asana>'
                    }
                })
                blocks.append({'type': 'divider'})

            message = None

        try:
            self.slack_client.chat_postMessage(
                channel=self.slack_channel_id,
                text=message,
                blocks=blocks if not message else None
            )
            logger.info(f'Posted sprint update to Slack channel {self.slack_channel_id}')

        except SlackApiError as e:
            logger.error(f'Slack API error posting update: {e.response["error"]}')
            raise

    def run_daily_sync(self):
        '''Execute full sync workflow'''
        logger.info('Starting daily Asana → Slack sprint sync...')
        try:
            active_sprints = self.fetch_active_sprints()
            progress_reports = [self.calculate_sprint_progress(sprint) for sprint in active_sprints]
            self.post_sprint_update_to_slack(progress_reports)
            logger.info('Daily sync completed successfully')
        except Exception as e:
            logger.error(f'Daily sync failed: {str(e)}')
            # Post error alert to Slack
            self.slack_client.chat_postMessage(
                channel=self.slack_channel_id,
                text=f'⚠️ Sprint sync failed: {str(e)}'
            )

if __name__ == '__main__':
    required_vars = ['ASANA_ACCESS_TOKEN', 'SLACK_BOT_TOKEN', 'ASANA_WORKSPACE_ID', 'SLACK_STANDUP_CHANNEL_ID']
    missing = [var for var in required_vars if not os.environ.get(var)]
    if missing:
        raise ValueError(f'Missing env vars: {missing}')

    sync = AsanaSprintSync()
    # Run immediately, then schedule daily (in production use cron or scheduler)
    sync.run_daily_sync()

import os
import logging
from asana import Client as AsanaClient
from asana.errors import AsanaError
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from datetime import datetime
from typing import Dict, List
import json

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

class WorkloadAlerter:
    '''Alerts Slack leads when Asana 2026.0 team workload exceeds 80% capacity'''

    def __init__(self):
        self.asana_client = AsanaClient.access_token(os.environ['ASANA_ACCESS_TOKEN'])
        self.slack_client = WebClient(token=os.environ['SLACK_BOT_TOKEN'])
        self.asana_team_id = os.environ['ASANA_TEAM_ID']
        self.slack_lead_channel_id = os.environ['SLACK_LEAD_CHANNEL_ID']
        # 2026.0 workload API endpoint (new in Asana 2026.0)
        self.workload_api = self.asana_client.workload

    def fetch_team_workload(self) -> List[Dict]:
        '''Retrieve per-engineer workload from Asana 2026.0 Workload Graph API'''
        try:
            # Asana 2026.0 Workload Graph returns capacity and assigned work hours
            workload = self.workload_api.get_team_workload(
                team_gid=self.asana_team_id,
                opt_fields=['user.name', 'user.gid', 'capacity_hours', 'assigned_hours', 'overloaded']
            )
            logger.info(f'Fetched workload for {len(workload)} team members')
            return workload

        except AsanaError as e:
            logger.error(f'Asana workload API error: {e.message}')
            raise

    def identify_overloaded_engineers(self, workload: List[Dict], threshold: float = 0.8) -> List[Dict]:
        '''Filter engineers with assigned hours > threshold * capacity'''
        overloaded = []
        for member in workload:
            capacity = member.get('capacity_hours', 0)
            assigned = member.get('assigned_hours', 0)
            if capacity == 0:
                continue

            utilization = assigned / capacity
            if utilization >= threshold:
                overloaded.append({
                    'name': member['user']['name'],
                    'asana_gid': member['user']['gid'],
                    'capacity': capacity,
                    'assigned': assigned,
                    'utilization': round(utilization * 100, 1),
                    'slack_user_id': self.get_slack_user_id(member['user']['gid'])
                })

        logger.info(f'Found {len(overloaded)} overloaded engineers (threshold {threshold*100}%)')
        return overloaded

    def get_slack_user_id(self, asana_user_gid: str) -> str:
        '''Map Asana user GID to Slack user ID via 2026.0 directory API'''
        try:
            # Asana 2026.0 directory API returns linked Slack identities
            user = self.asana_client.users.get_user(asana_user_gid, opt_fields=['linked_accounts'])
            for account in user.get('linked_accounts', []):
                if account['service'] == 'slack':
                    return account['service_user_id']
            return None

        except AsanaError as e:
            logger.warning(f'Could not fetch Slack ID for Asana user {asana_user_gid}: {e.message}')
            return None

    def send_overload_alerts(self, overloaded: List[Dict]):
        '''Send Slack 2026.0 alerts to leads and overloaded engineers'''
        if not overloaded:
            logger.info('No overloaded engineers, no alerts sent')
            return

        # Alert leads first
        lead_message = f'⚠️ *Workload Alert*: {len(overloaded)} engineers over 80% capacity:\n'
        for eng in overloaded:
            lead_message += f'- {eng["name"]}: {eng["utilization"]}% utilization ({eng["assigned"]}/{eng["capacity"]} hours)\n'

        try:
            self.slack_client.chat_postMessage(
                channel=self.slack_lead_channel_id,
                text=lead_message
            )
            logger.info(f'Sent lead alert to channel {self.slack_lead_channel_id}')
        except SlackApiError as e:
            logger.error(f'Failed to send lead alert: {e.response["error"]}')

        # Send individual alerts to engineers
        for eng in overloaded:
            if not eng['slack_user_id']:
                logger.warning(f'No Slack ID for {eng["name"]}, skipping individual alert')
                continue

            eng_message = f'Hi {eng["name"]}, you’re currently at {eng["utilization"]}% workload capacity ({eng["assigned"]}/{eng["capacity"]} hours). Please review your Asana tasks and let leads know if you need to reassign work.'
            try:
                self.slack_client.chat_postMessage(
                    channel=eng['slack_user_id'],  # DM in Slack 2026.0
                    text=eng_message
                )
                logger.info(f'Sent individual alert to {eng["name"]}')
            except SlackApiError as e:
                logger.error(f'Failed to send alert to {eng["name"]}: {e.response["error"]}')

    def run_check(self):
        '''Execute full workload check'''
        logger.info('Starting workload check...')
        try:
            workload = self.fetch_team_workload()
            overloaded = self.identify_overloaded_engineers(workload)
            self.send_overload_alerts(overloaded)
            logger.info('Workload check completed')
        except Exception as e:
            logger.error(f'Workload check failed: {str(e)}')
            self.slack_client.chat_postMessage(
                channel=self.slack_lead_channel_id,
                text=f'⚠️ Workload check failed: {str(e)}'
            )

if __name__ == '__main__':
    required_vars = ['ASANA_ACCESS_TOKEN', 'SLACK_BOT_TOKEN', 'ASANA_TEAM_ID', 'SLACK_LEAD_CHANNEL_ID']
    missing = [var for var in required_vars if not os.environ.get(var)]
    if missing:
        raise ValueError(f'Missing env vars: {missing}')

    alerter = WorkloadAlerter()
    alerter.run_check()

# Idempotency check for Asana task creation
existing_tasks = self.asana_client.tasks.get_tasks(
    project=self.asana_project_id,
    tags=['slack-auto'],
    opt_fields=['name', 'notes']
)
thread_ts_tag = f'slack-thread-{thread_ts}'
for task in existing_tasks:
    if thread_ts_tag in task.get('notes', ''):
        logger.info(f'Task already exists for thread {thread_ts}')
        return task

# Fetch overloaded flag directly from Asana 2026.0 Workload API
workload = self.asana_client.workload.get_team_workload(team_gid=self.asana_team_id)
for member in workload:
    if member.get('overloaded', False):
        # Send alert immediately
        self.send_overload_alerts([member])

# Disable Zapier workflow via API (if using Zapier)
import requests
zapier_api_key = os.environ['ZAPIER_API_KEY']
response = requests.post(
    f'https://api.zapier.com/v1/zaps/{zap_id}/deactivate',
    headers={'Authorization': f'Bearer {zapier_api_key}'}
)

slack-asana-2026-orchestrator/
├── README.md                # Setup instructions, benchmark results
├── requirements.txt         # Python dependencies (slack-sdk==2026.0.1, asana==2026.0.0)
├── .env.example             # Example environment variables
├── src/
│   ├── __init__.py
│   ├── slack_thread_resolver.py  # Code example 1
│   ├── asana_sprint_sync.py      # Code example 2
│   ├── workload_alerter.py       # Code example 3
│   └── utils.py                  # Shared utilities (idempotency, mapping)
├── tests/
│   ├── test_slack_resolver.py    # Unit tests for thread resolver
│   ├── test_asana_sync.py        # Unit tests for sprint sync
│   └── test_workload.py          # Unit tests for alerter
└── benchmarks/
    ├── latency_results.json      # p99 latency benchmarks
    └── cost_savings.csv          # Monthly cost savings data