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を追加します。

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処理を二重実装しません。

例えば、

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へ接続します。

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段階に分けることで、どこで失敗したかを判断しやすくします。

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へ渡します。

処理の流れは次です。

ここでは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なども、処理結果を共通の仕組みで管理しやすくなります。

参考資料

FastMCP Tools
https://gofastmcp.com/servers/tools
FastMCP Resources & Templates
https://gofastmcp.com/servers/resources
FastMCP Server
https://gofastmcp.com/servers/server
FastMCP Running Your Server
https://gofastmcp.com/deployment/running-server
FastMCP CLI
https://gofastmcp.com/cli/overview
FastMCP PyPI
https://pypi.org/project/fastmcp/