MCPサーバーを作る: 要件・実装・失敗・再現手順

この記事には広告・アフィリエイトリンクも、広告・提携関係もありません。

MCPサーバーは、関数を数個登録するだけなら短いコードで動きます。しかし、既存データをAI Agentから参照させるには、ツールの数より先に境界を決める必要があります。

私は検索結果のAI Overview出現状況を記録するaio-probeに、ローカルSQLiteを照会するMCP companionを追加しました。新しい計測や有料API呼び出しは行いません。read-onlyを多層で実装した理由、集計ルール、テスト後も残った問題まで記録します。これはローカルでの小規模実装であり、クラウド本番運用の実証ではありません。

最初に決めた要件

実装前に、次の要件へ絞りました。

SQLiteを読むだけにし、runやobservationを変更しない
外部API、ブラウザー、ファイル書き込みをtoolから実行しない
toolは3つに絞る
failedをAIO率の分母から除外する
計測開始時のsnapshotを過去runの正本にする
DB由来の文字列、処理時間、返却量を制限する
adapterを計測本体から分離する

注釈だけでは1番目の要件を満たせません。注釈、SQLite接続、クエリ、返却値、テストの各層に境界を置きました。

構造: stdioの先にread-only Storeを置く

構成は次の一方向です。

MCP host
  └─ stdio ─> MCPServer ─> ReadOnlyStore ─> existing SQLite
                         （外部API呼び出しなし）

MCP側はtool schemaを担当し、SQL、集計、入力検証、sanitizeはReadOnlyStoreへ集約しました。公開範囲と負荷を固定するため、任意SQLを受けるtoolは作っていません。

uv.lockで固定している公式のMCP Python SDK v2.0.0でMCPServerを使い、server.run()を引数なしで起動します。公式ドキュメントでdefaultとされるstdio transportなので、HTTP endpointは持ちません。

3つのtoolと、それぞれの判断

ツール	返すもの	主な用途
`list_runs`	run条件、対象数、観測数、成功・失敗・AIO件数	どの計測を比較するか選ぶ
`compare_buckets`	4 bucketのAIO率、失敗率、除外件数	クエリ形状ごとの差を見る
`get_keyword_evidence`	1 keywordの観測、citation、organic結果	集計値の根拠へ戻る

run_id省略時は「完了済みでobservationが1件以上ある最新run」を選びます。過去runのkeywordとbucketは計測開始時のsnapshotを優先し、現在の設定を混ぜません。snapshot導入前のlegacy runだけ現在値へfallbackします。

実装1: 注釈は意思表示、制御は別に置く

3つのtoolには同じToolAnnotationsを付けました。

annotations=ToolAnnotations(
    readOnlyHint=True,
    destructiveHint=False,
    idempotentHint=True,
    openWorldHint=False,
)

このツール群はDBを読むだけで、外部serviceへ接続しません。ただし、MCPのtool仕様と公式のTool Annotations解説が示す通り、annotationは契約ではなくhintです。readOnlyHint=Trueは安全性を保証しないため、書き込み拒否はSQLite層で行います。

実装2: SQLiteを二重にread-onlyへする

接続時はSQLite URIへmode=roを付け、さらにconnection単位でPRAGMA query_only = ONを設定しました。概念上の中心は次の部分です。

uri = f"{db_path.as_uri()}?mode=ro"
conn = sqlite3.connect(uri, uri=True, timeout=5.0)
conn.execute("PRAGMA query_only = ON")

SQLite URIの仕様ではmode=roはread-onlyで開く指定です。query_onlyの仕様では、CREATE、DELETE、DROP、INSERT、UPDATEがSQLITE_READONLYになります。ただしquery_only単独は完全なread-onlyではないため、mode=roと重ねました。

起動時には要求スキーマを確認し、必要tableがviewやvirtual tableなら拒否します。DB本体とWALは合計512 MiB、クエリは5秒、同時4件、SQLite cellは1 MiB、list_runsは100 run、citationとorganic結果はそれぞれ100件、keyword入力は512文字を上限にしました。

実装3: 集計失敗を「AIOなし」にしない

最も重要な集計ルールは、status='failed'をAIOなしへ数えないことです。

aio_rate     = aio_count / ok
failure_rate = failed / total

取得やparseに失敗した観測は「AIOなし」ではなく「判定不能」です。分母はokだけにし、failedはexcluded_failedと失敗率で返します。成功0件ならaio_rateはnullです。

実装4: DBの内容もuntrustedとして返す

read-onlyでもDB内の文字列は信頼できません。get_keyword_evidenceはdata_trust: "untrusted_external"を付け、次を処理します。

