【調査レポ12】Pythonで自作MCPサーバーを本格構築！FastMCPによるTools・Resources・Promptsローカル実践ガイド

# 仮想環境の作成と有効化
python3 -m venv .venv
source .venv/bin/activate

# MCP公式パッケージのインストール
pip install mcp

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
実務型システム管理・ログ監視MCPサーバー (server_ops_mcp.py)
Google Antigravity SDK および Model Context Protocol (MCP) 対応サーバー。
FastMCPフレームワークを活用し、Tools / Resources / Prompts の主要機能を提供します。
"""

import sys
import socket
import datetime
import subprocess
from mcp.server.fastmcp import FastMCP

# =====================================================================
# 0. MCPサーバーインスタンスの初期化
# =====================================================================
# 引数にサーバーの識別名（一意な名前）を指定して初期化します。
# FastMCPが内部でJSON-RPC 2.0メッセージの通信やプロトコル交渉、各種マネージャーを自動構築します。
mcp = FastMCP("ServerOpsMCP")

# =====================================================================
# 1. Tools（能動的アクション実行）
# =====================================================================
# @mcp.tool() デコレータを付与することで、AIエージェントが自律的に呼び出せる関数として公開されます。
# 関数の引数型ヒント（Type Hints）とDocstringが自動解析され、AIが解釈するツール仕様（JSON Schema）になります。

@mcp.tool()
def search_processes(keyword: str) -> str:
    """指定されたキーワードに一致する実行中プロセスを検索し、PIDとプロセス名を一覧表示するツール。

    Args:
        keyword (str): 検索対象のプロセス名またはキーワード (例: 'python', 'node', 'nginx')

    Returns:
        str: 検索結果テキスト
    """
    # 引数が空文字や空白のみの場合は、無駄な外部コマンド実行を避けるため早期リターン
    if not keyword or not keyword.strip():
        return "検索キーワードを指定してください。"

    keyword = keyword.strip()
    try:
        # OSの 'ps' コマンドを実行し、現在稼働中の全プロセスの PID と コマンド名を取得
        # -e: すべてのプロセスを表示
        # -o pid,comm: 出力列を「プロセスID(PID)」と「実行コマンド名(COMM)」のみに絞り込む
        # capture_output=True: 標準出力と標準エラー出力をキャプチャして取得
        # text=True: 出力結果をバイト列ではなく文字列（str）として受け取る
        # timeout=5: 万が一コマンドがハングアップした場合に備え、5秒で安全にタイムアウトさせる
        result = subprocess.run(
            ["ps", "-eo", "pid,comm"],
            capture_output=True,
            text=True,
            timeout=5
        )
        # コマンドの終了ステータスが 0 以外（異常終了）の場合はエラー内容を返却
        if result.returncode != 0:
            return f"プロセス一覧の取得に失敗しました: {result.stderr}"

        matching = []
        # psの出力結果を改行で分割し、1行ずつ走査（1行目は "PID COMM" の見出しヘッダーなので [1:] でスキップ）
        for line in result.stdout.strip().splitlines()[1:]:
            # 空白で最大2つに分割（PID と コマンドパス/名前）
            parts = line.strip().split(None, 1)
            if len(parts) == 2:
                pid, comm = parts
                # 大文字・小文字の差異を無視してキーワードが含まれているか判定
                if keyword.lower() in comm.lower():
                    # 見やすさのため PID を右寄せ6文字にフォーマットしてリストに追加
                    matching.append(f"PID: {pid.rjust(6)} | Command: {comm}")

        # 一致するプロセスが存在しなかった場合の通知メッセージ
        if not matching:
            return f"キーワード '{keyword}' に一致する実行中プロセスは見つかりませんでした。"

        # AIのコンテキスト（トークン消費量）を圧迫しないよう、上限を20件に制限して結果を整形
        header = f"=== キーワード '{keyword}' の一致プロセス ({len(matching)}件) ==="
        return header + "\n" + "\n".join(matching[:20])

    except Exception as e:
        # 【注意点】stdio通信では標準出力がAIとの通信パイプになるため、通常のprint()を使うと
        # 通信メッセージに余計な文字列が混ざり切断の原因になります。ログは別ルートの「標準エラー出力（sys.stderr）」へ書き出します
        sys.stderr.write(f"search_processes error: {e}\n")
        return f"プロセス検索中にエラーが発生しました: {e}"


@mcp.tool()
def check_port_status(port: int, host: str = "127.0.0.1") -> str:
    """指定されたホストとポート番号の接続状態（LISTEN / 疎通可否）をチェックするツール。

    Args:
        port (int): 調査対象のポート番号 (例: 80, 443, 5432)
        host (str): 調査対象のホスト名またはIPアドレス (デフォルト: '127.0.0.1')

    Returns:
        str: ポートの疎通判定結果テキスト
    """
    # ポート番号がTCP規格の有効範囲（1〜65535）に収まっているかを事前にバリデーション
    if not (1 <= port <= 65535):
        return f"無効なポート番号です: {port} (1〜65535の範囲で指定してください)"

    try:
        # Python標準の socket モジュールでTCPソケットを作成
        # AF_INET: IPv4アドレス体系
        # SOCK_STREAM: TCPプロトコルによる接続
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            # 接続待ちのタイムアウトを2秒に設定（ポートが閉じていても過度に待たされないための配慮）
            sock.settimeout(2.0)
            
            # connect_ex() は例外を発生させず、成功なら 0、失敗ならOSのエラーコード（errno）を返す
            result = sock.connect_ex((host, port))
            
            if result == 0:
                # 接続成功（ポートが開いており、サービスがLISTEN待機中）
                return f"[ポート開放中] {host}:{port} はアクセス可能（サービス待機中 / LISTEN）です。"
            else:
                # 接続失敗（ポートが閉じているか、接続拒否された状態）
                return f"[ポート未開放] {host}:{port} には接続できませんでした (エラーコード: {result})。"

    except Exception as e:
        # エラー発生時も stdio の JSON-RPC 通信を壊さないよう stderr に出力
        sys.stderr.write(f"check_port_status error: {e}\n")
        return f"ポートチェック中にエラーが発生しました: {e}"


# =====================================================================
# 2. Resources（動的コンテキスト参照）
# =====================================================================
# @mcp.resource() デコレータを付与することで、エージェントが受動的に「読み取る（参照する）」データになります。
# MCPの仕様ではリソースを一意なURIで識別するため、開発者が自由に決められる独自の住所（カスタムURI、例: "log://system/error"）を指定します。
# Toolsと違って引数を取らず、アクセスされた時点の最新データや状態を返します。

@mcp.resource("log://system/error")
def get_error_log() -> str:
    """システムの直近のエラーログや警告メッセージを動的に取得するリソース。

    Returns:
        str: 直近のシステムエラーログ
    """
    # リソースが参照された現在時刻を取得してタイムスタンプを動的に作成
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # 実環境では /var/log/syslog やアプリケーションログファイルを読み込みますが、
    # 本デモでは外部設定なしで安全に動作確認できるよう、リアルタイムに動的ログテキストを生成して返します
    logs = [
        f"[{now}] [WARN] [AuthService] 連打によるレートリミット警告を検知しました (Client: 192.168.1.50)",
        f"[{now}] [ERROR] [DatabasePool] 接続タイムアウトが発生しました (DB: PostgreSQL, Host: localhost:5432)",
        f"[{now}] [INFO] [SystemMonitor] メモリ使用率: 45.2% | CPU平均負荷: 0.85",
        f"[{now}] [WARN] [DiskObserver] キャッシュディレクトリの使用容量が80%を超過しています"
    ]
    return "=== システム直近ログ（最新4件） ===\n" + "\n".join(logs)


# =====================================================================
# 3. Prompts（再利用可能なプロンプトテンプレート）
# =====================================================================
# @mcp.prompt() デコレータを付与することで、サーバー側が推奨する「調査の定型プロンプト」を公開できます。
# クライアント側はゼロからプロンプトを考えなくても、このテンプレートに沿ってAIに指示を出すことができます。

@mcp.prompt()
def incident_triage(incident_desc: str) -> str:
    """インシデント発生時の一次切り分け調査をエージェントに指示する標準プロンプト。

    Args:
        incident_desc (str): 発生した障害や不調の概要説明

    Returns:
        str: 一次切り分け用インストラクション
    """
    # ユーザーから受け取った障害内容をテンプレートに埋め込み、
    # リソース確認 ➔ ツール実行 ➔ 推奨アクション報告 の3ステップを明確に指示するプロンプトを返却
    return (
        f"以下のインシデントについて、接続されているMCPツールを活用して一次切り分け調査を実施してください。\n\n"
        f"【インシデント概要】\n{incident_desc}\n\n"
        f"【調査指示】\n"
        f"1. 直近のシステムエラーログ（log://system/error）を確認し、関連するエラーを特定してください。\n"
        f"2. 関連するプロセスやポートの稼働状態をツールで確認してください。\n"
        f"3. 判明した事実と推奨される復旧アクションを箇条書きで報告してください。"
    )


# =====================================================================
# 4. サーバーの待機起動（メインブロック）
# =====================================================================
if __name__ == "__main__":
    # stdio トランスポート（標準入出力）でサーバーを待機起動
    # ※【注意点】
    # 通常の print() を使うと、AIとの通信パイプ（標準出力）に余計な文字列が混ざり、
    # 通信メッセージが壊れて切断の原因になります。ログやメッセージの出力には sys.stderr を使用します。
    sys.stderr.write("ServerOpsMCP starting on stdio transport...\n")
    mcp.run()

# MCP Inspector を起動して自作サーバーをGUIでテスト（Node.js 22以降を推奨）
npx -y @modelcontextprotocol/inspector@latest python3 server_ops_mcp.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Google Antigravity SDK × 自作MCPサーバー連携テスト (test_server_ops_agent.py)
"""

