Zenn
Accenture Japan (有志)
🙌

【Claude MCP】自然言語でAWSリソースを操作する

Koichi Kiriyama
2025/01/03に公開
AWS
boto3
mcp
Claude
AIエージェント
tech

https://www.anthropic.com/news/model-context-protocol

1. はじめに

アクセンチュア株式会社の桐山です。
今回は、独自のMCPサーバーを実装してみましたので、紹介したいと思います。
MCP(Model Context Protocol) とは、以下の記事でも紹介しましたが、LLMと様々なリソース・データソースとを接続する標準的なプロトコルです。

既にClaude MCP公式からは、ファイル操作、ブラウザ操作、データベース操作等のMCPサーバーが公開されており、これらを利用することができます。しかし、その他のリソースと接続するためには、独自のMCPサーバーを実装する必要があります。

今回はその一例として、AWSのEC2を操作するMCPサーバーを実装してみました。

https://zenn.dev/acntechjp/articles/2847a97c673d34

2. ゴール

まず冒頭、今回のゴールを示したいと思います。

  1. Claudeデスクトップで「全てのEC2の状態を確認して」のプロンプトにより、EC2の状態を確認することができました！

  2. 「ec2-lnx-01を起動して」のプロンプトにより、対象のEC2を起動することができました！

  3. 「全てのEC2を停止して」のプロンプトにより、対象のEC2を停止することができました！

3. インストール

今回は、Windows上のuvによる仮想環境で進めていきたいと思います。

  1. uvをインストール
winget install --id=astral-sh.uv -e
  1. create-mcp-serverでMCPサーバーのテンプレートを作成
    下記コマンド入力後、プロジェクト名やバージョン等求められますので、指示に従い入力を進めます
uvx create-mcp-server --path boto3-ec2
create-mcp-server で生成されたテンプレート (\<プロジェクト名>\src\server.py)
import asyncio

from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio

# Store notes as a simple key-value dict to demonstrate state management
notes: dict[str, str] = {}

server = Server("boto3-ec2")

@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    """
    List available note resources.
    Each note is exposed as a resource with a custom note:// URI scheme.
    """
    return [
        types.Resource(
            uri=AnyUrl(f"note://internal/{name}"),
            name=f"Note: {name}",
            description=f"A simple note named {name}",
            mimeType="text/plain",
        )
        for name in notes
    ]

@server.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
    """
    Read a specific note's content by its URI.
    The note name is extracted from the URI host component.
    """
    if uri.scheme != "note":
        raise ValueError(f"Unsupported URI scheme: {uri.scheme}")

    name = uri.path
    if name is not None:
        name = name.lstrip("/")
        return notes[name]
    raise ValueError(f"Note not found: {name}")

@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    """
    List available prompts.
    Each prompt can have optional arguments to customize its behavior.
    """
    return [
        types.Prompt(
            name="summarize-notes",
            description="Creates a summary of all notes",
            arguments=[
                types.PromptArgument(
                    name="style",
                    description="Style of the summary (brief/detailed)",
                    required=False,
                )
            ],
        )
    ]

@server.get_prompt()
async def handle_get_prompt(
    name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
    """
    Generate a prompt by combining arguments with server state.
    The prompt includes all current notes and can be customized via arguments.
    """
    if name != "summarize-notes":
        raise ValueError(f"Unknown prompt: {name}")

    style = (arguments or {}).get("style", "brief")
    detail_prompt = " Give extensive details." if style == "detailed" else ""

    return types.GetPromptResult(
        description="Summarize the current notes",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(
                    type="text",
                    text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
                    + "\n".join(
                        f"- {name}: {content}"
                        for name, content in notes.items()
                    ),
                ),
            )
        ],
    )

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    List available tools.
    Each tool specifies its arguments using JSON Schema validation.
    """
    return [
        types.Tool(
            name="add-note",
            description="Add a new note",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["name", "content"],
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    Handle tool execution requests.
    Tools can modify server state and notify clients of changes.
    """
    if name != "add-note":
        raise ValueError(f"Unknown tool: {name}")

    if not arguments:
        raise ValueError("Missing arguments")

    note_name = arguments.get("name")
    content = arguments.get("content")

    if not note_name or not content:
        raise ValueError("Missing name or content")

    # Update server state
    notes[note_name] = content

    # Notify clients that resources have changed
    await server.request_context.session.send_resource_list_changed()

    return [
        types.TextContent(
            type="text",
            text=f"Added note '{note_name}' with content: {content}",
        )
    ]

