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