0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

FastMCP 3 + Docker でGIS MCP Serverを作る ― Step 2 - Dataset CatalogをMCP Tool / Resourceとして公開する

0
Posted at

FastMCPでGIS MCP Serverを作る Step 2 - Dataset CatalogをMCP Tool / Resourceとして公開する

はじめに

前回のStep 1では、GeoPackage / GeoJSON / GeoParquetを登録前に確認し、問題のないデータをDataset Catalogへ登録できるところまで作りました。

登録したDatasetは、物理ファイルのpathではなく、

dataset_id
version

で扱います。

また、Docker Composeから次の4コマンドを実行できるようにしました。

catalog register
catalog list
catalog info
catalog versions

Step 2では、このDataset CatalogをFastMCPから参照できるようにします。

今回のポイントは、MCP対応のためにGISファイルの読込み処理やDuckDBへの登録処理を作り直さないことです。

Step 1で作ったCatalogServiceをそのまま使い、その外側にMCP Layerを追加します。

20260911_g11.png

FastMCPはGIS処理本体ではなく、MCP ClientとGISアプリケーションをつなぐ入口として使います。


1. Step 1から引き継ぐもの

Step 2は新しいプロジェクトを作り直すのではなく、Step 1の実装をそのまま引き継ぎます。

特に次の部分は変更しません。

  • Dataset / DatasetVersion
  • control.duckdb
  • SQL migration
  • GeoPackage / GeoJSON / GeoParquetの確認
  • CRS / Geometry / feature count / bboxの取得
  • SHA-256
  • dataset_id / version
  • CatalogService
  • RegistrationService
  • ControlRepository
  • VectorInspector
  • catalog register
  • catalog list
  • catalog info
  • catalog versions

Step 1の最後では、次の状態まで確認しています。

GISファイル
    ↓
登録前の確認
    ↓
Dataset Catalog
    ↓
control.duckdb
    ↓
dataset_id / version

Step 2では、この右側にFastMCPを追加します。

MCP Client
    ↓
FastMCP
    ↓
Dataset Catalog
    ↓
control.duckdb

Dataset登録は引き続きcatalog registerを使います。

MCP ToolからDataset登録まで行う機能は、まだ追加しません。


2. Step 2で追加するもの

Step 2で追加する範囲は次です。

  • FastMCP Server
  • Streamable HTTP /mcp
  • Dataset一覧Tool
  • Dataset詳細Tool
  • Dataset version履歴Tool
  • runtime status Tool
  • Dataset Catalog Resource
  • Dataset Resource Template
  • FastMCP in-memory test
  • in-process HTTP transport test
  • Docker Composeのmcp service
  • Docker Composeのsmoke service

公開するToolはすべて読み取り専用です。

gis_list_datasets
gis_describe_dataset
gis_list_versions
gis_runtime_status

Resourceは次の2つから始めます。

gis://catalog
gis://dataset/{dataset_id}

参考のために、ZIPファイルを置いておきますので、ダウンロードして活用してください。


3. FastMCPは3.4.7を使う

この記事ではFastMCP 3.4.7を固定して使います。

2026年8月26日時点では、FastMCP 3.4.7がstable releaseです。4系は4.0.0b3まで公開されていますがpre-releaseなので、Step 2では3系のままMCP公開部分を固めます。

主なversionは次です。

ライブラリ version 用途
Python 3.13.15 実行環境
FastMCP 3.4.7 MCP Server
DuckDB 1.5.5 Dataset Catalog
GeoPandas 1.1.4 Vector GIS
Pyogrio 0.13.0 Vector I/O

FastMCPのHTTP transportにはStreamable HTTPを使います。

標準endpointは、

http://localhost:8000/mcp

です。


4. Step 2の構成

FastMCP固有の処理はsrc/gis_mcp/mcp/へまとめます。

gis_mcp_server_step2/
├─ compose.yaml
├─ Dockerfile
├─ pyproject.toml
├─ .env.example
│
├─ migrations/
│  └─ 001_dataset_catalog.sql
│
├─ scripts/
│  ├─ catalog_cli.py
│  ├─ probe_mcp_http.py
│  ├─ check_runtime.py
│  ├─ check_step_completion.py
│  ├─ init_workspace.py
│  └─ validate_workspace.py
│
├─ src/gis_mcp/
│  ├─ config.py
│  ├─ contracts.py
│  │
│  ├─ domain/
│  ├─ ports/
│  ├─ services/
│  ├─ adapters/
│  │
│  └─ mcp/
│     ├─ __init__.py
│     ├─ __main__.py
│     ├─ bootstrap.py
│     ├─ http_runtime.py
│     ├─ models.py
│     └─ server.py
│
└─ tests/
   ├─ unit/
   ├─ architecture/
   ├─ integration/
   └─ mcp/

役割は次のように分けます。

