AIエージェント入門｜ローカルファイル操作AIを作る【第5週・上級編】

第5週・実装編では、MCPの仕組みを理解するために、仲介層（疑似MCP）を自分で実装しました。

しかし実際のAIエージェントでは、本物のMCPサーバーを通じて外部ツールと接続する構成が使われます。

本記事では、Filesystem MCPサーバーを利用し、AIがローカルファイルを操作できるAIエージェントを構築します。

AIエージェントとは

AIエージェントは、判断して動くだけでなく、外部ツールやデータと接続して初めて実用性が高まります。

第5週・上級編では、Filesystem MCPサーバーを利用し、ローカルLLM（Ollama）と接続する方法を解説します。

AIがディレクトリ内のファイル一覧取得やファイル読み取り・作成を行う実用的なエージェントを実装します。

具体的には、次のような流れをコードで実装します。

AI（Ollama）
↓
MCPクライアント（FilesystemMCP Client)
↓
MCPサーバー（mcp_workspace）
↓
ローカルファイル操作
↓
結果
↓
AIに戻す
↓
最終回答

学習ロードマップ（拡張版）

作業環境について

本記事は、第1週の内容を前提として進めています。
そのため、以下の環境がすでに整っている状態を想定しています。

• ローカルLLM実行環境：Ollama
• 使用モデル：qwen3
• Python仮想環境（venv）
• ollama Pythonライブラリのインストール済み

作業環境の準備ができていない場合

第1週の記事で、環境構築からJSON出力までを解説しています。
未実施の場合は、学習ロードマップに従って第1週から進めることをおすすめします。

AIエージェント入門｜ローカルLLMで学ぶチャットとJSON出力【第1週】

第5週・上級編のゴール

• 疑似MCPとの違いを理解する
• 本物のMCPサーバーに接続できる
• AIがローカルファイルを操作できる

ローカルファイル操作MCPの実装

事前準備

このエージェントを動かすために、以下の2つを準備します。

MCP関連ライブラリをインストール

CLI機能を含んだ最新版のModel Context Protocol（MCP）公式SDKをインストールします。

python.exe -m pip install -U ollama "mcp[cli]"

これにより、PythonからMCPサーバーに接続できるようになり、
さらにCLI（コマンドラインツール）を使って、MCPサーバーの動作確認やデバッグが行えるようになります。

ollama

• ローカルLLMをPythonから使うためのライブラリ

mcp[cli]

• MCPサーバーと通信するための公式SDK
• CLIツール（確認・デバッグ用）も含まれる

uv（MCPサーバー実行ツール）をインストール

curl -LsSf https://astral.sh/uv/install.sh | sh

uv は高速なPython実行ツールで、MCPサーバーを簡単に起動できます。

動作イメージ

Python（Pythonコード）
↓
MCP SDK（mcp[cli]）
↓
MCPサーバー（uvで起動）
↓
ファイル操作

実装してみる

完成コード

mcp_client.py

mcp_client.py は、MCPサーバーとやり取りするための専用クラスです。

• MCPサーバー（mcp_workspace）を起動してstdio（標準入出力）で接続
• MCPサーバーが持っている機能を取得 AIして（Ollama）が使える形式に変換
• ツールを実行する
  （AIの指示をMCPサーバーに送り、実際の処理を実行して結果を受け取る）

import os
import sys
from contextlib import AsyncExitStack
from typing import Any

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

class FilesystemMCPClient:
    def __init__(self) -> None:
        self._stack = AsyncExitStack()
        self._session: ClientSession | None = None

    async def connect(self) -> None:
        env = dict(os.environ)
        env["PYTHONIOENCODING"] = "utf-8"
        env["PYTHONUNBUFFERED"] = "1"

        # 実行中の仮想環境の Python 自体を使用してサーバーを起動
        executable = sys.executable 
        # 許可するディレクトリ（絶対パス）
        allowed_dir = os.path.abspath(".")

        server_params = StdioServerParameters(
            command=executable,
            args=[
                "-m", "mcp_workspace", 
                "--project-dir", allowed_dir  # ヘルプに従い --project-dir を指定
            ],
            env=env,
        )

        try:
            stdio_transport = await self._stack.enter_async_context(
                stdio_client(server_params)
            )
            read, write = stdio_transport
            self._session = await self._stack.enter_async_context(
                ClientSession(read, write)
            )
            await self._session.initialize()
            print(f"Filesystem MCP (mcp_workspace) 接続成功: {allowed_dir}", file=sys.stderr)
        except Exception as e:
            print(f"接続エラー詳細: {e}", file=sys.stderr)
            raise

    async def close(self) -> None:
        await self._stack.aclose()

    async def list_tools_for_ollama(self) -> list[dict[str, Any]]:
        if not self._session: return []
        result = await self._session.list_tools()
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema,
                },
            }
            for t in result.tools
        ]

    async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
        if not self._session: return "Error: Not connected"
        try:
            result = await self._session.call_tool(name, arguments=arguments)
            # content からテキストを抽出して結合
            text_contents = []
            for b in result.content:
                if hasattr(b, 'text'):
                    text_contents.append(b.text)
                else:
                    text_contents.append(str(b))
            return "\n".join(text_contents)
        except Exception as e:
            return f"ツール実行エラー: {e}"

MCP_Advance.py

AIエージェント本体（司令塔）で、以下の動作を行います。

• ユーザー入力を受け取る（user_input = input()）
• LLMに考えさせる（ chat(...)）
• ツールが必要なら実行（client.call_tool(...)）
• 会話履歴を蓄積する（messages.append({...})）
• 最終回答は、ツール呼び出しが不要になったタイミングで、最後のchatの結果として返されます。
  （※最終回答専用の処理はなく、AIが「これ以上ツールは不要」と判断した時点で回答を返します。）

