ローカル AI システムの構築：Qwen3.6 と MCP の活用

![image](https://www.kdnuggets.com/wp-content/uploads/KDN-Shittu-Buildu-Building-Local-AI-Systems-Qwen-MCPs-scaled.png) ## MCP の紹介 ローカル AI を活用して開発を行うエンジニアは、誰もが最終的に同じ壁にぶつかります。モデル自体は優秀で、推論能力も高く、堅牢なコードを記述し、複雑な質問にも答えてくれます。しかし、万能ではありません。データベースへの照会や GitHub のイシュー作成、社内 API への呼び出しといったタスクを実行することはできません。 その結果、必要なツールごとに独自のカスタム Python ラッパーを作成し、モデルの出力とツールの実行をつなぐ接着剤をハードコードし、API が変更されるたびにそれらのラッパーを維持管理するという作業に追われることになります。 この課題を解決するために考案されたのが [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) です。これは Anthropic によるオープンスタンダードであり、AI ツール接続のためのユニバーサルでプラグ可能なプロトコルです。ツールは MCP サーバーとして一度定義するだけで、MCP に準拠したクライアントであれば、どのモデルやフレームワークでも、モデルごとのカスタム統合コードをゼロにしながら、そのツールを検索して呼び出すことが可能になります。 現在、この種のタスクにおいて最も能力が高いローカルモデルが [Qwen3.6-35B-A3B](https://huggingface.co/Qwen/Qwen3.6-35B-A3B) です。262,144 トクンのコンテキストウィンドウを持ち、Mixture of Experts (MoE) アーキテクチャを採用しています。このアーキテクチャにより、1 つのフォワードパスで 350 億パラメータのうち 30 億パラメータのみが活性化します。そのため、本来は 35B モデルを動作させることが不可能なハードウェアでも実行可能なのです。さらに、MCP ベースのエージェントタスクに対して明示的にトレーニングされ、評価されています。 この記事では、ローカル環境で動作する GitHub 開発者アシスタントの構築方法を解説します。このエージェントはリポジトリ内の未解決課題を読み込み、関連コードを検索し、修正案を作成してプルリクエストを提出する一連の作業を担います。すべてがあなたのハードウェア上で完結し、MCP サーバーを経由してクラウドへの依存なしに動作します。 ## Qwen3.6-35B-A3B の仕組みを理解する このモデルのアーキテクチャを理解することは重要です。なぜなら、それが必要なハードウェア要件や、エージェントタスクにおけるパフォーマンスの根拠を直接説明しているからです。 モデル名には重要な情報が含まれています。「35B」は総パラメータ数が 350 億であることを示し、「A3B」は1回の推論パスで活性化されるパラメータが 30 億であることを意味します。これは MoE（Mixture of Experts）アーキテクチャを採用しており、[各層に 256 のエキスパートを持ち、トークンごとに 8 つのエキスパートと 1 つの共有エキスパートをルーティング](https://huggingface.co/Qwen/Qwen3.6-35B-A3B) する仕組みです。これにより、350 億パラメータモデル並みの知識容量を持ちながら、推論にかかる計算コストは 30 億パラメータモデルレベルに抑えられます。このトレードオフこそが、密結合型の 35B モデルでは処理しきれないハードウェアでも動作可能にする理由です。 Qwen3.6 が他の MoE モデルと最も異なる点は、その内部レイアウトにあります。40層のスタックを構成する各ブロックは、ゲート付き DeltaNet レイヤーとゲート付きアテンションレイヤーが 3:1 の比率で配置されています。DeltaNet は線形アテンション機構であり、完全な二次関数アテンションに比べてシーケンス処理が効率的です。特に長いコンテキスト長においてその差は顕著になります。一方、完全にゲートされたアテンションレイヤーが組み合わさることで、線形アテンションだけでは捉えきれない深い関係性の推論が可能になります。500 件以上のファイルからなるリポジトリを扱うエージェントにとって、この組み合わせは決定的に重要です。長い文脈を効率的に処理しつつ、必要な箇所については精密な推論を行うことができるからです。 コンテキストウィンドウはネイティブで 262,144 トークンをサポートし、YaRN スケーリングを用いれば最大 101 万トークンまで拡張可能です。エージェント処理において、この長さは単なる快適機能ではなく、運用上の制約そのものです。ソースファイルの参照、ツール呼び出し履歴の維持、多段階プランの追跡、そして結果をコンテキストへ再挿入といった作業には、十分な余裕が不可欠です。多くの 7B や 13B モデルはトークン数が 8k または 32k で上限に達しており、タスク途中でコンテキスト不足に陥ると、エージェントは自身の履歴を失い、ツール結果の捏造（ハルシネーション）を始めかねません。 Qwen3.6 は MCP ベースのエージェントベンチマークに基づいて明示的に訓練・評価されました。この訓練から浮き彫りになった 2 つの主要機能があります。 - エージェントによるコーディング：フロントエンドワークフローやリポジトリ全体の推論を処理します。単一のファイル編集に留まらず、複数のファイルをまたぐリファクタリングタスクでも一貫した推論が可能です。 - 思考の保持（Thinking Preservation）：「preserve_thinking」フラグにより、多段階会話における過去の推論トレースが維持されます。エージェントが 1 巡目でプランを推論し、2〜5 巡目でツール呼び出しを実行する場合、このフラグを有効にすると、1 巡目の推論結果が KV キャッシュ内に保持され続けます。その後の各ステップで、過去の推論を活用できるため、再計算というコストを負担することなく処理を進められます。 ## # システム要件 実用的なデプロイメントには 3 つの現実的なパスがあり、いずれを選ぶかはハードウェア次第です。 - GPU 推論（本番環境のエージェントワークロードに推奨）: Qwen3.6-35B-A3B を bfloat16 で動作させるには約 70 GB の VRAM が必要です。Q4 量子化であれば、20〜24 GB で収まります。RTX 4090（24 GB）1 枚で Q4 モデルを処理可能ですし、Tensor Parallelism を用いた RTX 3090 2 枚でも同様に動作します。A100 80 GB はフルサイズの bfloat16 モデルを扱えます。 - CPU またはハイブリッド構成（KTransformers 経由）: 24 GB の GPU を持っていない開発者にとって、KTransformers が手軽な選択肢です。利用可能な場合は計算負荷の高い層を GPU にオフロードし、残りを CPU で処理します。システム RAM が 64 GB あれば、Qwen3.6-35B-A3B を実用的に（ただし低速で）動作させることが可能です。応答レイテンシは 1 ターンあたり 30〜120 秒程度となり、CPU の性能によります。これはバックグラウンドでのリポジトリ分析を行うエージェントには許容範囲ですが、対話型のコーディングセッションでは使い物になりません。 - チュートリアル用の小規模モデル: この記事で紹介する MCP 統合パターンは、モデルのサイズに関わらず同じです。もし 35B モデルを動かすためのハードウェアがない場合は、Ollama を使って Qwen/Qwen2.5-7B-Instruct（コマンド：ollama pull qwen2.5:7b）や Qwen3-8B モデルを使用してください。サービング API もコードも同じなので、ハードウェアが許すようになったら 35B モデルに差し替えるだけで済みます。 ソフトウェア要件: # Python 3.11 以上が必要です python --version python -m venv qwen-mcp-env source qwen-mcp-env/bin/activate # macOS / Linux qwen-mcp-env\Scripts\activate # Windows # コアパッケージ pip install \ "openai>=1.30.0" \ "qwen-agent>=0.0.10" \ "mcp>=1.0.0" \ "httpx>=0.27.0" # サービングフレームワークの選択 NVIDIA GPU を使用する場合は、以下のいずれかをインストールしてください。 - `vllm>=0.19.0` - `sglang>=0.5.10`（長文コンテキストでの事前処理速度が速い） CPU またはハイブリッド環境の場合は `ktransformers` を使用します。 また、MCP サーバーを npx でインストールする場合は Node.js 18 以上が必要です。バージョン確認には以下のコマンドを使用してください。 node --version ## OpenAI 互換 API を備えたローカル環境で Qwen3.6 を実行 MCP サーバーを設定する前に、まず推論サーバーを起動しておく必要があります。SGLang と vLLM の両方は OpenAI 互換の API を提供しており、MCP 統合層はこの API に接続します。API の仕様は同じですが、エンドポイントは `api.openai.com` ではなくローカルの localhost を指す点だけが異なります。 ## // SGLang（長文コンテキストを扱うエージェントワークロードに推奨） SGLang をフル機能でインストールするには以下のコマンドを実行します。 pip install "sglang[all]>=0.5.10" Qwen3.6-35B-A3B モデルを、推論解析とツール呼び出しパーサーを有効にした状態で起動します。以下のオプションが重要です。 - `--reasoning-parser qwen3`：... ブロックの処理を正しく行います。 - `--tool-call-parser qwen3_coder`：ツール呼び出しの出力を適切な形式にルーティングします。 - `--enable-prefix-caching`：エージェントワークロードにおいて必須です。これにより、ターン間で KV キャッシュが再利用され、思考プロセスの効率的な維持が可能になります。 python -m sglang.launch_server \ --model-path Qwen/Qwen3.6-35B-A3B \ --host 0.0.0.0 \ --port 30000 \ --reasoning-parser qwen3 \ --tool-call-parser qwen3_coder \ --enable-prefix-caching \ --tp 2 # 2 枚の GPU にわたるテンソル並列化。単一 GPU の場合はこのオプションを削除 ## // vLLM pip install "vllm>=0.19.0" # vLLM 同等の環境を同じ重要なフラグで構築 vllm serve Qwen/Qwen3.6-35B-A3B \ --host 0.0.0.0 \ --port 8000 \ --reasoning-parser qwen3 \ --tool-call-parser qwen3_coder \ --enable-prefix-caching-v2 \ --tensor-parallel-size 2 ## // Ollama を用いた軽量モデルの活用 ollama pull qwen2.5:7b ollama serve # Ollama の API は http://localhost:11434/v1 で OpenAI 互換です サーバー起動後、これ以上進む前に必ず動作確認を行ってください。 # ヘルスチェック -- {"status": "ok"} または同等のレスポンスが返るはずです curl http://localhost:30000/health # 簡単なクエリでチャット完了エンドポイントをテスト curl http://localhost:30000/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "Qwen/Qwen3.6-35B-A3B", "messages": [{"role": "user", "content": "Reply with: ready"}], "max_tokens": 10 }' JSON 応答に choices アレイが含まれていれば、サーバーは正常に稼働しています。MCP の設定に進む前に、必ずこれが動作していることを確認してください。サービング層が安定していれば、その後の統合で発生するあらゆる不具合も、原因特定が容易になります。 ## # MCP とエージェントアーキテクチャの変容を理解する エージェントコードを書く前に、プロトコルレベルで MCP が実際に何を行うのか理解しておくことが重要です。MCP を単なる高度な関数呼び出し API と捉えることから生じるバグの一種を未然に防ぐためです。 MCP は、stdio または HTTP トランスポート上で動作する JSON-RPC 2.0 プロトコルです。MCP クライアントがサーバーに接続すると、まず tools/list を呼び出して、サーバーが提供するツールを検索します。各ツールからは、名前、説明、そして JSON Schema で定義された入力スキーマが返されます。モデルはこのスキーマを読み込み、これがモデルとツールの契約となります。 モデルがツールを呼び出したい場合、構造化されたツール呼び出しオブジェクトを生成します。ただし、実際に呼び出しを実行するのは MCP クライアントであり、サーバーに対して tools/call リクエストを送信します。サーバー側で実行が行われ、結果が返されます。その結果はクライアントによってツールロールメッセージとして会話に挿入され、モデルがそれを読み取って次のステップを決定します。 この分離構造は重要です。モデルは「何を」「どの引数で」呼び出すかを判断し、クライアントが実行を担当し、サーバーが実際の処理を行います。コード側でツールとモデルを硬く結びつける必要はなく、利用可能なサーバーをクライアントに指定するだけで済みます。 Qwen3.6 と MCP を組み合わせるには、主に以下の 2 つの方法があります。 - Qwen-Agent を経由する方法：公式の qwen_agent ライブラリが、ツールの検出、呼び出しの解析、結果の挿入、複数ターンにわたる会話管理を自動的に処理します。コード量が少なく制御性は低くなりますが、多くのユースケースに適しています。 MCP Python SDK を直接利用するアプローチでは、`mcp.ClientSession` を用いてエージェントループを自分で実装します。コード量は増えますが、すべてのメッセージを可視化でき、エラーハンドリングやリトライロジックを完全に制御できます。これは各ステップの監視が必要な本番環境システムに適しています。 この記事では、Qwen-Agent から始める両方のアプローチを取り上げます。 ## ローカル GitHub 開発アシスタントの構築 このエージェントは、以下の 4 つの手順を順に実行します。GitHub リポジトリから未解決のイシューを読み込み、関連するコードを検索し、修正草案を作成してプルリクエストを開くものです。すべてローカル環境で、MCP を通じて完結します。 ## // Part 1: 環境と MCP サーバーの設定 # GitHub のパーソナルアクセストークンを設定してください # API 呼び出しには GitHub MCP サーバーが必要となります export GITHUB_TOKEN=ghp_your_token_here # npx を用いた事前構築済み MCP サーバーのインストール # 個別のインストール手順は不要です。エージェントがサーバーを開始する際に、npx が初回使用時に自動処理します # npx の利用可否を確認するには： npx --version プロジェクトディレクトリを作成します。 mkdir qwen-github-agent cd qwen-github-agent ## // Part 2: Qwen-Agent の実装 動作するエージェントへの最速ルートです。Qwen-Agent は、ループ処理全体を自動的に処理します。 # github_agent_qwenagent.py # 事前準備：pip install qwen-agent openai # MCP サーバーには npm / npx のインストールが必要です # GITHUB_TOKEN 環境変数の設定が必要です # ローカルサーバーエンドポイントが稼働している必要があります（前節を参照） # # 実行方法： # python github_agent_qwenagent.py from qwen_agent.agents import Assistant # ── サーバー設定 ────────────────────────────────────────────────────── # ローカルで動作するエンドポイントへのパスを指定します。 # 起動したサーバーに合わせて base_url を変更してください： # SGLang: http://localhost:30000/v1 # vLLM: http://localhost:8000/v1 # Ollama: http://localhost:11434/v1 LLM_CONFIG = { "model": "Qwen/Qwen3.6-35B-A3B", "model_server": "http://localhost:30000/v1", "api_key": "EMPTY", # ローカルサーバーでは実際のキーは不要 # 思考モードのサンプリングパラメータ（公式モデルカードのベストプラクティスに基づく） "generate_cfg": { "temperature": 0.6, "top_p": 0.95, "top_k": 20, "min_p": 0.0, "thought_in_history": True, # これは Qwen-Agent における preserve_thinking フラグに相当 }, } # ── MCP サーバー設定 ────────────────────────────────────────────────── # 各サーバーのキーはサーバー名を表し、値には stdio 経由で起動するコマンドを記述します。 # Qwen-Agent は各サーバーをサブプロセスとして起動し、MCP セッションを管理します。 MCP サーバーの設定例 以下は、ファイルシステムと GitHub へのアクセスを可能にする MCP サーバーの構成です。 - `filesystem` サーバー：現在の作業ディレクトリ（`.`）へのアクセス権限を与えます。本番環境では、特定のレポジトリパスに制限することを推奨します。 - `github` サーバー：GitHub API 認証のために環境変数 `GITHUB_TOKEN` を使用します。 システムプロンプトの例 このプロンプトは、MCP ツールを通じて GitHub レポジトリにフルアクセスできるシニアソフトウェアエンジニアとして AI に指示を出すためのものです。 具体的なタスクを実行する際は、以下の手順に従います： 1. 解決すべき課題を把握するため、オープンなイシューを一覧表示する。 2. ファイルシステムツールを使用して、関連するソースコードやテストファイルを読み込む。 3. コードとイシューの説明に基づき、根本原因を特定する。 4. バグに直接対応した最小限の変更を行い、不要なリファクタリングは行わない修正コードを作成する。 5. イシューへの言及を含む明確なタイトルと説明でプルリクエストを作成する。 各ステップでは必ずその理由を説明してください。コードを実装する前に、エッジケースについても十分に検討してください。ファイルの目的が不明確な場合は、変更を加える前に必ず内容を確認します。 ## エージェントの設定 エージェントは、ローカル環境で MCP（Model Context Protocol）を通じて動作します。具体的には、GitHub の開発者アシスタントとして機能し、課題の読み込み、バグ修正、プルリクエストの作成を行います。 ```python agent = Assistant( llm=LLM_CONFIG, name="GitHub Developer Assistant", description="Reads issues, fixes bugs, opens pull requests -- locally via MCP.", system_message=SYSTEM_PROMPT, mcp_servers=MCP_SERVERS, ) ``` ## エージェントの実行 タスクを実行する関数 `run_agent` は、指定されたタスクの説明に基づいてエージェントを起動し、結果をストリーミング出力します。Qwen-Agent がツール呼び出しの自動実行から結果の注入まで、一連のプロセスを完全に管理します。 ```python def run_agent(task: str): """ Run the agent on a task description and stream the output. The agent will make tool calls automatically; Qwen-Agent handles the full loop including tool execution and result injection. """ messages = [{"role": "user", "content": task}] print(f"Task: {task}\n{'─' * 70}") # Qwen-Agent's run() is a generator that yields intermediate steps # Each yielded message shows a tool call, a tool result, or the final answer for response in agent.run(messages=messages): # response is a list of messages representing the conversation so far # The last message contains the most recent output last = response[-1] role = last.get("role", "") content = last.get("content", "") ``` if role == "assistant" and content: # Strip and display the thinking block separately for readability import re thinking = re.search(r"(.*?)", content, re.DOTALL) if thinking: print(f"[thinking] {thinking.group(1).strip()[:200]}...") clean = re.sub(r".*?", "", content, flags=re.DOTALL).strip() if clean: print(f"[agent] {clean}") elif role == "tool": tool_name = last.get("name", "unknown_tool") print(f"[tool:{tool_name}] result received") if __name__ == "__main__": run_agent( "In the repository myorg/my-api-project, find the open issue about " "the login endpoint returning 200 for invalid tokens. Read the relevant " "code and tests, fix the bug, and open a pull request." ) How to run:** python github_agent_qwenagent.py ## // Part 3: Raw MCP SDK Implementation For teams who need full control over every protocol message, custom error handling, per-tool retry logic, and audit logging of every tool call and result: # github_agent_raw.py # Prerequisites: pip install mcp openai httpx # GITHUB_TOKEN env var must be set, local server must be running # # How to run: # python github_agent_raw.py import asyncio import json import os import re from openai import AsyncOpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # ローカル推論クライアントの設定 ```python client = AsyncOpenAI( base_url="http://localhost:30000/v1", api_key="EMPTY", ) MODEL = "Qwen/Qwen3.6-35B-A3B" ``` ## 応答処理の仕組み `strip_thinking` 関数は、思考ブロック（... ブロック）を除去する役割を持ちます。これは、実際のアクションのみが必要な場合に利用されます。 ```python def strip_thinking(text: str) -> str: """Remove ... blocks. Used when we only need the action.""" return re.sub(r".*?", "", text, flags=re.DOTALL).strip() ``` 一方、`extract_thinking` 関数は、ログ記録のために思考ブロックの内容を抽出します。 ```python def extract_thinking(text: str) -> str: """Extract the content of the thinking block for logging.""" m = re.search(r"(.*?)", text, re.DOTALL) return m.group(1).strip() if m else "" ``` `process_response` 関数は、Qwen3.6 から返されたチャット完了応答を処理します。この関数は、以下の2つの出力形式に対応しています。 1. `--tool-call-parser` オプションが有効な場合、API の `function_call` または `tool_calls` フィールドを通じてツール呼び出しが行われるケース 2. ツール呼び出しがメッセージ本文に JSON 形式で埋め込まれているケース ```python def process_response(response, preserve_thinking: bool = True) -> dict: """ Process a chat completion response from Qwen3.6. Handles two output formats: 1. Tool call via the API's function_call / tool_calls field (when --tool-call-parser is active) 2. Tool call embedded in the message content as JSON Args: response: The OpenAI-compatible completion response preserve_thinking: If True, keep thinking content in output for the next turn's KV cache benefit Returns: dict with thinking, tool_calls, final_answer, has_tool_calls, is_terminal """ choice = response.choices[0] message = choice.message ``` # パス1：構造化フィールドでのツール呼び出し（推奨） このパスでは、`tool-call-parser` フラグが必要です。メッセージに `tool_calls` が存在する場合、以下の処理が行われます。 まず、各ツール呼び出しから関数名、引数（JSON 形式）、および呼び出し ID を抽出し、リスト化します。次に、メッセージの内容から思考プロセスを抽出します。最後に、以下の構造を持つ辞書を返却します。 - `thinking`: 思考プロセス（`preserve_thinking` が有効な場合のみ） - `tool_calls`: 抽出されたツール呼び出しのリスト - `final_answer`: 空文字列 - `has_tool_calls`: True - `is_terminal`: False # パス2：コンテンツテキストに埋め込まれたツール呼び出し（フォールバック） このパスは、構造化フィールドが利用できない場合の代替手段です。メッセージの内容を解析し、特定のタグ形式で囲まれた JSON データを検出します。 検出した各マッチに対して JSON 解析を試みます。解析に成功した場合はツール呼び出しリストに追加され、失敗した場合は無視されます。その後、コンテンツ全体から思考プロセスと最終回答を抽出・整形します。 - `thinking`: 抽出された思考プロセス（`preserve_thinking` が有効な場合のみ） - `tool_calls`: 解析に成功したツール呼び出しのリスト - `final_answer`: ツール呼び出しタグを除いた残りのテキスト（空白除去済み） - `has_tool_calls`: ツール呼び出しが1件以上あるか - `is_terminal`: ツール呼び出しがなく、かつ最終回答が存在する場合のみ True 最後に、これらの情報をまとめた辞書を返却します。 # ── コアエージェントループ ─────────────────────────────────────────────────────────── async def run_github_agent(task: str, repo: str, max_turns: int = 20): """ GitHub 開発支援エージェントを実行します。 ファイルシステムと GitHub の MCP サーバーに接続し、利用可能なツールを検出。Qwen3.6 エージェントループを開始し、タスク完了または最大ターン数（デフォルト 20）に達するまで処理を続けます。 """ # 両方の MCP サーバーを起動し、セッション確立 fs_params = StdioServerParameters( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "."], ) gh_params = StdioServerParameters( command="npx", args=["-y", "@modelcontextprotocol/server-github"], env={**os.environ, "GITHUB_TOKEN": os.environ.get("GITHUB_TOKEN", "")}, ) async with stdio_client(fs_params) as (fs_read, fs_write), \ ClientSession(fs_read, fs_write) as fs_session, \ stdio_client(gh_params) as (gh_read, gh_write), \ ClientSession(gh_read, gh_write) as gh_session: # 両方のセッションを初期化 await fs_session.initialize() await gh_session.initialize() # 両サーバーから利用可能なツールをすべて取得 fs_tools_result = await fs_session.list_tools() gh_tools_result = await gh_session.list_tools() # モデル用の OpenAI 形式ツールリスト構築 all_tools = [] tool_to_session = {} # ツール名を所有する MCP セッションにマッピング fs_tools_result から取得したツールを順に処理し、関数定義として登録します。各ツールの名前、説明、入力スキーマを抽出して `all_tools` リストに追加し、同時にツール名とセッションの対応関係も記録します。 同様に、gh_tools_result からもツールを取得し、同じ形式でリストに登録・マッピングを行います。 その後、利用可能なツールの総数と、それぞれがファイルシステム用か GitHub 用の内訳を表示します。 最後に、会話履歴を構築します。システムプロンプトでは、リポジトリ {repo} にアクセスできるシニアソフトウェアエンジニアとして振る舞うよう指示し、利用可能なツールを使って問題調査、コード閲覧、修正の実施、プルリクエスト作成を行うことを明記しています。思考は段階的に行い、変更前には必ず確認し、最小限の変更に留めるよう定めています。 このプロンプトとタスクをメッセージリストとして初期化します。 # ── エージェントループ ───────────────────────────────────────────────────────── for turn in range(max_turns): print(f"\n[Turn {turn + 1}]") # モデルを呼び出す response = await client.chat.completions.create( model=MODEL, messages=messages, tools=all_tools if all_tools else None, tool_choice="auto", # 公式のベストプラクティスに基づく思考モードのサンプリングパラメータ temperature=0.6, top_p=0.95, top_k=20, min_p=0.0, max_tokens=4096, extra_body={ # preserve_thinking は、長期にわたるエージェントセッションにおいて KV キャッシュの効率化を図りつつ、思考コンテキストをターン間で維持する "preserve_thinking": True, } ) result = process_response(response, preserve_thinking=True) if result["thinking"]: print(f"[thinking] {result['thinking'][:200]}...") # 終端状態 -- エージェントが最終回答を生成した if result["is_terminal"]: print(f"\n[DONE]\n{result['final_answer']}") return result["final_answer"] # ツール呼び出しの状態管理 -- ツールの実行と結果の注入 ツール呼び出しが含まれている場合、各ツールを実行してその結果をシステムに反映させます。 まず、アシスタントからのメッセージ（ツール呼び出しを含む）を履歴に追加します。これには、生成されたテキスト内容と、実際に呼び出されたツールのリストが含まれます。 その後、呼び出されたツール一つひとつに対して処理を行います。各ツールについて、その名前、引数、および一意の呼び出しID を取得します。 実行時には、まず「[tool] ツール名 (引数...)」という形式でログを出力し、次に該当するツールのセッションが存在するか確認します。もし登録されていないツールが呼び出された場合は、「エラー：ツールが見つかりません」というメッセージを生成します。 セッションが存在する場合、非同期処理でツールを実行し、その結果を取得します。ただし、結果が非常に長い場合（12,000 文字を超える場合）は、コンテキストの容量を節約するために先頭 12,000 文字を切り捨て、「...[truncated]」というサロゲートを付加して処理します。 実行中に例外が発生した場合は、エラーメッセージとしてその内容を記録し、システムに返却します。 print(f"[result] {result_content[:150]}...") messages.append({ "role": "tool", "content": result_content, "tool_call_id": call_id, "name": tool_name, }) print(f"[WARNING] max_turns ({max_turns}) reached without terminal state") # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": asyncio.run(run_github_agent( task=( "Find the open issue about the login endpoint returning 200 for invalid tokens. " "Read src/auth.py and tests/test_auth.py to understand the bug. " "Fix the verify_token function and open a pull request with your changes." ), repo="myorg/my-api-project", )) **How to run:** python github_agent_raw.py The raw SDK path gives you what Qwen-Agent abstracts: you can see every tool call, every result, and every message injected into the conversation history. The tool_to_session routing dict is the key mechanism; it maps each tool name to the MCP session that owns it, so the agent can call any tool from any connected server without knowing which server provides it. ## # Writing a Custom MCP Server ** 既存の MCP サーバーはファイルシステムや GitHub へのアクセスを担います。しかし、社内データベースの照会や CI/CD API のラッパー作成、コード解析ツールの実行など、既存のものにない機能が必要になった場合は、独自の MCP サーバーを作成する必要があります。 以下は、ruff と pytest を MCP ツールとして公開する「code_quality」サーバーの完全な実装例です。 # code_quality_server.py # Qwen3.6 向けにコード品質ツールを公開するカスタム MCP サーバー。 # # 事前準備: # pip install mcp ruff pytest # # 単体実行（テスト用）: # python code_quality_server.py # # Qwen-Agent の設定に追加する場合: # "code_quality": { # "command": "python", # "args": ["/absolute/path/to/code_quality_server.py"] # } import asyncio import json import subprocess import sys from mcp.server.fastmcp import FastMCP # FastMCP は高レベルの MCP サーバーフレームワークで、ボイラープレートコードを大幅に削減します。 mcp = FastMCP("code_quality") @mcp.tool() def run_linter(file_path: str, fix: bool = False) -> str: """ Python ファイルに対して ruff リンターを実行し、構造化された結果を返します。 ファイルを修正する前に使用して現在の品質状態を確認し、変更を加えた後は 新たな問題が発生していないか検証するために利用してください。 Args: file_path: リント対象の Python ファイルへの絶対パスまたは相対パス。 fix: true の場合、安全な問題を自動的に修正します。 返却値: 問題一覧、問題数、および変更されたファイルを含む JSON 文字列。 """ cmd = ["python", "-m", "ruff", "check", file_path, "--output-format=json"] if fix: cmd.append("--fix") try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) # ruff は問題が見つかった場合に終了コード 1 を返す（これはエラーではない） output = result.stdout or result.stderr # ruff の JSON 出力を解析 try: issues = json.loads(output) if output.strip() else [] except json.JSONDecodeError: issues = [] formatted = [ { "line": issue.get("location", {}).get("row", 0), "col": issue.get("location", {}).get("column", 0), "code": issue.get("code", ""), "message": issue.get("message", ""), "fix_available": issue.get("fix") is not None, } for issue in issues if isinstance(issue, dict) ] return json.dumps({ "file": file_path, "issues": formatted, "total_issues": len(formatted), "fixed": "auto-fix applied" if fix else "no auto-fix", }, indent=2) except subprocess.TimeoutExpired: return json.dumps({"error": "Linter timed out after 30s", "file": file_path}) except FileNotFoundError: return json.dumps({"error": "ruff not found -- install with: pip install ruff"}) @mcp.tool() def run_tests(target: str, verbose: bool = False) -> str: """ Run pytest on a module or directory and return structured pass/fail results. Use this after writing a fix to verify the fix makes failing tests pass without breaking other tests. Args: target: Path to the test file or directory to run (e.g. tests/, tests/test_auth.py) verbose: If true, include full pytest output in the result. Returns: JSON string with pass count, fail count, failure details, and duration. """ cmd = ["python", "-m", "pytest", target, "--json-report", "--json-report-file=-", "-q"] if verbose: cmd.append("-v") try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) output = result.stdout # Parse pytest-json-report output if available try: report = json.loads(output) summary = report.get("summary", {}) failures = [ { "test": t["nodeid"], "message": t.get("call", {}).get("longrepr", "")[:500], } for t in report.get("tests", []) if t.get("outcome") == "failed" ] return json.dumps({ "target": target, "passed": summary.get("passed", 0), "failed": summary.get("failed", 0), "errors": summary.get("error", 0), "total": summary.get("total", 0), "duration": summary.get("duration", 0), "failures": failures, "stdout": result.stdout[:2000] if verbose else "", }, indent=2) except json.JSONDecodeError: # Fallback: return raw output if JSON report not available return json.dumps({ "target": target, "stdout": result.stdout[:3000], "stderr": result.stderr[:1000], "exit_code": result.returncode, }) except subprocess.TimeoutExpired: return json.dumps({"error": f"Tests timed out after 120s for target: {target}"}) except FileNotFoundError: return json.dumps({"error": "pytest not found -- install with: pip install pytest"}) if __name__ == "__main__": mcp.run(transport="stdio") Add it to either agent implementation's server config: # In Qwen-Agent MCP_SERVERS dict: "code_quality": { "command": "python", "args": ["/absolute/path/to/code_quality_server.py"] } # In the raw SDK, add a third StdioServerParameters: cq_params = StdioServerParameters( command="python", args=["/absolute/path/to/code_quality_server.py"], ) Test the server standalone before connecting the agent: # Test the server in MCP inspector mode npx @modelcontextprotocol/inspector python code_quality_server.py # Opens a browser UI where you can call run_linter and run_tests directly ## # Tuning Thinking Mode and Preserving Reasoning The thinking mode decision affects latency significantly enough that it is worth treating as an explicit architecture choice, not an afterthought. In thinking mode, Qwen3.6 generates a chain-of-thought reasoning trace inside ... tags before producing its action. For a 5-step agent task, that trace adds 1,000 to 5,000 tokens per turn depending on task complexity. Those tokens take time to generate and consume context budget. When that cost is worth paying: - Planning steps where the agent decides what to do next. 問題の本質が曖昧なデバッグセッションや、ファイル間での副作用を推論しながら複数ファイルをリファクタリングするケースでは、思考トレースがツール呼び出しの誤りを事前に検出できます。 一方、ディレクトリのリストアップ→ファイルの読み込み→ファイルへの書き込み→コミットといった機械的なツール呼び出しループのように、各ステップが明確で推論コストに見合わない場合は、無理に思考させる必要はありません。この「非思考モード」の方が高速であり、かつ同等の品質を維持できます。 これらのモードはグローバル設定ではなく、リクエストごとに切り替えるべきです。 ### 思考モード（計画策定、デバッグ、複雑な複数ファイルタスク向け） THINKING_PARAMS = { "temperature": 0.6, "top_p": 0.95, "top_k": 20, "min_p": 0.0, } ### 非思考モード（機械的なループ、高速なステータス確認向け） チャットテンプレートで enable_thinking=False を指定するか、システムプロンプトに「/no_think」を追加して思考モードを抑制します。 NON_THINKING_PARAMS = { "temperature": 0.7, "top_p": 0.8, "top_k": 20, "min_p": 0.0, } 「preserve_thinking」フラグは、Qwen3.6 固有の機能で、会話のやり取りを通じて推論コンテキストを保持する能力を持っています。これがプリフィックスキャッシュが有効な場合に推論効率に直接影響を与えます。実際的な重要性としては、10 ターンにわたるエージェントセッションでは、各ターンで会話履歴の共通部分（プリフィックス）が共有されます。「preserve_thinking=True」を設定すると、過去のターンからの完全な推論トレースが履歴に残り続けます。サーバー側の KV キャッシュはこの共通部分を認識し、再計算を回避します。その結果、長期間にわたるセッションにおける有効なトークン処理速度は、この機能を無効にした場合よりも著しく向上します。特に SGLang などのインフラストラクチャで「--enable-prefix-caching」オプションが有効になっている環境では、その効果が顕著です。 実用的なルールとして、5 ターン以上続くエージェントセッションでは「preserve_thinking=True」を使用してください。一方、単発のクエリや短いパイプライン処理では、オーバーヘッドが無駄になるため、「preserve_thinking=False」（または思考モードをオフにした状態）に設定するのが適切です。 ## # 結論 Qwen3.6-35B-A3B の MoE（Mixture of Experts）アーキテクチャにより、3B の活性化コストで 35B モデル並みの性能を実現できます。また、262k という広大なコンテキストウィンドウを備えているため、コードレビューセッション全体をコンテキスト内に保持することも可能です。さらに、MCP ベースのエージェントベンチマークに対する明示的なトレーニングにより、単にツールを呼び出すだけでなく、正しく活用する方法も理解しています。 MCP は、システムをつなぐ接着剤のような役割を果たします。ツールは MCP サーバーとして一度定義すれば、Qwen3.6 のすべてのセッションや他の MCP 対応モデルが、特別な接続コードなしでそのツールを検索して呼び出すことができます。本記事で紹介している GitHub サーバーとファイルシステムサーバーは、[MCP エコシステム](https://modelcontextprotocol.io/examples) に存在する数百の事前構築済みサーバーのうちの一つに過ぎません。一方、カスタムで作成した code_quality サーバーは、既存のものがない場合でも同様のパターンを適用できることを示しています。 本記事で取り上げた GitHub 開発者アシスタントは、このパターンの一つの応用例です。ローカルモデル、MCP ツール、そしてエージェントによるループという同じアーキテクチャは、学術データベースを検索して文献レビューの草案を作成する研究アシスタントや、CloudWatch ログを読み込んでインシデントチケットを起票する DevOps エージェント、SQL スキーマを読み込み変換コードを書き出力を検証するデータパイプラインエージェントなど、さまざまな用途で活用可能です。MCP エコシステムは急速に成長しており、ローカルモデルの能力もすでに確立されています。 [Shittu Olumide](https://www.linkedin.com/in/olumide-shittu/) はソフトウェアエンジニアであり技術ライターです。最先端技術を駆使して説得力のある物語を紡ぐことに情熱を注ぎ、細部への鋭い眼と複雑な概念を簡潔に説明する才能を持っています。また、[Twitter](https://twitter.com/Shittu_Olumide_) でも活動しています。

ローカル AI システムの構築：Qwen3.6 と MCP の活用

背景や根拠まで確認しますか？

関連記事

調べる

選ぶ

サイト