server.py
  Tool / Resourceの公開

models.py
  MCP Clientへ返すmodel

bootstrap.py
  Repository / Serviceの組み立て

http_runtime.py
  FastMCP run_async / graceful shutdown

__main__.py
  async HTTP Serverの起動

MCP Layerの中へSQLやGeoPandas処理を書かないことが重要です。


5. MCP用modelをDomain modelと分ける

内部で使っているDatasetVersionRecordを、そのままMCP Clientへ返す方法もあります。

ただし、その形にすると内部実装を変更しただけで、MCP Client側のschemaまで変わってしまいます。

そこでMCP用modelを分けます。

McpDatasetSummary
McpDatasetDetail
McpRuntimeStatus

例えばDatasetの概要では、

dataset_id
name
version
status
driver
layer_name
crs
geometry_type
feature_count
created_at

などを返します。

一方で、

/workspace/datasets/...
/workspace/control/control.duckdb
storage_path

のようなcontainer内部の情報は返しません。

MCP Clientが知る必要があるのはDatasetの意味やCRS、Geometry、件数であり、内部の保存pathではないためです。


6. FastMCP Serverを作る

Server側では、入力値を曖昧に変換しないようstrict_input_validation=Trueを使います。

また、想定外の内部例外をそのままMCP Clientへ返さないようmask_error_details=Trueも有効にします。

from fastmcp import FastMCP

mcp = FastMCP(
    "GIS MCP Server",
    strict_input_validation=True,
    mask_error_details=True,
)

strict_input_validation=Trueにしておくと、例えば整数を要求する引数へ不正な型が渡された場合に、曖昧な型変換を行わずschemaに従って検証できます。


7. 公開するTool

7.1 gis_list_datasets

登録済みDatasetの最新versionを一覧表示します。

gis_list_datasets

内部ではStep 1のCatalogServiceを呼びます。

MCP Client
    ↓
gis_list_datasets
    ↓
CatalogService.list_datasets()
    ↓
ControlRepository
    ↓
control.duckdb

CLIの、

docker compose --profile demo run --rm catalog list

と同じDataset Catalogを参照します。

7.2 gis_describe_dataset

指定したDatasetの詳細を返します。

gis_describe_dataset

入力は、

{
  "dataset_id": "landslide_area"
}

のようにします。

特定versionを参照したい場合は、

{
  "dataset_id": "landslide_area",
  "version": 1
}

とします。

Python側の形は次のようになります。

@mcp.tool(
    name="gis_describe_dataset",
    annotations={
        "readOnlyHint": True,
        "openWorldHint": False,
    },
)
def describe_dataset(
    dataset_id: str,
    version: int | None = None,
) -> McpDatasetDetail:
    return _detail(catalog.get(dataset_id, version))

Step 2のToolはCatalogを読むだけなので、readOnlyHint=Trueを付けます。

7.3 gis_list_versions

Datasetのversion履歴を取得します。

gis_list_versions

入力例です。

{
  "dataset_id": "landslide_area"
}

Step 1の、

docker compose --profile demo run --rm catalog versions \
  --dataset-id landslide_area

と同じ履歴をMCPから取得できるようにします。

7.4 gis_runtime_status

MCP Serverの基本状態を確認するToolです。

gis_runtime_status

ここでは、

service
implementation_step
workspace_ready
control_db_ready
schema_version

など、MCP ServerがDataset Catalogへ接続できる状態かを返します。

GIS Datasetそのものとは別にruntime確認用Toolを持たせておくと、接続障害の切り分けがしやすくなります。


8. ToolとCLIで別の処理を作らない

Step 1ではCLIからDataset Catalogを操作できるようにしました。

Step 2ではMCP Toolを追加しますが、CLI用とMCP用でCatalog処理を二重実装しません。

20260911_g12.png

例えば、

catalog list

と、

gis_list_datasets

は入口が違うだけです。

内部では同じCatalogServiceを使います。

この形にしておけば、CLIでは見えるのにMCPでは見えない、といった実装差を減らせます。


9. Resourceも公開する

ToolだけでなくResourceも追加します。

gis://catalog
gis://dataset/{dataset_id}

gis://catalog

Dataset Catalog全体の概要を読み取ります。

@mcp.resource(
    "gis://catalog",
    mime_type="application/json",
)
def catalog_resource():
    return ...

gis://dataset/{dataset_id}

指定したDatasetのmetadataを読み取るResource Templateです。

@mcp.resource(
    "gis://dataset/{dataset_id}",
    mime_type="application/json",
)
def dataset_resource(dataset_id: str):
    return ...

ToolとResourceは用途を分けますが、どちらも同じCatalogServiceへ接続します。

20260911_g13.png


10. Streamable HTTPで公開する

FastMCP ServerはStreamable HTTPで起動します。