import asyncio
import sys
from typing import Any

from ollama import chat
from mcp_client import FilesystemMCPClient

MODEL_NAME = "qwen3"
MAX_STEPS = 5

def normalize_arguments(raw_args: Any) -> dict[str, Any]:
    if isinstance(raw_args, dict):
        return raw_args
    return {}

async def run_agent(user_input: str) -> str:
    client = FilesystemMCPClient()
    await client.connect()

    try:
        ollama_tools = await client.list_tools_for_ollama()

        messages: list[dict[str, Any]] = [
            {
                "role": "system",
                "content": (
                    "あなたはローカルファイルを操作できるAIアシスタントです。"
                    "提供されたツールを使用して、ディレクトリ内のファイル一覧の取得、"
                    "ファイルの内容の読み取り、新しいファイルの作成などが行えます。"
                    "ユーザーの指示に従い、正確に操作を実行してください。"
                ),
            },
            {
                "role": "user",
                "content": user_input,
            },
        ]

        for _ in range(MAX_STEPS):
            response = chat(
                model=MODEL_NAME,
                messages=messages,
                tools=ollama_tools,
            )

            message = response["message"]
            messages.append(message)

            tool_calls = message.get("tool_calls", [])
            if not tool_calls:
                return message.get("content", "") or "回答を生成できませんでした。"

            for tool_call in tool_calls:
                function = tool_call.get("function", {})
                name = function.get("name", "")
                arguments = normalize_arguments(function.get("arguments"))

                # 実行ログを標準エラーに出力
                print(f"\n[MCP tool] {name}", file=sys.stderr)
                print(f"[args] {arguments}", file=sys.stderr)

                result_text = await client.call_tool(name, arguments)

                # 結果のプレビューを表示
                print(f"[result preview] {result_text[:100]}...", file=sys.stderr)

                messages.append(
                    {
                        "role": "tool",
                        "content": result_text,
                    }
                )

        return "処理回数が上限に達しました。"

    finally:
        await client.close()

async def main() -> None:
    print("Filesystem MCP エージェントを起動しました。")
    print("例: '現在のディレクトリのファイルを見せて', 'test.txtの内容を読んで'")

    while True:
        try:
            user_input = input("\nあなた > ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n終了します。")
            break

        if not user_input or user_input.lower() in {"exit", "quit"}:
            break

        answer = await run_agent(user_input)
        print(f"\nAI > {answer}")

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

AIエージェント実行手順

AIエージェント入門｜ローカルLLMで学ぶチャットとJSON出力【第1週】の作業環境確認後に、以下の手順で実行する。

1. 作業ディレクトリへ移動（cd）
2. uvインストール（curl -LsSf https://astral.sh/uv/install.sh | sh ）
3. ollama起動タイムアウト防止を設定（export OLLAMA_KEEP_ALIVE="-1"）
4. ollamaを起動（ollama run qwen3）
5. Python仮想環境をアクティベイト（source .venv/Scripts/activate）
6. MCP公式SDK+CLIをインストール（.venv/Scripts/python.exe -m pip install -U ollama "mcp[cli]"）
7. AIエージェントの起動（.venv/Scripts/python.exe MCP_Advance.py）

テスト入力例

• 現在のディレクトリのファイル一覧を表示して
• hello.txtというファイルを作って「こんにちは」と書いて
• hello.txtの内容を読んで
• not_exist.txtを読んで（存在しないファイルを指定してエラーを確認）
• このディレクトリで一番大きそうなファイルを見つけて内容を要約して

MCP_Advance.pyの処理フロー

MCP_Advance.py は、AIエージェントの本体となるプログラムです。
ユーザーの入力を受け取り、LLMと会話を行い、必要に応じてMCPサーバーを通じて外部処理を実行し、その結果を返します。

関数名を含めた全体の流れ

main()（asyncio.run(main()))
↓
ユーザー入力を受け取る（input()）
↓
run_agent(user_input) を実行
↓
MCPサーバー（mcp_workspace）を起動し、接続する（client.connect()）
↓
利用可能なツールを取得し、AIに渡せる形式に変換する（list_tools_for_ollama()）
↓
chat() でAIが「何をするべきか」を判断する
↓
tool_calls を確認
├─ ツール呼び出しが不要
│   ↓
│   AIの回答（message["content"]）を返して終了
└─ ツール呼び出しが必要
    ↓
    client.call_tool() でMCPサーバー経由の処理を実行
    ↓
    実行結果を messages に追加
    ↓
    次の chat() でAIが再判断
    ↓
    完了するまでループ

まとめ

第5週・上級編では、Filesystem MCPサーバーを使い、AIがローカルファイルを操作できるエージェントを実装しました。

MCPは外部接続の標準

MCPは、AIと外部ツールをつなぐための共通ルールです。
これにより、ファイル操作やWebアクセスなどを統一された方法で扱えるようになります。

AIは判断する役割

AI（ローカルLLM）は、「何をするべきか」「どのツールを使うか」を判断します。
実際の処理は行いません。

MCPが仲介する

MCPクライアント（FilesystemMCPClient）が、「AIの指示を受け取り」「MCPサーバーに伝え」「結果を受け取る」などの AIと外部ツールをつなぐ役割を担います。

外部ツールが実行する

今回の構成では、Filesystem MCPサーバーが実際に処理を行い、以下の様なローカル環境の操作を実現しました。

• ファイル一覧の取得
• ファイルの読み取り
• ファイルの作成

エージェントは「繰り返し」で動く

このエージェントは、判断 → 実行 → 再判断 → 完了というループで動いています。
1回で終わらず、結果を見ながら処理を進めています。