async def main():
    # Run the server using stdin/stdout streams
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="boto3-ec2",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )
  1. 必要なライブラリをインストール
uv add boto3 python-dotenv

4. 実装

実装と言いましても、今回は Claude 3.5 Sonnet にコードを書いてもらいました。
Claude 3.5 Sonnet のカットオフは2024年4月ですので、MCPについて学習されていません。
そこで、以下のMCP公式情報やサンプルコードを追加のナレッジとして Claude 3.5 Sonnet に与えました。

https://modelcontextprotocol.io/quickstart/server
https://spec.modelcontextprotocol.io/specification/server/

以下が Claude 3.5 Sonnet により生成されたMCPサーバーのPythonコードとなります。(<プロジェクト名>\src\server.py)

import os
import json
import logging
from typing import Any, Sequence
import boto3
from botocore.exceptions import ClientError
from dotenv import load_dotenv
from mcp.server import Server
from mcp.types import (
    Resource,
    Tool,
    TextContent,
    EmbeddedResource
)
from pydantic import AnyUrl

# 環境変数の読み込み
load_dotenv()

# ログの設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ec2-mcp-server")

# 必要な環境変数の確認
required_env_vars = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]
missing_env_vars = [var for var in required_env_vars if not os.getenv(var)]
if missing_env_vars:
    raise ValueError(f"Missing required environment variables: {', '.join(missing_env_vars)}")

# AWSセッションの設定
session = boto3.Session(
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
    region_name=os.getenv("AWS_REGION", "ap-northeast-1")
)

# EC2クライアントの初期化
ec2_client = session.client('ec2')

# MCPサーバーの初期化
app = Server("ec2-service-server")

def get_instance_details(instance):
    """EC2インスタンスの詳細情報を取得"""
    name = ''
    for tag in instance.get('Tags', []):
        if tag['Key'] == 'Name':
            name = tag['Value']
            break

    return {
        'instance_id': instance['InstanceId'],
        'name': name,
        'instance_type': instance['InstanceType'],
        'state': instance['State']['Name'],
        'private_ip': instance.get('PrivateIpAddress', ''),
        'public_ip': instance.get('PublicIpAddress', ''),
        'launch_time': instance['LaunchTime'].isoformat()
    }