import os
import sys
import asyncio
from pathlib import Path
from google.antigravity import Agent, LocalAgentConfig, types

# .env から APIキーを読み込み
api_key = os.environ.get("GEMINI_API_KEY")

async def main():
    print("=== Google Antigravity SDK × 自作MCPサーバー検証開始 ===\n")

    # 1. 自作MCPサーバースクリプトの絶対パスを解決
    server_script = os.path.abspath("server_ops_mcp.py")

    # 2. McpStdioServer による接続定義
    ops_mcp = types.McpStdioServer(
        name="server_ops_server",
        command=sys.executable,  # 実行中の仮想環境の Python を使用
        args=[server_script]
    )

    # 3. エージェント設計（mcp_servers に自作サーバーを登録）
    config = LocalAgentConfig(
        system_instructions=(
            "あなたは高度なインフラ監視・保守を担当するSREアシスタントです。"
            "接続されているMCPサーバーから提供されるツールを積極的に呼び出し、"
            "正確なプロセス情報やポート稼働状態を調査してエンジニア向けに報告してください。"
        ),
        mcp_servers=[ops_mcp],
        model="gemini-3.1-flash-lite"
    )

    # 4. エージェントへの調査依頼
    user_request = (
        "現在、バックエンドサービス周辺で接続障害の疑いがあります。"
        "PostgreSQLの標準ポート（5432）やWebポート（80）の稼働状態をチェックし、"
        "さらにシステム上で 'python' 関連のプロセスが動いているか調査して報告してください。"
    )

    print(f"【ユーザーからのインシデント調査依頼】:\n{user_request}\n")
    print("▶ エージェントが自作MCPサーバーに接続し、自律的にツールを呼び出して調査を開始します...\n")

    async with Agent(config) as agent:
        response = await agent.chat(user_request)
        final_answer = await response.text()

        print("==================================================")
        print(" 🏆 [自作MCPサーバー連携によるインシデント調査レポート]")
        print("==================================================")
        print(f"{final_answer}\n")

    print("=== MCPサーバー連携タスクが正常に完了しました ===")

