airbyte_ops_mcp.mcp.human_in_the_loop

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for human-in-the-loop workflows: escalation to a human, team-roster lookup, and Slack newsletter posting.
  3
  4## MCP reference
  5
  6.. include:: ../../../docs/mcp-generated/human_in_the_loop.md
  7    :start-line: 2
  8"""
  9
 10from __future__ import annotations
 11
 12__all__: list[str] = []
 13
 14import json
 15import logging
 16from enum import Enum, StrEnum
 17from typing import Annotated, Literal
 18
 19from fastmcp import FastMCP
 20from fastmcp_extensions import mcp_tool, register_mcp_tools
 21from pydantic import BaseModel, Field
 22
 23from airbyte_ops_mcp.github_actions import (
 24    WorkflowDispatchResult,
 25    WorkflowRunStatus,
 26    resolve_default_workflow_branch,
 27    trigger_workflow_dispatch,
 28    wait_for_workflow_completion,
 29)
 30from airbyte_ops_mcp.github_api import resolve_ci_trigger_github_token
 31from airbyte_ops_mcp.human_in_the_loop import (
 32    HITL_SLACK_CHANNEL_URL,
 33    dispatch_escalation,
 34)
 35from airbyte_ops_mcp.internal_team_roster import fetch_roster, search_roster
 36from airbyte_ops_mcp.slack_api import lookup_slack_usergroup as find_slack_usergroups
 37from airbyte_ops_mcp.slack_ops.blocks import (
 38    build_blocks,
 39    validate_message,
 40)
 41
 42
 43class PersonRecord(BaseModel):
 44    """A person in the internal team roster."""
 45
 46    slack_id: str | None = Field(default=None, description="Slack user ID")
 47    slack_display_name: str | None = Field(
 48        default=None, description="Slack display name"
 49    )
 50    slack_email: str | None = Field(
 51        default=None, description="Email address from Slack profile"
 52    )
 53    github_id: int | None = Field(default=None, description="GitHub numeric user ID")
 54    github_handle: str | None = Field(default=None, description="GitHub username")
 55    github_public_email: str | None = Field(
 56        default=None, description="Public email from GitHub profile"
 57    )
 58
 59
 60class PeopleLookupResponse(BaseModel):
 61    """Response from a people lookup query."""
 62
 63    query: str = Field(description="The search query that was used")
 64    total_matches: int = Field(description="Number of matching records")
 65    matches: list[PersonRecord] = Field(description="Matching person records")
 66
 67
 68class RosterListResponse(BaseModel):
 69    """Response listing the full internal team roster."""
 70
 71    total_members: int = Field(description="Total number of members in the roster")
 72    members: list[PersonRecord] = Field(description="All person records in the roster")
 73
 74
 75class SlackUsergroupRecord(BaseModel):
 76    """A Slack usergroup and its mention metadata."""
 77
 78    id: str = Field(description="Slack usergroup ID used in <!subteam^ID|@handle>")
 79    handle: str = Field(description="Slack usergroup handle without the leading @")
 80    name: str = Field(description="Slack usergroup display name")
 81    description: str = Field(description="Slack usergroup description")
 82    user_count: int = Field(description="Number of members in the usergroup")
 83
 84
 85class SlackUsergroupLookupResponse(BaseModel):
 86    """Response from a Slack usergroup lookup."""
 87
 88    id_or_handle: str = Field(description="The usergroup ID or handle/name lookup")
 89    total_matches: int = Field(description="Number of matching usergroups")
 90    matches: list[SlackUsergroupRecord] = Field(description="Matching Slack usergroups")
 91
 92
 93@mcp_tool(
 94    read_only=True,
 95    idempotent=True,
 96)
 97def lookup_person(
 98    query: Annotated[
 99        str,
100        Field(
101            description=(
102                "Search query to match against any field: email address, "
103                "Slack display name, Slack user ID, GitHub handle, or GitHub user ID. "
104                "Case-insensitive partial matching for strings, exact match for numeric IDs."
105            )
106        ),
107    ],
108) -> PeopleLookupResponse:
109    """Look up a person in the Airbyte internal team roster.
110
111    Searches the daily-generated roster artifact by any field value.
112    The roster is built from Slack and GitHub org membership data,
113    cross-referenced by email address.
114
115    Use this to find someone's Slack ID for messaging, GitHub handle
116    for code review, or to cross-reference identities across platforms.
117    """
118    roster = fetch_roster()
119    matches = search_roster(roster, query)
120    return PeopleLookupResponse(
121        query=query,
122        total_matches=len(matches),
123        matches=[PersonRecord.model_validate(person) for person in matches],
124    )
125
126
127@mcp_tool(
128    read_only=True,
129    idempotent=True,
130)
131def list_team_roster() -> RosterListResponse:
132    """List the full Airbyte internal team roster.
133
134    Returns all members from the daily-generated roster artifact.
135    The roster is sorted by Slack display name for easy scanning.
136    Use lookup_person for targeted searches instead of scanning the full list.
137    """
138    roster = fetch_roster()
139    return RosterListResponse(
140        total_members=len(roster),
141        members=[PersonRecord.model_validate(person) for person in roster],
142    )
143
144
145@mcp_tool(
146    read_only=True,
147    idempotent=True,
148)
149def lookup_slack_usergroup(
150    id_or_handle: Annotated[
151        str,
152        Field(
153            description=(
154                "Required Slack usergroup handle/name or S-prefixed usergroup ID. "
155                "Handle/name matching is "
156                "case-insensitive and partial; a leading @ is ignored. "
157                "S-prefixed IDs are matched exactly."
158            )
159        ),
160    ],
161) -> SlackUsergroupLookupResponse:
162    """Resolve Slack usergroups to IDs for usergroup mentions.
163
164    Pass a handle or name to resolve oncall aliases such as `@oc-apis`, or
165    pass an S-prefixed usergroup ID for the corresponding handle and name.
166    Results include the ID and metadata needed for the
167    `<!subteam^ID|@alias>` mention syntax used by Slack messages.
168    """
169    usergroups = find_slack_usergroups(id_or_handle)
170    return SlackUsergroupLookupResponse(
171        id_or_handle=id_or_handle,
172        total_matches=len(usergroups),
173        matches=[
174            SlackUsergroupRecord(
175                id=usergroup.id,
176                handle=usergroup.handle,
177                name=usergroup.name,
178                description=usergroup.description,
179                user_count=usergroup.user_count,
180            )
181            for usergroup in usergroups
182        ],
183    )
184
185
186class RequestType(StrEnum):
187    """Type of escalation request, controlling the Slack header emoji and label."""
188
189    ACTION = "action"
190    REVIEW = "review"
191    INPUT = "input"
192    GUIDANCE = "guidance"
193    APPROVAL = "approval"
194    BLOCKED = "blocked"
195
196
197_REQUEST_TYPE_HEADERS: dict[RequestType, tuple[str, str]] = {
198    RequestType.ACTION: ("🔧", "Action Requested"),
199    RequestType.REVIEW: ("👀", "Review Requested"),
200    RequestType.INPUT: ("❓", "Input Needed"),
201    RequestType.GUIDANCE: ("🧭", "Guidance Needed"),
202    RequestType.APPROVAL: ("✅", "Approval Requested"),
203    RequestType.BLOCKED: ("🚫", "Still Blocked"),
204}
205
206
207class EscalateToHumanResponse(BaseModel):
208    """Response from the human-in-the-loop escalation tool."""
209
210    success: bool = Field(description="Whether the workflow was triggered successfully")
211    message: str = Field(description="Human-readable status message")
212    slack_channel_url: str = Field(
213        default=HITL_SLACK_CHANNEL_URL,
214        description="Direct URL to the #human-in-the-loop Slack channel",
215    )
216    workflow_url: str | None = Field(
217        default=None,
218        description="URL to view the GitHub Actions workflow file",
219    )
220    run_id: int | None = Field(
221        default=None,
222        description="GitHub Actions workflow run ID",
223    )
224    run_url: str | None = Field(
225        default=None,
226        description="Direct URL to the GitHub Actions workflow run",
227    )
228
229
230@mcp_tool(
231    read_only=False,
232    idempotent=False,
233    open_world=True,
234)
235def escalate_to_human(
236    target_person: Annotated[
237        str,
238        "Primary person to notify. Accepts an email address (e.g. 'aj@airbyte.io'), "
239        "a GitHub handle prefixed with @ (e.g. '@aaronsteers'), "
240        "a Slack user ID (e.g. 'U05AKF1BCC9'), or a Slack usergroup ID "
241        "(e.g. 'S0BKR63VAN5' for @oc-internal-ai). Slack usergroup handles "
242        "such as '@oc-internal-ai' are not resolvable; use the ID form.",
243    ],
244    message: Annotated[
245        str,
246        "The message body to deliver to the human. Format using Slack mrkdwn: "
247        "*bold*, _italic_, `code`, ```code blocks```, > blockquotes, "
248        "- bullet lists, and <url|label> links. Should clearly explain "
249        "what you need help with or what decision is required.",
250    ],
251    agent_session_url: Annotated[
252        str,
253        "Your agent session URL so the human can view the full context. "
254        "Use the session URL from your system prompt.",
255    ],
256    cc: Annotated[
257        list[str] | None,
258        "Optional list of additional people or Slack usergroups to tag on the "
259        "message. Each entry uses the same identifier format as target_person. "
260        "For usergroups, use the S-prefixed ID (e.g. 'S0BKR63VAN5' for "
261        "@oc-internal-ai), not the handle.",
262    ] = None,
263    pr_url: Annotated[
264        str | None,
265        "Optional URL to a related pull request for the 'View PR' button.",
266    ] = None,
267    issue_url: Annotated[
268        str | None,
269        "Optional URL to a related GitHub issue for the 'View Issue' button.",
270    ] = None,
271    additional_actions: Annotated[
272        dict[str, str] | None,
273        "Optional dictionary of label -> URL pairs for extra action buttons. "
274        "Example: {'Start Workflow': 'https://github.com/...actions/...'}.",
275    ] = None,
276    approval_requested: Annotated[
277        bool,
278        "Set to True to add 'Approve' and 'Reject' buttons that post back to the Slack app. "
279        "Each button includes a confirmation dialog (if approval_request_summary is provided). "
280        "When either button is clicked, both buttons morph into non-interactive status text "
281        "(e.g. ':white_check_mark: Approved by @user' or ':x: Rejected by @user').",
282    ] = False,
283    approval_request_summary: Annotated[
284        str | None,
285        "Short description of what the user is approving, shown in a Slack "
286        "confirmation dialog before the Approve button fires. Rendered as a "
287        "blockquote. MUST be at most 280 characters; over-limit or "
288        "unbalanced-backtick inputs are rejected at call time. Keep it to the "
289        "minimum info the approver needs to identify what they are approving. "
290        "Example: 'Pinning source-hubspot prerelease 4.5.3-preview to workspace for testing'.",
291    ] = None,
292    approval_request_detail_url: Annotated[
293        str | None,
294        "Optional URL where the reviewer can read full details of what they are "
295        "being asked to approve. Rendered as a 'View Details' button in the Slack message.",
296    ] = None,
297    connector_name: Annotated[
298        str | None,
299        "Optional connector name to display prominently in the Slack message. "
300        "For example, when combined with request_type='action', the header may be "
301        "rendered as '🔧 Action Requested — source-salesloft'. If request_type is omitted, "
302        "the header remains the generic '🙋 Human-in-the-loop request' and the connector "
303        "name is still included for additional context. Always provide this when the "
304        "escalation is about a specific connector.",
305    ] = None,
306    request_type: Annotated[
307        RequestType | None,
308        "Type of escalation request. Controls the Slack message header emoji and label. "
309        "Accepted values: 'action' (🔧 Action Requested), "
310        "'review' (👀 Review Requested), 'input' (❓ Input Needed), "
311        "'guidance' (🧭 Guidance Needed), 'approval' (✅ Approval Requested), "
312        "'blocked' (🚫 Still Blocked). "
313        "When omitted, defaults to the generic 'Human-in-the-loop request' header.",
314    ] = None,
315) -> EscalateToHumanResponse:
316    """Escalate to a human or Slack usergroup via Slack.
317
318    Posts a formatted message to the #human-in-the-loop Slack channel,
319    tagging the specified person(s) or usergroup(s). The message includes
320    clickable buttons for the Devin session, PR, and issue links when provided,
321    plus any additional freeform action buttons.
322
323    The Slack message is sent by a GitHub Actions workflow so that Slack
324    credentials are never exposed to the calling agent. The workflow
325    resolves person identifiers (email, GitHub handle, or Slack ID) to Slack
326    user IDs using the internal team roster. S-prefixed Slack usergroup IDs
327    bypass roster resolution and render as usergroup mentions.
328
329    Use this tool when you need human input, approval, or help that you
330    cannot resolve on your own.
331    """
332    # Resolve request_type to header_emoji and header_label
333    header_emoji: str | None = None
334    header_label: str | None = None
335    if request_type is not None:
336        header_emoji, header_label = _REQUEST_TYPE_HEADERS[request_type]
337
338    result = dispatch_escalation(
339        target_person=target_person,
340        message=message,
341        agent_session_url=agent_session_url,
342        cc=cc,
343        pr_url=pr_url,
344        issue_url=issue_url,
345        additional_actions=additional_actions,
346        approval_requested=approval_requested,
347        approval_request_summary=approval_request_summary,
348        approval_request_detail_url=approval_request_detail_url,
349        connector_name=connector_name,
350        header_emoji=header_emoji,
351        header_label=header_label,
352    )
353
354    view_url = result.run_url or result.workflow_url
355    return EscalateToHumanResponse(
356        success=True,
357        message=(
358            f"Escalation sent to '{target_person}' via #human-in-the-loop "
359            f"({HITL_SLACK_CHANNEL_URL}). "
360            f"View progress at: {view_url}"
361        ),
362        workflow_url=result.workflow_url,
363        run_id=result.run_id,
364        run_url=result.run_url,
365    )
366
367
368logger = logging.getLogger(__name__)
369
370_NEWSLETTER_CHANNELS: dict[str, tuple[str, str]] = {
371    "DR": ("C0AH48172M6", "#daily-newsletters"),
372    "Internal AI": ("C0AH48172M6", "#daily-newsletters"),
373    "AJ": ("C0BLUPJ0X0R", "#aj-release-notes"),
374}
375
376_REPO_OWNER = "airbytehq"
377
378_REPO_NAME = "airbyte-ops-mcp"
379
380_WORKFLOW_FILE = "slack-post-message.yml"
381
382_DEFAULT_BRANCH = "main"
383
384_DRY_RUN_CHANNEL = "C0AEN317Z7T"
385
386
387class DryRunMode(str, Enum):
388    """Controls how dry-run behaves."""
389
390    off = "off"
391    """Normal mode — post to the real channel."""
392
393    local = "local"
394    """Local-only preview. Builds Block Kit JSON and returns both the
395    blocks JSON (for agent-side rendering) and a Block Kit Builder URL.
396    No workflow is triggered."""
397
398    slack_test_channel = "slack_test_channel"
399    """Triggers the real workflow but posts to a test channel instead of
400    the production newsletter channel."""
401
402
403class PostToSlackChannelResponse(BaseModel):
404    """Response from the post_slack_newsletter tool."""
405
406    success: bool = Field(description="Whether the operation completed successfully")
407    message: str = Field(description="Human-readable status message")
408    blocks_json: str | None = Field(
409        default=None,
410        description=(
411            "Serialised JSON array of Block Kit blocks. "
412            "Populated in 'local' dry-run mode so the agent can "
413            "render or inspect the blocks directly."
414        ),
415    )
416    workflow_url: str | None = Field(
417        default=None,
418        description="URL to view the GitHub Actions workflow file",
419    )
420    run_id: int | None = Field(
421        default=None,
422        description="GitHub Actions workflow run ID",
423    )
424    channel_name: str | None = Field(
425        default=None,
426        description="Human-readable Slack channel name (e.g. '#daily-newsletters')",
427    )
428    run_url: str | None = Field(
429        default=None,
430        description="Direct URL to the GitHub Actions workflow run",
431    )
432
433
434def _format_failure_details(run_status: WorkflowRunStatus) -> str:
435    """Build a human-readable summary of failed jobs from a workflow run."""
436    if not run_status.jobs:
437        return ""
438    failed_jobs = [j for j in run_status.jobs if j.conclusion == "failure"]
439    if not failed_jobs:
440        return ""
441    job_summaries = [f"  - {j.name} (job_id={j.job_id})" for j in failed_jobs]
442    return " Failed jobs:\n" + "\n".join(job_summaries)
443
444
445@mcp_tool(
446    read_only=False,
447    idempotent=False,
448    open_world=True,
449)
450def post_slack_newsletter(
451    message_text: Annotated[
452        str,
453        "The formatted message to post, using Slack mrkdwn syntax: "
454        "*bold*, _italic_, `code`, ```code blocks```, > blockquotes, "
455        "- bullet lists, and <url|label> links. "
456        "Use ## for section headers (translated to Block Kit header blocks) "
457        "and ### for sub-headers (translated to bold text). "
458        "Double newlines (\\n\\n) split the text into separate visual "
459        "sections in the Slack message. "
460        "Do NOT include markdown tables — they will be rejected.",
461    ],
462    newsletter_name: Annotated[
463        Literal["DR", "Internal AI", "AJ"],
464        "Name of the newsletter to post to. "
465        "Determines which Slack channel receives the message. "
466        "DR and Internal AI both post to #daily-newsletters; AJ posts to "
467        "#aj-release-notes. "
468        "Ignored when dry_run is 'slack_test_channel'.",
469    ],
470    dry_run: Annotated[
471        str,
472        "Controls dry-run behaviour. "
473        "'off' (default): post to the real channel. "
474        "'local': build blocks locally and return blocks JSON plus a "
475        "Block Kit Builder URL — no workflow is triggered. "
476        "'slack_test_channel': trigger the workflow but post to a test "
477        "channel instead of the production newsletter channel.",
478    ] = "off",
479) -> PostToSlackChannelResponse:
480    """Post a formatted newsletter digest to a Slack channel.
481
482    Converts Slack mrkdwn text into Block Kit blocks locally (with
483    validation), then dispatches them to a GitHub Actions workflow for
484    posting.  The workflow receives finished Block Kit JSON — it does
485    not perform any markdown conversion.
486
487    Dry-run modes:
488    - `local`: returns blocks JSON for local rendering — no workflow triggered
489    - `slack_test_channel`: posts to a test channel for formatting verification
490    """
491    try:
492        mode = DryRunMode(dry_run)
493    except ValueError:
494        allowed_values = ", ".join(m.value for m in DryRunMode)
495        return PostToSlackChannelResponse(
496            success=False,
497            message=(
498                f"Invalid dry_run value {dry_run!r}. "
499                f"Allowed values are: {allowed_values}."
500            ),
501        )
502
503    # --- Validate input ---
504    try:
505        validate_message(message_text)
506    except ValueError as exc:
507        return PostToSlackChannelResponse(
508            success=False,
509            message=str(exc),
510        )
511
512    # --- Build Block Kit blocks locally ---
513    blocks_dict = build_blocks(message_text)
514    blocks_json = json.dumps(blocks_dict)
515
516    # --- Local mode: return blocks JSON, no dispatch ---
517    if mode is DryRunMode.local:
518        return PostToSlackChannelResponse(
519            success=True,
520            message=(
521                "Local preview generated. Use the blocks_json field to "
522                "render the message locally."
523            ),
524            blocks_json=blocks_json,
525        )
526
527    # --- Resolve target channel ---
528    if mode is DryRunMode.slack_test_channel:
529        resolved_channel = _DRY_RUN_CHANNEL
530        resolved_channel_name = "#slackbot-testing-channel--ignore-plz"
531    else:
532        resolved_channel, resolved_channel_name = _NEWSLETTER_CHANNELS[newsletter_name]
533
534    resolved_fallback = message_text[:120].replace("\n", " ")
535
536    # --- Dispatch to GitHub Actions ---
537    token = resolve_ci_trigger_github_token()
538    result: WorkflowDispatchResult = trigger_workflow_dispatch(
539        owner=_REPO_OWNER,
540        repo=_REPO_NAME,
541        workflow_file=_WORKFLOW_FILE,
542        ref=resolve_default_workflow_branch(_DEFAULT_BRANCH),
543        inputs={
544            "channel_id": resolved_channel,
545            "blocks_json": blocks_json,
546            "fallback_text": resolved_fallback,
547        },
548        token=token,
549    )
550
551    view_url = result.run_url or result.workflow_url
552    mode_label = (
553        " (dry run — test channel)" if mode is DryRunMode.slack_test_channel else ""
554    )
555
556    # --- Guard: if we couldn't discover the run, return early ---
557    if result.run_id is None:
558        return PostToSlackChannelResponse(
559            success=False,
560            message=(
561                f"Workflow dispatched to {resolved_channel_name}{mode_label} "
562                f"but could not discover the run ID to verify completion. "
563                f"Check manually: {view_url}"
564            ),
565            channel_name=resolved_channel_name,
566            workflow_url=result.workflow_url,
567        )
568
569    # --- Wait for workflow completion and report failures ---
570    run_status: WorkflowRunStatus = wait_for_workflow_completion(
571        owner=_REPO_OWNER,
572        repo=_REPO_NAME,
573        run_id=result.run_id,
574        token=token,
575        poll_interval_seconds=5.0,
576        max_wait_seconds=120.0,
577    )
578
579    if run_status.failed:
580        failure_details = _format_failure_details(run_status)
581        return PostToSlackChannelResponse(
582            success=False,
583            message=(
584                f"Workflow run FAILED (conclusion={run_status.conclusion}) "
585                f"for {resolved_channel_name}{mode_label}. "
586                f"Run: {run_status.run_url or view_url}"
587                f"{failure_details}"
588            ),
589            channel_name=resolved_channel_name,
590            workflow_url=result.workflow_url,
591            run_id=result.run_id,
592            run_url=run_status.run_url or result.run_url,
593        )
594
595    if not run_status.succeeded:
596        return PostToSlackChannelResponse(
597            success=False,
598            message=(
599                f"Workflow run did not complete within 120s "
600                f"(status={run_status.status}, "
601                f"conclusion={run_status.conclusion}){mode_label}. "
602                f"Run: {run_status.run_url or view_url}"
603            ),
604            channel_name=resolved_channel_name,
605            workflow_url=result.workflow_url,
606            run_id=result.run_id,
607            run_url=run_status.run_url or result.run_url,
608        )
609
610    return PostToSlackChannelResponse(
611        success=True,
612        message=(
613            f"Message posted to {resolved_channel_name} "
614            f"({resolved_channel}){mode_label}. "
615            f"Run: {view_url}"
616        ),
617        channel_name=resolved_channel_name,
618        workflow_url=result.workflow_url,
619        run_id=result.run_id,
620        run_url=result.run_url,
621    )
622
623
624def register_human_in_the_loop_tools(app: FastMCP) -> None:
625    """Register human-in-the-loop tools with the FastMCP app."""
626    register_mcp_tools(app, mcp_module=__name__)