【2026年】『AIエージェント自動開発』完全ガイド：Python×Gemini×MCPサーバー構築と運用最適化

+-------------------------------------------------------------+
|                      Python Host Client                     |
|                                                             |
|   +-------------------+              +------------------+   |
|   |  Gemini 2.5 API   | <==========> |  Orchestrator    |   |
|   | (LLM Reasoning)   |  JSON-RPC    |  (Task State)    |   |
|   +-------------------+              +--------+---------+   |
+-----------------------------------------------|-------------+
                                                | MCP Protocol
                                                | (stdio / SSE)
+-----------------------------------------------v-------------+
|                     Local / Remote MCP Server               |
|                                                             |
|   +-------------------+  +-------------------+  +-------+   |
|   | Tool: SQL Query   |  | Resource: FS Read |  | Prompts   |
|   +-------------------+  +-------------------+  +-------+   |
+-------------------------------------------------------------+

# 仮想環境作成とアクティベート
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 2026年最新SDK群のインストール
pip install google-genai mcp fastmcp pydantic python-dotenv

# server.py
import os
import platform
from fastmcp import FastMCP

mcp = FastMCP(name="DevOps-Agent-Tools")

@mcp.tool()
def get_system_diagnostics() -> dict:
    """実行環境のOS、CPUアーキテクチャ、稼働状態を取得する"""
    return {
        "os": platform.system(),
        "release": platform.release(),
        "architecture": platform.machine(),
        "python_version": platform.python_version()
    }

@mcp.tool()
def list_directory_files(directory_path: str = ".") -> list[str]:
    """指定ディレクトリ内のファイル一覧を返す（セキュリティのため相対パス推奨）"""
    target = os.path.abspath(directory_path)
    # パストラバーサル対策（プロジェクト配下のみ許可）
    if not target.startswith(os.path.abspath(".")):
        raise ValueError("セキュリティ違反: プロジェクト外部へのアクセスは禁止されています。")
    return os.listdir(target)

if __name__ == "__main__":
    # stdio通信でMCPサーバーを起動
    mcp.run(transport="stdio")

# agent.py
import asyncio
import os
from dotenv import load_dotenv
from google import genai
from google.genai import types
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv()
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
client = genai.Client(api_key=GEMINI_API_KEY)

async def run_agent():
    # MCPサーバーの起動パラメータ設定
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env=None
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # MCPサーバーから提供ツール一覧を取得
            mcp_tools = await session.list_tools()

            # GeminiのFunction Declaration形式へマッピング
            gemini_tools = []
            for tool in mcp_tools.tools:
                gemini_tools.append({
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                })

            # ユーザー指示
            task_prompt = "現在のシステム情報を取得し、カレントディレクトリのファイル一覧を確認してレポートしてください。"
            print(f"[Task Initialized]: {task_prompt}")

            # 会話履歴管理
            history = [types.Content(role="user", parts=[types.Part.from_text(task_prompt)])]

            # 思考・自律実行ループ（最大5ターン）
            for step in range(5):
                response = client.models.generate_content(
                    model="gemini-2.5-flash",
                    contents=history,
                    config=types.GenerateContentConfig(
                        tools=[types.Tool(function_declarations=gemini_tools)],
                        temperature=0.1
                    )
                )

                candidate = response.candidates[0]
                history.append(candidate.content)

                # ツール呼び出し判定
                function_calls = [part.function_call for part in candidate.content.parts if part.function_call]

                if not function_calls:
                    # 完了
                    print("\n[Agent Final Answer]:\n", response.text)
                    break

                # ツール実行
                for fc in function_calls:
                    tool_name = fc.name
                    tool_args = dict(fc.args)
                    print(f"  -> Tool Executing: {tool_name}({tool_args})")

                    result = await session.call_tool(tool_name, arguments=tool_args)

                    # 結果を会話履歴に追加
                    history.append(types.Content(
                        role="tool",
                        parts=[types.Part.from_function_response(
                            name=tool_name,
                            response={"result": result.content[0].text}
                        )]
                    ))

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