0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

2026年!初アウトプットをしよう
@BrandonWatanabe(つきぼし)

【初めまして】初心者によるMCPとADKの紹介

0
Posted at

※初投稿になります、初めに挨拶をさせてください。

はじめに

初めまして、つきぼしと申します。
自社の活動の一環で、人生初めての記事を作成します。
元々1人のIT人材として、アウトプットをする・ノウハウを溜めるという意味でも、記事投稿には興味ありました。
これをきっかけに、自分が興味あるもの、面白いと思ったもの、自分の経験段、加えて役に立つものを中心に投稿できればと思います。
主にAI技術やDX系が好きです。

内容

初投稿の本記事では、初めて触れたMCPの知識と、Google ADK (Agent Development Kit)のフレームワークを使った基礎的な実装をまとめる。

MCPとは？

MCP (Model Context Protocol) とは、生成AIモデルに文脈情報を渡しやすくするための規格(AIエージェント用のツール接続規格(お約束ごと))である。Anthropic社が2024年11月に発表し、簡単に使えるSDKと一緒に公開した。

嬉しいこと：AIエージェントが様々なツール (Web検索、ファイル作成、メール送信、など)を使用するための導入・実装方法を統一化した。
　→ツールの細かい仕様を気にすることなく、簡単かつ安全に接続して利用できるようになった。

この先の登場人物

MCPクライアント：ツールを使う側、AIエージェント
MCPサーバー：ツールやリソースを提供する側

※補足
・MCPサーバーはクライアントアプリ内で動作するケースが多い。
　・ローカルMCPサーバー、プラグイン拡張機能的な使い方

・遠隔(リモート)のMCPサーバーの場合、クライアント側からはHTTP通信で、サーバー側からはSSE通信で接続している。(旧方式)
　→新方式ではStreamable HTTP通信を使用

ADKにおけるMCPの実装

AIエージェントでは、FastMCPと呼ばれるライブラリを使用している。
以下コマンドでライブラリをインストールする。

requirements.txt
pip install fastmcp==2.12.4

FastMCPの基本的な使い方(サーバー側)

from fastmcp import FastMCP        # FastMCPのライブラリをインポート

mcp = FastMCP("Demo")              # サーバーの本体 (左記の引数はname)

@mcp.tool                          # サーバーのツールを宣言 (mcp.tool(...)でツールの仕様(名前、説明)も定義可能)
def add(a: int, b: int) -> int:    # ツールの実際の処理を関数で定義
    """Add two numbers"""
    return a + b

if __name__ == "__main__":
    mcp.run()                      # サーバーを起動

具体的な実装 (簡易)

例として、Microsoft Sentinelを紹介する。

サービス名　　　　 開発元　　　　　 説明・役割 MCP化の目的
Sentinel Microsoft 高度なセキュリティ監視・分析・プラットフォーム(SIEM、SOARなど)
ログを集約しAIで脅威を分析・検出・対応する		 セキュリティ状況を問い合わせるため

1. サーバー側

MCPサーバーはDockerを使用し、コンテナとして起動されている。

__main__.py
# 1. MCPサーバーの定義
from fastmcp import FastMCP
from decouple import config

mcp = FastMCP(
    name="Sentinel-MCP-Server",
    instructions="このMCPサーバーはMicrosoft Sentinelのインシデント情報を取得するためのものです。",
    json_response=True,
)

FASTMCP_PORT: Final[int] = config("FASTMCP_PORT", 8003, cast=int)

# 2. ツールの宣言、定義
@mcp.tool(
    name="get_incident_entity_info",
    annotations={
        "title": "Sentinel Entity MCP",
        "readOnlyHint": True,
        "openWorldHint": True,
    },
    enabled=True,
)
def get_incident_entity_info(
    incident_id: Annotated[str, Field(description="インシデントID")],
) -> str:
    """指定されたインシデントIDのエンティティ情報を取得する

    インシデントIDは「XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX」形式の英数字で構成される文字列です。

    Args:
        incident_id (str): インシデントID
            Example:
                "11111111-2222-3333-4444-555555555555"

    Returns:
        str: インシデントのエンティティ情報を含むJSON文字列。エラーが発生した場合はエラーメッセージが含まれます。
    """
    if not incident_id:
        error_message = "incident_id required"
        return error_message
    
    client_id = os.getenv("AAA")
    client_secret = os.getenv("BBB")
    tenant_id = os.getenv("CCC")
    
    try:
        token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
        token_payload = {
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "resource": "https://management.azure.com",
        }
        token_response = requests.post(token_url, data=token_payload)
        token_response.raise_for_status()
        
        access_token = token_response.json().get("access_token")
        if not access_token:
            error_message = "Access token not found in the response."
            return error_message
            
        subscriptionId = os.getenv("XXX")
        resourceGroupName = os.getenv("YYY")
        workspaceName = os.getenv("ZZZ")
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-type": "application/json",
        }
        url = f"https://management.azure.com/subscriptions/{subscriptionId}/resourceGroupName/{resourceGroupName}/providers/Microsoft.OperationalInsights/workspaces/{workspaceName}/providers/Microsoft.SecurityInsights/incidents/{incident_id}/entities?api-version=2023-11-01"
        response = requests.post(url, headers=headers)
        response.raise_for_status()
        
        return json.dumps(reponse.json(), ensure_ascii=False, indent=2)
        
    except requests.RequestsException as e:
        error_message = "An error occurred while processing your request. Please try again later."
        return error_message

# 3. サーバーの起動
if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=FASTMCP_PORT,
        path""/sentinel-mcp,
    )
  1. MCPサーバーの定義
    1. 必要ライブラリをインポート
    2. サーバー本体をインスタンス化
    3. ポート番号を宣言
  2. ツールの宣言・定義
    1. デコレータを定義 (以下引数を指定)
      1. name: ツールの名前
      2. annotations: アノテーション
      3. enabled: 活性化パラメータ (True/False)
    2. 実際の処理を関数で定義
  3. サーバーの起動
    1. run()メソッドのパラメータを指定し実行

2. クライアント側

各サービスで使用するツールを定義し、AIエージェント側にて登録する。

tools.py
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams

toolsets = [
    MCPToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="http://localhost:8003/sentinel-mcp",
        ),
    ),
]
  1. 必要ライブラリをインポート
  2. 各MCPサーバーをリスト形式で定義
    1. サーバー側のポートを指定し、URLを設定
agent.py
from google.adk.agents import LlmAgent
from .tools import toolsets

agent = LlmAgent(
    name="mcp_client_agent",
    model="gemini-2.5-flash",
    description="インシデント調査のためのツール実行計画を立て、ツールの実行を行います。",
    instruction="SYSTEM_PROMPT",
    tools=toolsets,
)
  1. 必要ライブラリをインポート
  2. 対象のAIエージェントにMCPサーバーを指定しインスタンス化
    1. toolsパラメータにツール群を指定

余談

仕事を通してMCPに触れたが、今のシステムではもう使っていない...要件にかなわず...

最後に

簡単な記事になってますが、初めて書けたということで嬉しいです。
これからもどうぞよろしくお願いします！

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?