@app.list_resources()
async def list_resources() -> list[Resource]:
    """利用可能なEC2リソースの一覧を返す"""
    try:
        # EC2インスタンスの一覧を取得
        response = ec2_client.describe_instances()
        resources = []
        
        for reservation in response['Reservations']:
            for instance in reservation['Instances']:
                instance_details = get_instance_details(instance)
                uri = AnyUrl(f"ec2://{instance['InstanceId']}")
                resources.append(
                    Resource(
                        uri=uri,
                        name=f"EC2 Instance: {instance_details['name'] or instance['InstanceId']}",
                        mimeType="application/x-ec2-instance",
                        description=f"Type: {instance['InstanceType']}, State: {instance['State']['Name']}"
                    )
                )
        
        return resources
    except ClientError as e:
        logger.error(f"Failed to list EC2 instances: {str(e)}")
        return []

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    """指定されたEC2インスタンスの詳細情報を返す"""
    if not str(uri).startswith("ec2://"):
        raise ValueError(f"Unsupported URI scheme: {uri}")
    
    try:
        instance_id = str(uri)[6:]
        response = ec2_client.describe_instances(
            InstanceIds=[instance_id]
        )
        
        if not response['Reservations']:
            raise ValueError(f"Instance not found: {instance_id}")
            
        instance = response['Reservations'][0]['Instances'][0]
        instance_details = get_instance_details(instance)
        
        return json.dumps(instance_details, indent=2)
            
    except ClientError as e:
        logger.error(f"Failed to read EC2 instance: {str(e)}")
        raise RuntimeError(f"EC2 access error: {str(e)}")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なEC2ツールの一覧を返す"""
    return [
        Tool(
            name="list_instances",
            description="List EC2 instances by state",
            inputSchema={
                "type": "object",
                "properties": {
                    "state": {
                        "type": "string",
                        "description": "Instance state (running, stopped, all)",
                        "enum": ["running", "stopped", "all"]
                    }
                },
                "required": ["state"]
            }
        ),
        Tool(
            name="start_instance",
            description="Start a stopped EC2 instance",
            inputSchema={
                "type": "object",
                "properties": {
                    "instance_id": {
                        "type": "string",
                        "description": "EC2 instance ID"
                    }
                },
                "required": ["instance_id"]
            }
        ),
        Tool(
            name="stop_instance",
            description="Stop a running EC2 instance",
            inputSchema={
                "type": "object",
                "properties": {
                    "instance_id": {
                        "type": "string",
                        "description": "EC2 instance ID"
                    }
                },
                "required": ["instance_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | EmbeddedResource]:
    """指定されたEC2ツールを実行"""
    if name == "list_instances":
        try:
            filters = []
            if arguments["state"] != "all":
                filters.append({
                    'Name': 'instance-state-name',
                    'Values': [arguments["state"]]
                })
            
            response = ec2_client.describe_instances(Filters=filters)
            instances = []
            
            for reservation in response['Reservations']:
                for instance in reservation['Instances']:
                    instances.append(get_instance_details(instance))
            
            return [
                TextContent(
                    type="text",
                    text=json.dumps(instances, indent=2)
                )
            ]
        except ClientError as e:
            raise RuntimeError(f"EC2 list error: {str(e)}")
            
    elif name == "start_instance":
        try:
            response = ec2_client.start_instances(
                InstanceIds=[arguments["instance_id"]]
            )
            return [
                TextContent(
                    type="text",
                    text=json.dumps({
                        "message": f"Starting instance {arguments['instance_id']}",
                        "state_change": response['StartingInstances'][0]['CurrentState']['Name']
                    }, indent=2)
                )
            ]
        except ClientError as e:
            raise RuntimeError(f"EC2 start error: {str(e)}")
            
    elif name == "stop_instance":
        try:
            response = ec2_client.stop_instances(
                InstanceIds=[arguments["instance_id"]]
            )
            return [
                TextContent(
                    type="text",
                    text=json.dumps({
                        "message": f"Stopping instance {arguments['instance_id']}",
                        "state_change": response['StoppingInstances'][0]['CurrentState']['Name']
                    }, indent=2)
                )
            ]
        except ClientError as e:
            raise RuntimeError(f"EC2 stop error: {str(e)}")
    
    else:
        raise ValueError(f"Unknown tool: {name}")

async def main():
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Claudeデスクトップの claude_desktop_config.json に、以下を指定します。(パスは任意)

{
  "mcpServers": {
    "boto3-ec2": {
      "command": "uv",
      "args": [
        "--directory",
        "C:\\MCP\\boto3-ec2",
        "run",
        "boto3-ec2"
      ],
      "env": {
        "AWS_ACCESS_KEY_ID": <YOUR_AWS_ACCESS_KEY_ID>,
        "AWS_SECRET_ACCESS_KEY": <YOUR_AWS_SECRET_ACCESS_KEY>
      }
    }
  }
}

5. さいごに

いかがでしたでしょうか。
今回は、Claude MCPを活用してAWS EC2を操作する独自MCPサーバーの実装を試してみました。
今回試しましたEC2の操作はあくまで一例であり、MCPを活用することでLLMからさまざまなリソースを操作することが可能となります。（制約はありますが）

次回以降、MCPによる他のリソース操作も試してみたいと思います。

Accenture Japan (有志)
Accenture Japan (有志)

アクセンチュア株式会社所属の社員有志による運営です。アクセンチュアの社員による様々な発信をまとめています。なお、投稿内容は社員個人の見解であり、所属する組織を代表するものではありません。

Discussion