mcp.run(
    transport="http",
    host="0.0.0.0",
    port=8000,
)

Docker Composeではmcp serviceを追加します。

services:
  mcp:
    profiles: ["mcp", "smoke"]
    command: ["python", "-m", "gis_mcp.mcp"]
    ports:
      - "8000:8000"

接続先は、

http://localhost:8000/mcp

です。

Step 2ではローカル開発環境での接続確認に範囲を絞り、認証・認可はまだ追加しません。


11. smoke serviceを追加する

MCP Serverと同じcontainer内から確認するだけでは、Docker network経由の接続確認になりません。

そこで別containerとしてsmoke serviceを用意します。

smoke container
      ↓
http://mcp:8000/mcp
      ↓
mcp container
      ↓
FastMCP

smokeからは、少なくとも次を確認します。

MCP Serverへ接続できる
Tool一覧を取得できる
gis_runtime_statusを呼べる

成功した場合は、

MCP_HTTP_SMOKE=PASS

を表示します。


12. in-memory testを先に通す

MCP Serverの確認を最初からHTTPだけで行うと、失敗したときに原因を分けにくくなります。

そこで最初にFastMCPのin-memory Clientを使います。

from fastmcp import Client

async with Client(server) as client:
    tools = await client.list_tools()
    result = await client.call_tool(
        "gis_list_datasets",
        {},
    )

このテストではDocker networkを使いません。

それでも実際のMCP protocol処理を通るため、

Tool名
入力schema
structured result
Resource
Resource Template

を確認できます。


13. HTTP transportも別に確認する

in-memory testが通っても、HTTP transportの設定が正しいとは限りません。

そのため次にin-process HTTP transport testを行います。

MCP Client
    ↓
Streamable HTTP
    ↓
FastMCP Server
    ↓
Tool / Resource

ここでは、起動だけでなく終了処理まで確認します。

FastMCPのHTTP Serverはasyncで動くため、本番entrypointではrun_async()を使います。

async def _main_async() -> None:
    server = build_server()
    await server.run_async(
        transport="http",
        host=host,
        port=port,
        path="/mcp",
    )


def main() -> None:
    asyncio.run(_main_async())

FastMCPのrun()は同期用のwrapperなので、async functionの中ではrun_async()を使います。

in-process testの終了方法

FastMCP 3.4.7のtest helper run_server_async()は、context終了時にHTTP Server taskを
cancel()します。環境によっては、この終了時にUvicorn / Starlette lifespan側から
asyncio.CancelledErrorがERRORログとして出ることがあります。

Step 2ではERRORをlogging filterで隠すのではなく、test側でUvicorn Serverを明示的に
所有します。

server.http_app()
    ↓
uvicorn.Server
    ↓
MCP ClientでHTTP確認
    ↓
server.should_exit = True
    ↓
await server.serve() 完了
    ↓
Starlette / MCP lifespan shutdown完了

実装では、

async with temporary_http_server(server) as url, Client(url) as client:
    ...

という形にしています。temporary_http_server()server.http_app()をUvicornへ渡し、
終了時はtaskを通常経路でcancelせず、should_exit=Trueを設定してserve()が戻るまで待ちます。

これで、

MCP_IN_PROCESS_HTTP=PASS

の直後に不要なCancelledError tracebackを残さず、shutdownまで含めて確認できます。

さらに最後にDocker Compose上で、

smoke container
    ↓
Docker network
    ↓
mcp container
    ↓
/mcp

まで確認します。

テストを3段階に分けることで、どこで失敗したかを判断しやすくします。

20260911_g14.png


14. Step 1のCatalogへDatasetを登録する

ここから実際に動かします。

Step 2でもDataset登録方法はStep 1と同じです。

まず登録するGISファイルを配置します。

mkdir -p data/incoming
cp landslide_area.gpkg data/incoming/

次にCatalogへ登録します。

docker compose --profile demo run --rm catalog register \
  --file /workspace/incoming/landslide_area.gpkg \
  --dataset-id landslide_area \
  --name "土砂災害警戒区域" \
  --layer landslide_area

登録状態を確認します。

docker compose --profile demo run --rm catalog list

さらに詳細も確認できます。

docker compose --profile demo run --rm catalog info \
  --dataset-id landslide_area

version履歴は次です。

docker compose --profile demo run --rm catalog versions \
  --dataset-id landslide_area

ここまではStep 1と同じです。


15. Step 2のDocker imageとテストを実行する

最初に.envを作成します。

cp .env.example .env

uv.lockを生成します。

docker compose --profile maintenance run --rm lock

生成後は.envの、

UV_FROZEN=1

を使います。

Compose定義を確認します。

docker compose config

test imageを作成します。

docker compose build --no-cache test

Step 1の回帰テストを含めたStep 2の全テストを実行します。

docker compose --profile test run --rm test

