airbyte_ops_mcp.mcp.devin_secret_request

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for on-demand Devin secret request and discovery.
  3
  4Tools:
  5
  6- **list_devin_secrets**: Lists all available secret names in the
  7  1Password vault, so the agent can discover valid secret aliases.
  8- **request_devin_secret**: Two-phase approval workflow to request
  9  and deliver a secret.
 10
 11## MCP reference
 12
 13.. include:: ../../../docs/mcp-generated/devin_secret_request.md
 14    :start-line: 2
 15"""
 16
 17# NOTE: We intentionally do NOT use `from __future__ import annotations` here.
 18# FastMCP has issues resolving forward references when PEP 563 deferred annotations
 19# are used. See: https://github.com/jlowin/fastmcp/issues/905
 20
 21__all__: list[str] = []
 22
 23import json
 24import logging
 25import re
 26from typing import Annotated
 27
 28import requests
 29from fastmcp import FastMCP
 30from fastmcp_extensions import mcp_tool, register_mcp_tools
 31from pydantic import BaseModel, Field
 32
 33from airbyte_ops_mcp.github_actions import (
 34    download_job_logs,
 35    get_workflow_jobs,
 36    resolve_default_workflow_branch,
 37    trigger_workflow_dispatch,
 38    wait_for_workflow_completion,
 39)
 40from airbyte_ops_mcp.github_api import resolve_ci_trigger_github_token
 41from airbyte_ops_mcp.human_in_the_loop import HITL_SLACK_CHANNEL_URL
 42
 43logger = logging.getLogger(__name__)
 44
 45# ---------------------------------------------------------------------------
 46# Constants
 47# ---------------------------------------------------------------------------
 48
 49WORKFLOW_REPO_OWNER = "airbytehq"
 50WORKFLOW_REPO_NAME = "airbyte-ops-mcp"
 51WORKFLOW_FILE = "devin-secret-request.yml"
 52WORKFLOW_DEFAULT_BRANCH = "main"
 53
 54
 55_SESSION_ID_PATTERN = re.compile(r"[0-9a-fA-F]{32}")
 56
 57
 58class SecretListResponse(BaseModel):
 59    """Response from the list_devin_secrets tool."""
 60
 61    success: bool = Field(description="Whether the operation succeeded")
 62    message: str = Field(description="Human-readable status message")
 63    available_secrets: list[str] = Field(
 64        default_factory=list,
 65        description="Sorted list of available secret names in the vault",
 66    )
 67
 68
 69class SecretRequestResponse(BaseModel):
 70    """Response from the request_devin_secret tool."""
 71
 72    success: bool = Field(description="Whether the operation succeeded")
 73    phase: str = Field(
 74        description=(
 75            "Current phase: 'approval_requested' (Phase 1) or "
 76            "'delivery_dispatched' (Phase 2)"
 77        ),
 78    )
 79    message: str = Field(description="Human-readable status message")
 80    slack_channel_url: str = Field(
 81        default=HITL_SLACK_CHANNEL_URL,
 82        description="Direct URL to the #human-in-the-loop Slack channel",
 83    )
 84    secret_alias: str = Field(description="The requested secret alias")
 85    session_id: str = Field(description="The Devin session ID")
 86    workflow_url: str | None = Field(
 87        default=None,
 88        description="URL to the GitHub Actions workflow",
 89    )
 90    run_id: int | None = Field(
 91        default=None,
 92        description="GitHub Actions workflow run ID",
 93    )
 94    run_url: str | None = Field(
 95        default=None,
 96        description="Direct URL to the GitHub Actions workflow run",
 97    )
 98    request_id: str | None = Field(
 99        default=None,
100        description=(
101            "Unique request identifier (UUID). Returned in Phase 1; "
102            "pass it back in Phase 2 for replay-protection validation."
103        ),
104    )
105
106
107@mcp_tool(
108    read_only=False,
109    idempotent=False,
110    open_world=True,
111)
112def list_devin_secrets() -> SecretListResponse:
113    """List all available secret names in the 1Password vault.
114
115    Returns the sorted list of item titles from the
116    'devin-on-demand-secrets' vault. Use this to discover valid
117    secret aliases before calling request_devin_secret.
118
119    This dispatches a GitHub Actions workflow (which has the
120    1Password credentials), waits for it to complete, then reads
121    the list from the job logs.
122    """
123    return _list_secrets_via_workflow()
124
125
126@mcp_tool(
127    read_only=False,
128    idempotent=False,
129    open_world=True,
130)
131def request_devin_secret(
132    secret_alias: Annotated[
133        str,
134        "The name of the secret to request. This must exactly match an item "
135        "title in the 'devin-on-demand-secrets' 1Password vault.",
136    ],
137    session_url: Annotated[
138        str,
139        "Your Devin session URL (e.g. 'https://app.devin.ai/sessions/abc123...'). "
140        "Use the session URL from your system prompt.",
141    ],
142    approval_evidence_url: Annotated[
143        str | None,
144        "Slack approval record URL "
145        "(https://<workspace>.slack.com/archives/...). "
146        "Leave empty for Phase 1 (requesting approval). Provide the "
147        "Slack URL for Phase 2 (delivering the secret after approval).",
148    ] = None,
149    target_approver: Annotated[
150        str | None,
151        "Person to notify for approval (GitHub handle, email, or Slack user ID). "
152        "Required for Phase 1 (approval request).",
153    ] = None,
154    request_id: Annotated[
155        str | None,
156        "Request ID returned by Phase 1. Pass it back in Phase 2 "
157        "so the approval record can be validated against the original request. "
158        "Leave empty for Phase 1.",
159    ] = None,
160) -> SecretRequestResponse:
161    """Request a secret on demand via an approval workflow.
162
163    This tool operates in two phases:
164
165    **Phase 1** (no approval_evidence_url): Dispatches a GitHub Actions
166    workflow that validates the secret name against the 1Password vault
167    and, if valid, sends a Slack approval request. If the secret name is
168    not found, returns immediately with the list of available secret
169    names so you can correct any typos.
170
171    **Phase 2** (with approval_evidence_url): After a human approves the
172    request, call this tool again with the approval evidence URL. This
173    triggers a GitHub Actions workflow that reads the secret from
174    1Password and sends you a time-limited share link.
175    Open the link in your browser to view and copy the secret.
176
177    Typical workflow:
178    0. (Optional) Call list_devin_secrets first to see available names.
179    1. Call this tool without approval_evidence_url to request approval.
180    2. Note the `request_id` in the response.
181    3. Wait for a human to approve the request in Slack.
182    4. Obtain the approval evidence URL (Slack approval record URL).
183    5. Call this tool again with the approval_evidence_url **and** the
184       request_id from step 2.
185    6. You will receive a 1Password share link -- open it in your
186       browser to view and copy the secret values.
187    """
188    # Extract session ID from URL
189    match = _SESSION_ID_PATTERN.search(session_url)
190    if not match:
191        return SecretRequestResponse(
192            success=False,
193            phase="error",
194            message=(
195                f"No valid session ID found in URL: {session_url}. "
196                "Expected a 32-character hex string."
197            ),
198            secret_alias=secret_alias,
199            session_id="",
200        )
201    session_id = match.group(0)
202
203    if not approval_evidence_url:
204        # Phase 1: Dispatch the request workflow (validates secret name
205        # inline using op CLI, then sends Slack approval if valid).
206        if not target_approver:
207            return SecretRequestResponse(
208                success=False,
209                phase="error",
210                message=(
211                    "target_approver is required when requesting approval "
212                    "(no approval_evidence_url provided)."
213                ),
214                secret_alias=secret_alias,
215                session_id=session_id,
216            )
217
218        return _request_secret_via_workflow(
219            secret_alias=secret_alias,
220            session_id=session_id,
221            session_url=session_url,
222            target_approver=target_approver,
223        )
224
225    # Phase 2: Deliver secret via GitHub Actions workflow
226    token = resolve_ci_trigger_github_token()
227
228    workflow_inputs: dict[str, str] = {
229        "action": "deliver",
230        "secret_alias": secret_alias,
231        "session_id": session_id,
232        "approval_evidence_url": approval_evidence_url,
233    }
234    if request_id:
235        workflow_inputs["expected_request_id"] = request_id
236
237    result = trigger_workflow_dispatch(
238        owner=WORKFLOW_REPO_OWNER,
239        repo=WORKFLOW_REPO_NAME,
240        workflow_file=WORKFLOW_FILE,
241        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
242        inputs=workflow_inputs,
243        token=token,
244    )
245
246    view_url = result.run_url or result.workflow_url
247    return SecretRequestResponse(
248        success=True,
249        phase="delivery_dispatched",
250        message=(
251            f"Secret delivery workflow dispatched for '{secret_alias}'. "
252            f"The workflow will read the secret from 1Password and send "
253            f"you a time-limited share link. Once you receive the link, "
254            f"open it in your browser to view and copy the secret. "
255            f"View progress: {view_url}"
256        ),
257        secret_alias=secret_alias,
258        session_id=session_id,
259        workflow_url=result.workflow_url,
260        run_id=result.run_id,
261        run_url=result.run_url,
262        request_id=request_id,
263    )
264
265
266def _request_secret_via_workflow(
267    secret_alias: str,
268    session_id: str,
269    session_url: str,
270    target_approver: str,
271) -> SecretRequestResponse:
272    """Dispatch the request workflow, wait, and parse the result from job logs.
273
274    The workflow validates the secret alias against the vault inline,
275    then sends the Slack approval if valid.  On a bad alias the workflow
276    fails and the job logs contain a JSON object with `available_secrets`.
277    """
278    token = resolve_ci_trigger_github_token()
279
280    dispatch_result = trigger_workflow_dispatch(
281        owner=WORKFLOW_REPO_OWNER,
282        repo=WORKFLOW_REPO_NAME,
283        workflow_file=WORKFLOW_FILE,
284        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
285        inputs={
286            "action": "request",
287            "secret_alias": secret_alias,
288            "session_id": session_id,
289            "target_approver": target_approver,
290        },
291        token=token,
292    )
293    if not dispatch_result.run_id:
294        return SecretRequestResponse(
295            success=False,
296            phase="error",
297            message=(
298                "Workflow dispatched but no run ID returned. "
299                f"Check: {dispatch_result.workflow_url}"
300            ),
301            secret_alias=secret_alias,
302            session_id=session_id,
303            workflow_url=dispatch_result.workflow_url,
304        )
305
306    run_status = wait_for_workflow_completion(
307        owner=WORKFLOW_REPO_OWNER,
308        repo=WORKFLOW_REPO_NAME,
309        run_id=dispatch_result.run_id,
310        token=token,
311    )
312
313    # Download logs from the validation job (multi-job workflow)
314    raw_logs = _download_run_logs(
315        dispatch_result.run_id, token, job_name="Validate Secret Name"
316    )
317
318    if run_status.succeeded:
319        # Parse the approval-requested JSON from the logs
320        result_data = _find_json_in_logs(raw_logs, "phase") if raw_logs else None
321        request_id = result_data.get("request_id") if result_data else None
322        view_url = run_status.run_url or dispatch_result.workflow_url
323        return SecretRequestResponse(
324            success=True,
325            phase="approval_requested",
326            message=(
327                f"Approval request for secret '{secret_alias}' sent to "
328                f"#human-in-the-loop ({HITL_SLACK_CHANNEL_URL}). "
329                f"Waiting for human approval. "
330                f"Once approved, call this tool again with the "
331                f"approval_evidence_url to deliver the secret. "
332                f"View progress: {view_url}"
333            ),
334            secret_alias=secret_alias,
335            session_id=session_id,
336            workflow_url=dispatch_result.workflow_url,
337            run_id=dispatch_result.run_id,
338            run_url=run_status.run_url,
339            request_id=request_id,
340        )
341
342    # Workflow failed — check if it was a validation failure
343    error_data = _find_json_in_logs(raw_logs, "available_secrets") if raw_logs else None
344    if error_data:
345        available = error_data.get("available_secrets", [])
346        formatted = ", ".join(f"`{s}`" for s in available)
347        return SecretRequestResponse(
348            success=False,
349            phase="validation_failed",
350            message=(
351                f"Secret '{secret_alias}' not found in the vault. "
352                f"Available secrets: {formatted}"
353            ),
354            secret_alias=secret_alias,
355            session_id=session_id,
356            workflow_url=dispatch_result.workflow_url,
357            run_id=dispatch_result.run_id,
358            run_url=run_status.run_url,
359        )
360
361    # Generic workflow failure
362    return SecretRequestResponse(
363        success=False,
364        phase="error",
365        message=(
366            f"Request workflow failed (conclusion={run_status.conclusion}). "
367            f"See: {run_status.run_url}"
368        ),
369        secret_alias=secret_alias,
370        session_id=session_id,
371        workflow_url=dispatch_result.workflow_url,
372        run_id=dispatch_result.run_id,
373        run_url=run_status.run_url,
374    )
375
376
377def _download_run_logs(
378    run_id: int,
379    token: str,
380    *,
381    job_name: str | None = None,
382) -> str | None:
383    """Best-effort download of a job's logs for a workflow run.
384
385    Args:
386        run_id: GitHub Actions workflow run ID.
387        token: GitHub API token used for log download. Note: job listing
388            uses `get_workflow_jobs` which resolves its own token via
389            `resolve_ci_trigger_github_token()`.
390        job_name: If provided, find the job whose name contains this
391            substring (case-insensitive). Skipped jobs are always
392            excluded. Falls back to the first non-skipped job.
393    """
394    try:
395        jobs = get_workflow_jobs(
396            owner=WORKFLOW_REPO_OWNER,
397            repo=WORKFLOW_REPO_NAME,
398            run_id=run_id,
399        )
400        # Filter out skipped jobs (common in multi-job conditional workflows)
401        active_jobs = [j for j in jobs if j.conclusion != "skipped"]
402        if not active_jobs:
403            return None
404
405        target = active_jobs[0]  # default: first non-skipped job
406        if job_name:
407            needle = job_name.lower()
408            for j in active_jobs:
409                if needle in j.name.lower():
410                    target = j
411                    break
412
413        return download_job_logs(
414            owner=WORKFLOW_REPO_OWNER,
415            repo=WORKFLOW_REPO_NAME,
416            job_id=target.job_id,
417            token=token,
418        )
419    except (requests.HTTPError, ValueError) as exc:
420        logger.warning("Failed to download job logs for run %s: %s", run_id, exc)
421        return None
422
423
424def _list_secrets_via_workflow() -> SecretListResponse:
425    """Dispatch the list workflow, wait for completion, and parse titles from job logs."""
426    token = resolve_ci_trigger_github_token()
427
428    # 1. Dispatch the workflow with action="list"
429    dispatch_result = trigger_workflow_dispatch(
430        owner=WORKFLOW_REPO_OWNER,
431        repo=WORKFLOW_REPO_NAME,
432        workflow_file=WORKFLOW_FILE,
433        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
434        inputs={"action": "list", "session_id": "0" * 32},
435        token=token,
436    )
437    if not dispatch_result.run_id:
438        return SecretListResponse(
439            success=False,
440            message=(
441                "Workflow dispatched but no run ID returned. "
442                f"Check: {dispatch_result.workflow_url}"
443            ),
444        )
445
446    # 2. Wait for the workflow to complete
447    run_status = wait_for_workflow_completion(
448        owner=WORKFLOW_REPO_OWNER,
449        repo=WORKFLOW_REPO_NAME,
450        run_id=dispatch_result.run_id,
451        token=token,
452    )
453    if not run_status.succeeded:
454        return SecretListResponse(
455            success=False,
456            message=(
457                f"Workflow run failed (conclusion={run_status.conclusion}). "
458                f"See: {run_status.run_url}"
459            ),
460        )
461
462    # 3. Find the job and download its logs
463    jobs = get_workflow_jobs(
464        owner=WORKFLOW_REPO_OWNER,
465        repo=WORKFLOW_REPO_NAME,
466        run_id=dispatch_result.run_id,
467    )
468    if not jobs:
469        return SecretListResponse(
470            success=False,
471            message="Workflow completed but no jobs found.",
472        )
473
474    # Find the list job (multi-job workflow; skip skipped jobs)
475    active_jobs = [j for j in jobs if j.conclusion != "skipped"]
476    if not active_jobs:
477        return SecretListResponse(
478            success=False,
479            message="Workflow completed but all jobs were skipped.",
480        )
481
482    target_job = active_jobs[0]
483    for j in active_jobs:
484        if "list" in j.name.lower():
485            target_job = j
486            break
487
488    raw_logs = download_job_logs(
489        owner=WORKFLOW_REPO_OWNER,
490        repo=WORKFLOW_REPO_NAME,
491        job_id=target_job.job_id,
492        token=token,
493    )
494
495    # 4. Parse JSON output from the logs
496    data = _find_json_in_logs(raw_logs, "available_secrets")
497    if data is None:
498        return SecretListResponse(
499            success=False,
500            message=(
501                "Could not parse secret list from workflow logs. "
502                f"See: {run_status.run_url}"
503            ),
504        )
505
506    secrets = data.get("available_secrets", [])
507    titles = [str(t) for t in secrets] if isinstance(secrets, list) else []
508    return SecretListResponse(
509        success=True,
510        message=f"Found {len(titles)} available secrets in the vault.",
511        available_secrets=sorted(titles),
512    )
513
514
515def _find_json_in_logs(raw_logs: str, required_key: str) -> dict | None:
516    """Find the first JSON object in job logs that contains *required_key*.
517
518    GitHub Actions job logs prefix each line with a timestamp.  We scan
519    every line looking for a JSON object that contains the given key.
520    Returns the parsed dict, or `None` if not found.
521    """
522    for line in raw_logs.splitlines():
523        stripped = line.strip()
524        if not stripped.startswith("{"):
525            idx = stripped.find("{")
526            if idx < 0:
527                continue
528            stripped = stripped[idx:]
529        try:
530            data = json.loads(stripped)
531        except json.JSONDecodeError:
532            continue
533        if isinstance(data, dict) and required_key in data:
534            return data
535    return None
536
537
538def register_devin_secret_request_tools(app: FastMCP) -> None:
539    """Register Devin secret request tools with the FastMCP app."""
540    register_mcp_tools(app, mcp_module=__name__)