refactor: use /slack/install URL and remove MCP fallback logic

Replace the deep-link authorize URL with the standard /slack/install
path for the App Home connect link. Remove the MCP server try/except
fallback from run_casey() to avoid silently dropping all MCP tools on
connection errors. Clean up unused AuthorizeUrlGenerator.

Author: mwbrooks

pydantic-ai/agent/casey.py

@@ -1,6 +1,7 @@
+import logging
 import os
 
-from pydantic_ai import Agent
+from pydantic_ai import Agent, RunContext
 from pydantic_ai.mcp import MCPServerStreamableHTTP
 
 from agent.deps import CaseyDeps
@@ -76,6 +77,8 @@
 - If unsure about a user's issue, ask clarifying questions before taking action
 """
 
+logger = logging.getLogger(__name__)
+
 _cached_model: str | None = None
 
 
@@ -117,16 +120,43 @@ def get_model() -> str:
 )
 
 
+@casey_agent.system_prompt
+def slack_mcp_prompt(ctx: RunContext[CaseyDeps]) -> str | None:
+    """Append Slack MCP Server instructions when the user token is available."""
+    if not ctx.deps.user_token:
+        return None
+    return """\
+
+## SLACK MCP SERVER
+You have access to the Slack MCP Server, which gives you powerful Slack tools beyond \
+your built-in IT helpdesk tools. Use them whenever they would help the user.
+
+Available capabilities:
+- **Search**: Search messages and files across public channels, search for channels by name
+- **Read**: Read channel message history, read thread replies, read canvas documents
+- **Write**: Send messages, create draft messages, schedule messages for later
+- **Canvases**: Create, read, and update Slack canvas documents
+
+Use these tools proactively when they can help resolve an IT issue — for example, \
+searching for related reports from other users, checking a channel for outage updates, \
+or creating a canvas to document a solution. Also use them when the user explicitly \
+asks you to perform a Slack action like sending a message or creating a canvas.
+"""
+
+
 def run_casey(text, deps, message_history=None):
     """Run the Casey agent, optionally connecting to the Slack MCP server."""
     toolsets = []
     if deps.user_token:
+        logger.info("Slack MCP Server enabled (user_token present)")
         toolsets.append(
             MCPServerStreamableHTTP(
                 SLACK_MCP_URL,
                 headers={"Authorization": f"Bearer {deps.user_token}"},
             )
         )
+    else:
+        logger.info("Slack MCP Server disabled (no user_token)")
 
     return casey_agent.run_sync(
         text,

pydantic-ai/listeners/events/app_home_opened.py

@@ -1,5 +1,6 @@
 import os
 from logging import Logger
+from urllib.parse import urljoin
 
 from slack_bolt import BoltContext
 from slack_sdk import WebClient
@@ -11,11 +12,11 @@ def handle_app_home_opened(client: WebClient, context: BoltContext, logger: Logg
     """Publish the App Home view when a user opens the app's Home tab."""
     try:
         user_id = context.user_id
-        authorize_url = None
+        install_url = None
         is_connected = False
 
         if os.environ.get("SLACK_CLIENT_ID"):