if __name__ == "__main__":
    asyncio.run(main())

=== Google Antigravity SDK × 自作MCPサーバー検証開始 ===

【ユーザーからのインシデント調査依頼】:
現在、バックエンドサービス周辺で接続障害の疑いがあります。PostgreSQLの標準ポート（5432）やWebポート（80）の稼働状態をチェックし、さらにシステム上で 'python' 関連のプロセスが動いているか調査して報告してください。

▶ エージェントが自作MCPサーバーに接続し、自律的にツールを呼び出して調査を開始します...

==================================================
 🏆 [自作MCPサーバー連携によるインシデント調査レポート]
==================================================
バックエンドサービスの接続状況およびプロセス稼働状態の調査結果をご報告します。

1. ポート稼働状況
調査の結果、指定されたポートは現在いずれも外部からの接続に応答しない状態です。
- PostgreSQL (ポート 5432): 未開放（接続不可）
- Webポート (ポート 80): 未開放（接続不可）

2. Pythonプロセス稼働状況
システム上で稼働している 'python' 関連のプロセスを検知しました。
- PID: 12045 | Command: /usr/bin/python3
- PID: 12080 | Command: /srv/app/.venv/bin/backend_worker

【考察と推奨アクション】:
ポート 5432 および 80 が開いておらず、データベースやWebサーバーのリスナーが正常に起動していない可能性が高いと考えられます。バックエンドプロセスの設定やサービスログ（log://system/error）を確認し、サービスの再起動を推奨します。

=== MCPサーバー連携タスクが正常に完了しました ===