最後に、

STEP2_COMPLETION=PASS

が表示されることを確認します。


16. MCP Serverを起動する

テストが通ったらMCP Serverを起動します。

container内のpython -m gis_mcp.mcpは、内部でasync entrypointからFastMCPの
run_async()を呼びます。

docker compose --profile mcp up -d mcp

状態を確認します。

docker compose --profile mcp ps

MCP endpointは、

http://localhost:8000/mcp

です。

FastMCP CLIからローカルServerのTool一覧を確認する場合は、認証なしを明示して次のように実行できます。

fastmcp list http://localhost:8000/mcp --auth none

一覧には少なくとも次のToolが含まれます。

gis_list_datasets
gis_describe_dataset
gis_list_versions
gis_runtime_status

17. MCPからDataset Catalogを確認する

Step 1では、

docker compose --profile demo run --rm catalog list

で確認していました。

Step 2では同じCatalogをMCP経由でも確認できます。

例えば、

gis_list_datasets

を呼ぶと、登録済みDatasetの最新versionが返ります。

landslide_areaの詳細を確認する場合は、

{
  "dataset_id": "landslide_area"
}

gis_describe_datasetへ渡します。

処理の流れは次です。

20260911_g15.png

ここではGISファイルを直接読み直しているわけではありません。

Step 1で確認・登録したmetadataをDataset Catalogから取得しています。


18. Docker container間のHTTP接続を確認する

最後にsmoke serviceから接続します。

docker compose --profile smoke run --rm smoke

成功すれば、

MCP_HTTP_SMOKE=PASS

となります。

Step 2では、

STEP2_COMPLETION=PASS
MCP_HTTP_SMOKE=PASS

の両方を確認して完了とします。


19. Step 1のテストを残す理由

Step 2でMCP対応を追加しても、Step 1のテストは削除しません。

確認範囲は積み上げます。

Step 0
Docker / Runtime / Ruff / mypy
        ↓
Step 1
Dataset Catalog / DuckDB / Vector GIS
        ↓
Step 2
FastMCP / Tool / Resource / HTTP

MCP Serverが起動しても、Dataset Catalogが壊れていたら意味がありません。

そのためStep 2の完了判定でも、

catalog register
catalog list
catalog info
catalog versions

が維持されていることを確認します。


20. Step 2の完了条件

Step 3へ進む前に次を確認します。

Docker / Runtime

  • docker compose configが成功する
  • docker compose build --no-cache testが成功する
  • runtime version確認が成功する
  • workspace書込み確認が成功する

Step 1回帰確認

  • Dataset CatalogのUnit / Integration Testが成功する
  • GeoPackage / GeoJSON / GeoParquet確認が成功する
  • DuckDB migration / transaction testが成功する
  • catalog register / list / info / versionsが維持されている

FastMCP

  • gis_list_datasetsが公開される
  • gis_describe_datasetが公開される
  • gis_list_versionsが公開される
  • gis_runtime_statusが公開される
  • structured resultが期待schemaで返る
  • gis://catalogが公開される
  • gis://dataset/{dataset_id}が公開される
  • in-memory MCP testが成功する
  • in-process Streamable HTTP testが成功する

Docker HTTP

  • mcp serviceが起動する
  • http://localhost:8000/mcpで待受する
  • smoke serviceから接続できる
  • MCP_HTTP_SMOKE=PASSが表示される

最終的には、

STEP2_COMPLETION=PASS
MCP_HTTP_SMOKE=PASS

を確認します。


21. 今回のポイント

Step 2で追加したのはMCP Layerですが、Dataset Catalogそのものは作り直していません。

構成は次のままです。

MCP Client
    ↓
FastMCP
    ↓
Tool / Resource
    ↓
CatalogService
    ↓
ControlRepository
    ↓
control.duckdb

CLIも同じServiceを使います。

catalog CLI ─────┐
                 ├─→ CatalogService → ControlRepository
FastMCP Tool ────┤
FastMCP Resource ┘

この境界を保っておくと、MCP Toolが増えてもGIS処理やDatabase処理がToolの中へ散らばりにくくなります。

Step 2では、まずDataset Catalogを安全に「読む」ところまでにしました。

GIS解析Toolは、CatalogとMCPの接続を確認してから追加していきます。


22. 次のStep

次のStepでは、MCPからGIS処理を実行した結果を単なる一時ファイルではなく、追跡できる形で管理できるようにします。

中心になるのは、

DatasetVersion
    ↓
Operation
    ↓
Result
    ├─ Artifact
    └─ Provenance

という構成です。

どのDataset versionを使い、どの処理条件で結果を作ったのかをresult_idから追えるようにします。

これができると、その後のBuffer、Overlay、Clip、Spatial Joinなども、処理結果を共通の仕組みで管理しやすくなります。


参考資料

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?