保存済みの内部error本文とraw保存pathを返さない
failed時は固定文言へ置き換える
URLはhttpまたはhttpsだけにし、userinfo、query、fragmentを落とす
制御文字などを除去し、domain、URL、metadataの長さを制限する
citationとorganic結果に返却上限を設け、totalとreturnedを分ける
sanitizeや省略が起きた場合はredacted、truncatedで示す

prompt injectionを無効化する保証はなく、hostとmodel側でも外部データとして扱う必要があります。

検証: 「書かなかった」ことも確認する

2026-08-26に現行テストスイートを再実行し、MCP対象テストは32 passed、全体回帰も成功しました。ruff check src testsとruff format --check src testsも通しています。

テストでは次を確認しました。

mode=roとquery_onlyで書き込みが拒否され、DB bytesとschemaが変わらない
snapshotのbucketが現在値より優先される
failedがAIO率の分母から除外される
最新run resolverが未完了runや観測0件runを選ばない
保存済みerrorや危険なURLを隠し、不正schemaと上限外入力を拒否する

実DB smokeでも呼び出し前後のSHA-256、size、mtimeが一致し、新しいsidecar fileはありませんでした。これは当該条件で変更しなかった証拠であり、あらゆる環境での不変を保証しません。

レビューで見つかった失敗

最初に動いたことより、レビューで境界の抜けが見つかったことの方が設計材料になりました。

「最新run」の定義がtool間で揃っていなかった
初期のget_keyword_evidenceは全runから取得時刻の新しい観測を選べたため、未完了runを拾う余地がありました。compare_bucketsと同じ「観測を持つ最新の完了run」resolverへ統一しました。
保存済みerrorを読んでから伏字にする設計では不十分だった
pathの形式はUNCや空白を含むものまであり、正規表現だけでは漏えい余地が残りました。error列自体をSELECTせず、failed時は固定文言を返す形へ変えました。
正常な小規模DBだけを前提にしていた
当初は全履歴集計や応答資源の上限が不足していました。対象runを先に最大100件へ絞り、DBサイズ、クエリ時間、同時実行数、cell長、返却件数の上限を追加しました。

いずれも「read-onlyだから安全」という一語では防げない問題でした。選択条件、取得する列、処理資源まで含めて境界を定義する必要があります。

まだ残っている限界

レビュー後も低重要度の問題が2件残っています。

改変DBでrun metadataがNULLの場合、str()によって文字列"None"になる箇所があります。NULLの拒否かJSON nullへの伝播が必要です。
観測0件のbucketではaio_rateがnullなのに、failure_rateは0.0です。「0%」と「データなし」を分ける余地があります。

また、認証・認可を備えたremote serviceではありません。ローカルstdio前提で、hostと実行ユーザーが読めるDBを同じ権限で参照します。秘密情報や顧客データ、クラウド配置、multi-tenant分離、監査、障害復旧は対象外です。

再現手順

前提はPython 3.12以降、uv、互換schemaのaio-probe SQLiteです。公開可能な検証用DBを使い、ほかのprocessが更新しない状態で確認します。

cd <repo-path>
uv sync
uv run pytest -q tests/test_mcp_store.py tests/test_mcp_server.py
uv run pytest -q
uv run ruff check src tests
uv run ruff format --check src tests

MCP hostへ次のcommandとargsを登録します。設定keyはhostの仕様に合わせます。

{
  "command": "uv",
  "args": [
    "run",
    "--project",
    "<repo-path>",
    "aio-probe-mcp",
    "--db",
    "<absolute-db-path>"
  ]
}

直接起動した場合、stdio serverはhostからの入力を待ちます。

uv run --project <repo-path> aio-probe-mcp --db <absolute-db-path>

接続後はlist_runs、compare_buckets、必要なkeywordだけget_keyword_evidenceの順に使います。

DB不変性は、起動前後に同じ停止状態でSHA-256、size、mtimeを比較します。

Get-FileHash -Algorithm SHA256 -LiteralPath "<absolute-db-path>"
(Get-Item -LiteralPath "<absolute-db-path>").Length
(Get-Item -LiteralPath "<absolute-db-path>").LastWriteTimeUtc

この実装をどう学習へつなげるか

identity、secret、data boundary、observability、costを実装上の問いへ変える流れはB-01で整理しています。資格が証明する範囲はC-01、公開記事の読む順番はロードマップで確認できます。

初版にはaffiliate linkを置きません。公式資料、ローカル実装、失敗条件、再現手順で判断できる形にします。公式仕様と実装事実の最終確認日は2026-08-26です。

最初に行うことは、公開可能な検証用DBでMCP対象テストを実行し、書き込み拒否と実行前後のhash不変を確認することです。そこで差分が出た場合は、MCP hostへ登録する前に止めます。