-            from oauth import authorize_url_generator, installation_store, state_store
+            from oauth import installation_store
 
             installation = installation_store.find_installation(
                 enterprise_id=context.enterprise_id or "",
@@ -25,11 +26,11 @@ def handle_app_home_opened(client: WebClient, context: BoltContext, logger: Logg
             if installation and installation.user_token:
                 is_connected = True
             else:
-                state = state_store.issue()
-                authorize_url = authorize_url_generator.generate(state)
+                redirect_uri = os.environ.get("SLACK_REDIRECT_URI", "")
+                install_url = urljoin(redirect_uri, "/slack/install")
 
         view = build_app_home_view(
-            authorize_url=authorize_url, is_connected=is_connected
+            install_url=install_url, is_connected=is_connected
         )
         client.views_publish(user_id=user_id, view=view)
     except Exception as e:

pydantic-ai/listeners/events/app_mentioned.py

@@ -7,6 +7,7 @@
 from agent import CaseyDeps, run_casey
 from thread_context import conversation_store
 from listeners.views.feedback_builder import build_feedback_blocks
+from oauth import installation_store
 
 
 def handle_app_mentioned(
@@ -58,14 +59,25 @@ def handle_app_mentioned(
         # Get conversation history
         history = conversation_store.get_history(channel_id, thread_ts)
 
+        # Look up the user token directly from the installation store rather
+        # than context.user_token. Bolt's InstallationStoreAuthorize may fail
+        # to resolve the user token when installer-latest is missing or the
+        # requesting user differs from the original installer.
+        installation = installation_store.find_installation(
+            enterprise_id=context.enterprise_id,
+            team_id=context.team_id,
+            user_id=user_id,
+        )
+        user_token = installation.user_token if installation else None
+
         # Run the agent
         deps = CaseyDeps(
             client=client,
             user_id=user_id,
             channel_id=channel_id,
             thread_ts=thread_ts,
             message_ts=event["ts"],
-            user_token=context.user_token,
+            user_token=user_token,
         )
         result = run_casey(cleaned_text, deps, message_history=history)
 

pydantic-ai/listeners/events/message.py

@@ -6,6 +6,7 @@
 from agent import CaseyDeps, run_casey
 from thread_context import conversation_store
 from listeners.views.feedback_builder import build_feedback_blocks
+from oauth import installation_store
 
 
 def handle_message(
@@ -81,14 +82,25 @@ def handle_message(
             ],
         )
 
+        # Look up the user token directly from the installation store rather
+        # than context.user_token. Bolt's InstallationStoreAuthorize may fail
+        # to resolve the user token when installer-latest is missing or the
+        # requesting user differs from the original installer.
+        installation = installation_store.find_installation(
+            enterprise_id=context.enterprise_id,
+            team_id=context.team_id,
+            user_id=user_id,
+        )
+        user_token = installation.user_token if installation else None
+
         # Run the agent
         deps = CaseyDeps(
             client=client,
             user_id=user_id,
             channel_id=channel_id,
             thread_ts=thread_ts,
             message_ts=event["ts"],
-            user_token=context.user_token,
+            user_token=user_token,
         )
         result = run_casey(text, deps, message_history=history)
 

pydantic-ai/listeners/views/app_home_builder.py

@@ -28,16 +28,15 @@
 
 
 def build_app_home_view(
-    authorize_url: str | None = None, is_connected: bool = False
+    install_url: str | None = None, is_connected: bool = False
 ) -> dict:
     """Build the App Home Block Kit view with category buttons.
 
     Args:
-        authorize_url: OAuth authorize URL. When provided, the user has not
-            connected and will see a "Connect" URL button.
-        is_connected: When ``True``, the user is connected and will see a
-            "Disconnect" action button. When both params are default, the
-            OAuth section is hidden (app.py mode).
+        install_url: OAuth install URL. When provided, the user has not
+            connected and will see a link to install.
+        is_connected: When ``True``, the user is connected and the MCP
+            status section shows as connected.
     """
     blocks = [
         {
@@ -109,13 +108,13 @@ def build_app_home_view(
                 ],
             }
         )
-    elif authorize_url:
+    elif install_url:
         blocks.append(
             {
                 "type": "section",
                 "text": {
                     "type": "mrkdwn",
-                    "text": f"🔴 *Slack MCP Server is disconnected.* <{authorize_url}|Connect the Slack MCP Server.>",
+                    "text": f"🔴 *Slack MCP Server is disconnected.* <{install_url}|Connect the Slack MCP Server.>",
                 },
             }
         )

pydantic-ai/manifest_oauth.json

@@ -30,7 +30,17 @@
         "groups:read",
         "im:history",
         "mpim:history",
-        "users:read"
+        "users:read",
+        "search:read.public",
+        "search:read.private",
+        "search:read.mpim",
+        "search:read.im",
+        "search:read.files",
+        "search:read.users",
+        "chat:write",
+        "canvases:read",
+        "canvases:write",
+        "users:read.email"
       ],
       "bot": [
         "app_mentions:read",

pydantic-ai/oauth.py

@@ -6,7 +6,6 @@
 from slack_bolt.authorization.authorize_result import AuthorizeResult
 from slack_bolt.oauth.oauth_settings import OAuthSettings
 from slack_sdk import WebClient
-from slack_sdk.oauth import AuthorizeUrlGenerator
 from slack_sdk.oauth.installation_store import FileInstallationStore
 from slack_sdk.oauth.state_store import FileOAuthStateStore
 
@@ -28,21 +27,13 @@
 
 installation_store = FileInstallationStore(
     base_dir="./data/installations",
-    historical_data_enabled=False,
 )
 
 state_store = FileOAuthStateStore(
     expiration_seconds=600,
     base_dir="./data/states",
 )
 
-authorize_url_generator = AuthorizeUrlGenerator(
-    client_id=os.environ.get("SLACK_CLIENT_ID", ""),
-    redirect_uri=os.environ.get("SLACK_REDIRECT_URI", ""),
-    scopes=BOT_SCOPES,
-    user_scopes=USER_SCOPES,
-)
-
 oauth_settings = OAuthSettings(
     client_id=os.environ.get("SLACK_CLIENT_ID"),
     client_secret=os.environ.get("SLACK_CLIENT_SECRET"),

pydantic-ai/tests/test_view_builders.py

@@ -26,7 +26,7 @@ def test_build_feedback_blocks():
 
 
 def test_build_app_home_view_default():
-    """Default args (app.py mode) — no OAuth section."""
+    """Default args (app.py mode) — no MCP status section."""
     view = build_app_home_view()
 
     assert view["type"] == "home"
@@ -35,36 +35,31 @@ def test_build_app_home_view_default():
     actions_block = next(b for b in view["blocks"] if b["type"] == "actions")
     assert len(actions_block["elements"]) == len(CATEGORIES)
 
-    # No connect or disconnect buttons
-    section_blocks = [b for b in view["blocks"] if b["type"] == "section"]
-    accessory_actions = [
-        b["accessory"]["action_id"] for b in section_blocks if "accessory" in b
+    # No MCP status section
+    section_texts = [
+        b["text"]["text"] for b in view["blocks"] if b["type"] == "section"
     ]
-    assert "connect_account" not in accessory_actions
-    assert "disconnect_account" not in accessory_actions
+    assert not any("Slack MCP Server" in t for t in section_texts)
 
 
 def test_build_app_home_view_connect():
-    """authorize_url provided — shows Connect URL button."""
-    view = build_app_home_view(authorize_url="https://example.com/oauth")
+    """install_url provided — shows disconnected status with install link."""
+    view = build_app_home_view(install_url="https://example.com/slack/install")
 
-    section_blocks = [b for b in view["blocks"] if b["type"] == "section"]
-    connect_section = next(
-        b
-        for b in section_blocks
-        if b.get("accessory", {}).get("action_id") == "connect_account"
-    )
-    assert connect_section["accessory"]["url"] == "https://example.com/oauth"
+    section_texts = [
+        b["text"]["text"] for b in view["blocks"] if b["type"] == "section"
+    ]
+    mcp_section = next(t for t in section_texts if "Slack MCP Server" in t)
+    assert "disconnected" in mcp_section
+    assert "https://example.com/slack/install" in mcp_section
 
 
-def test_build_app_home_view_disconnect():
-    """is_connected=True — shows Disconnect button."""
+def test_build_app_home_view_connected():
+    """is_connected=True — shows connected status."""
     view = build_app_home_view(is_connected=True)
 
-    section_blocks = [b for b in view["blocks"] if b["type"] == "section"]
-    disconnect_section = next(
-        b
-        for b in section_blocks
-        if b.get("accessory", {}).get("action_id") == "disconnect_account"
-    )
-    assert disconnect_section["accessory"]["style"] == "danger"
+    section_texts = [
+        b["text"]["text"] for b in view["blocks"] if b["type"] == "section"
+    ]
+    mcp_section = next(t for t in section_texts if "Slack MCP Server" in t)
+    assert "connected" in mcp_section