モジュールの routers に APIRouter を並べる。CompositeModule に並べた全モジュールの routers が 1 つにまとまる。 routers を書かないモジュール (購読だけのものなど) は何も載せない。

from fastapi import APIRouter


class 〇〇(AppModule):
    @property
    @override
    def routers(self) -> list[APIRouter]:
        return [UserResource().router]

api.py:

from fastapi import FastAPI
from fastship.common.port.adapter.resource import add_exception_handlers

from core import Backend  # 全モジュールを束ねた CompositeModule

backend = Backend()
app = FastAPI(lifespan=lifespan)  # lifespan で backend.startup() / backend.shutdown() を呼ぶ
add_exception_handlers(app)  # エラー応答を RFC 9457 にそろえる (「⚠️ エラー応答」)

for router in backend.routers:
    app.include_router(router)

• lifespan の中ではなく、app を作った直後に載せる。lifespan の外で呼ぶ app.openapi() にも載り、テストごとに lifespan を回してもルートが重複しない
• routers は起動 (DI 登録) の前に呼ばれる。Resource のコンストラクタでは DI を解決せず、リクエストを処理するときに解決する

MCP サーバーは FastMCP で立てる (fastmcp 4.1.3 で確かめた)。FastMCP は extras に分けている。 fastmcp を付けて入れ、fastmcp 自体も直接の依存として宣言する (宣言しないと deptry が推移的な依存 (DEP003) として検出する)。

uv add 'fastship-py[fastmcp]' fastmcp

モジュールの tools に FastMCP の Tool を並べる。CompositeModule に並べた全モジュールの tools が 1 つにまとまる。 tools を書かないモジュールは何も載せない。

CompositeModule.tools は全ツールの関数を包み (fastship.common.port.adapter.tool.translate_errors)、例外を API (add_exception_handlers) と同じ方針で MCP のエラーに変える。 モジュールは何も付けなくてよい。

from fastmcp.tools import Tool


class UserTool:
    def register_user(self, email_address: str) -> UserOutput: ...


class 〇〇(AppModule):
    @property
    @override
    def tools(self) -> list[Tool]:
        tool = UserTool()
        return [Tool.from_function(tool.register_user, name="register_user", description="ユーザーを登録する")]

例外の中身はクライアントに返さず、FastMCP の mask_error_details の設定にも頼らない。元の例外も切り離し、OTel のスパンなどにメッセージを残さない。

引数の検証 (不足・余分・型の誤り) は、本体の前に FastMCP が行い、どの引数が誤りかをモデルに返す (WARNING で記録する)。

mcp_server.py:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastmcp import FastMCP
from fastmcp.server.auth import RemoteAuthProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastship.common.port.adapter.tool import PublishNotificationsOnToolCall
from pydantic import AnyHttpUrl

from core import Backend  # 全モジュールを束ねた CompositeModule

backend = Backend()

host = os.environ["MCP_HOST"]  # この MCP サーバーを公開するホスト名 (例: mcp.example.com)
issuer = os.environ["MCP_AUTH_ISSUER"]  # IdP の issuer (例: https://example.us.auth0.com/)
# IdP が発行したアクセストークン (JWT) を検証する。トークンが無い・正しくないリクエストは 401 で断る。
# 認可サーバー (IdP) の場所を /.well-known/oauth-protected-resource で知らせ、MCP のクライアントにトークンを取らせる
auth = RemoteAuthProvider(
    token_verifier=JWTVerifier(
        jwks_uri=os.environ["MCP_AUTH_JWKS_URI"],
        issuer=issuer,
        audience=os.environ["MCP_AUTH_AUDIENCE"],
    ),
    authorization_servers=[AnyHttpUrl(issuer)],
    base_url=f"https://{host}",
)
# ToolError 以外の例外の中身 (SQL・入力値など) を、クライアントに返さない
# ツールの呼び出しが成功したら、ツールの中で起きたドメインイベントを応答の前に MQ へ発行する
mcp = FastMCP(
    "app", tools=backend.tools, auth=auth, middleware=[PublishNotificationsOnToolCall()], mask_error_details=True
)
# 公開するホスト名と localhost 以外の Host、信頼しない Origin を断る (DNS rebinding を防ぐ)
mcp_app = mcp.http_app(stateless_http=True, json_response=True, host_origin_protection=True, allowed_hosts=[host])


@asynccontextmanager
async def lifespan(_: FastAPI):
    await backend.startup()  # DI 登録。呼ばないと、DI を解決するツールがすべて失敗する
    async with mcp_app.lifespan(mcp_app):  # mount したアプリの lifespan は動かないため、ここで動かす
        yield
    await backend.shutdown()


app = FastAPI(lifespan=lifespan)
app.mount("/", mcp_app)  # エンドポイントは /mcp

• 認証は、IdP が発行した JWT を JWTVerifier で検証し、RemoteAuthProvider で認可サーバーの場所を知らせる。この形は、IdP が動的クライアント登録 (DCR) に対応している前提。 対応していない IdP は、FastMCP の専用のプロバイダーを auth= に渡す (Cognito は AWSCognitoProvider、Identity Platform は GoogleProvider、Auth0 の Auth for MCP は Auth0MCPProvider)
• ツールの中では、検証済みのトークンを fastmcp.server.dependencies.get_access_token() で読む。IdP のユーザー ID は claims["sub"]
• auth= を外すのは、localhost だけで動かすときに限る。外したまま公開すると、書き込みのツールも誰でも呼べる
• ツール名は全モジュールで 1 つの名前空間になる。重なると CompositeModule.tools が ValueError を送出する (FastMCP は警告だけで後勝ちに置き換えるため)
• ツールは Tool.from_function で、関数かメソッドから作る。Tool.from_tool やプロキシのツール、functools.partial や呼び出し可能なインスタンスから作ったツールは、例外を変換できないため CompositeModule.tools が断る
• ツール名は name= に ASCII ([A-Za-z0-9_.-]) で渡す。既定は関数名で、日本語のメソッド名のままだと、名前の形式を強制するクライアントに拒まれる
• ツールの引数・戻り値の型は実行時に import する。Tool.from_function は登録の時点で型ヒントを解決するため、TYPE_CHECKING に移すと NameError になる。ruff の TC001〜TC003 は、ツールのファイルでは外す (per-file-ignores)
• ツールは値を返す同期関数 (def) で書く。async def とジェネレータは CompositeModule.tools が断る。UnitOfWork はスレッドごとに DB のセッションを持ち、async のツールはイベントループの 1 スレッドで動くため、await の間に別の呼び出しのトランザクションが混ざる。ジェネレータは本体が変換の外で動く
• そのため、FastMCP のバックグラウンドタスク (task=True、async の関数を求める) は使えない
• ツールの説明 (description) はモデルが読み、ツールを選ぶのに使う。docstring を使うと実装のメモまで渡るため、モデル向けの説明は description= で渡す
• tools は起動 (DI 登録) の前に呼ばれる。Tool の元のクラスのコンストラクタでは DI を解決せず、ツールが呼ばれたときに解決する
• 同期 (def) のツールは、FastMCP がスレッドで動かす。API の def のルートと同じく、@transactional のアプリケーションサービスをそのまま呼べる
• ツールの中で起きたドメインイベントは、PublishNotificationsOnToolCall で発行する。ツールの呼び出しが成功したときだけ、応答の前に発行し、initialize・tools/list などでは発行しない (発行のまとめ方・失敗の扱いは「📮 API の応答で通知を発行する」と同じ)。HTTP に依らないので stdio でも動く

テーブルクラスは sqlalchemy を直接 import するため、uv add sqlalchemy で直接の依存として宣言する (fastship-py 経由で入っていても、宣言しないと deptry が推移的な依存 (DEP003) として検出する)。

〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver.users_table_row.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

# NOTE: テーブルクラスは `DataBase` を継承して作ること
from fastship.common.port.adapter.persistence.sqlalchemy import DataBase


class UsersTableRow(DataBase):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)  # index 名は ix_users_name

〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver.__init__.py

from .users_table_row import UsersTableRow
# 他のテーブルクラス

テーブルクラスを足したら、そのモジュールの tables にドライバーモジュールのパスを足す

class 〇〇(AppModule):
    @property
    @override
    def tables(self) -> list[str]:
        return ["〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver"]

事前にマイグレーション用のパッケージを生成

uv add alembic
uv run alembic init migration

migration/env.py:

# alembic init の生成物から、target_metadata = None をこの 3 行に差し替える

from fastship.common.port.adapter.persistence.sqlalchemy import DataBase
from core import Backend  # 全モジュールを束ねた CompositeModule

target_metadata = DataBase.metadata_of(Backend())

接続先は DATABASE_WRITER_URL / DATABASE_READER_URL で注入する (冒頭の「⚙️ 環境変数」を参照)。

from fastship.common.port.adapter.persistence.sqlalchemy import SQLAlchemyUnitOfWork

unit_of_work = SQLAlchemyUnitOfWork.from_settings(pool_pre_ping=True)

トランザクション (unit_of_work.session()) は、読み取りも含めてすべてプライマリで実行する。 レプリカの遅延した値を元に更新すると、プライマリの新しい値を上書きしてしまう (lost update) ため。 更新系の生 SQL (text()) もここで実行する。

トランザクション外の読み取り (unit_of_work.query()) は、以下の規則でリーダーとライターに振り分ける。

with unit_of_work.query() as session:
    session.use_writer()
    session.get(...)

認可の判定 (無効化された API キー、停止されたユーザー、取り消された権限など) は、レプリカ遅延の間に 古い状態で許可しないよう、session.use_writer() を呼んでプライマリから読むこと。

Read Pool のノード数を増減しても、エンドポイントは変わらないためアプリ側の変更は要らない。 SessionPreparer (RLS の SET LOCAL など) は、振り分け先のコネクションそれぞれに適用される。

ドメインイベントは、集約の更新と同じトランザクションで stored_events に保存する (outbox)。中継が未発行のイベントを MQ に発行する。 受信側は、listener の処理と同じトランザクションで consumed_notifications に記録し、同じ通知が 2 回届いても 1 回だけ処理する (inbox)。

テーブル (stored_events / consumed_notifications) と、ストア・中継の DI 登録は Common が行う (「🧩 既定のモジュール」)。 ストアは UnitOfWork に登録した SQLAlchemyUnitOfWork のセッションに書く

中継 (NotificationPublisherImpl)

• 未発行 (published_at が空) の行を FOR UPDATE SKIP LOCKED で 100 件まで取り出し、NOTIFICATION_PUBLISH_TO に発行して published_at を付ける
• 発行済みを行ごとに記録するため、id の小さいイベントが後から commit されても取りこぼさない
• 中継が同時に走っても、ほかの中継がロックしている行は飛ばして分担する
• 取り出した行はまとめて MQ に渡し、まとめて待つ (MessagePublisher.publish_all)。Pub/Sub は 1 回の RPC にまとめて送り、障害中も待つのは全体で 30 秒まで (CloudPubSubPublisher.PUBLISH_TIMEOUT)
• 発行できた行だけを発行済みにする。発行できなかった行は次の中継で発行する。まとめて送れない MQ の実装 (publish_all を上書きしないもの) は 1 件ずつ送り、最初の失敗で打ち切る
• 発行に失敗した行は後回しにする (失敗した回数の少ない順に取り出す)。発行できないイベントがあっても、ほかのイベントは止まらない
• 同じイベントを 2 回発行するのは、発行した後 commit する前に中継が落ちたときと、待つのをやめた (30 秒を過ぎた) 発行が後から届いたとき。受信側の inbox が重複を捨てる

処理済みの記録は、Cloud Scheduler などから ConsumedNotificationApplicationService.prune_consumed_notifications() を定期的に呼んで消す (既定は 14 日を過ぎたもの)。 保持期間は、MQ が再配送しうる期間 (Pub/Sub は最大 7 日) より長く取る。短いと、消した後に届いた再配送を二重に処理する。

GCP の SDK は extras に分けている。gcp を付けて入れる

uv add 'fastship-py[gcp]'

接続先は GOOGLE_CLOUD_PROJECT / PUBSUB_EMULATOR_HOST で注入する (冒頭の「⚙️ 環境変数」を参照)。 MessagePublisher / MessageSubscriber の DI 登録は Common が行う (「🧩 既定のモジュール」)。Pub/Sub 以外の MQ を使うときは、後ろのモジュールで差し替える。

• 発行は Pub/Sub が受け取るまで同期で待つ (最大 300 秒)。async の処理から呼ぶときは、asyncio.to_thread などでスレッドに逃がす
• 発行に失敗すると例外になる (黙って捨てない)
• 本文 (Notification の JSON) に加えて、event_type と partition_key を属性にも載せる。サブスクリプションのフィルタで、購読するイベントを絞れる

gcloud pubsub subscriptions create billing --topic=notifications \
  --message-filter='attributes.partition_key = "tenant-1"'

flowchart LR
    publisher["CloudPubSubPublisher"] --> topic[["Pub/Sub トピック"]]
    topic --> push["push サブスクリプション<br>(サービスアカウントの OIDC トークンを付ける)"]
    push -->|"Cloud Run の IAM が<br>run.invoker を確かめる"| subscriber["subscriber の Cloud Run<br>(CloudPubSubSubscriber が OIDC トークンを検証する)"]
    subscriber --> listener["ExchangeListener"]

CloudPubSubSubscriber は、push サブスクリプションが付けた OIDC トークン (JWT) の署名・有効期限・発行者・audience を検証し、 設定したサービスアカウントからのリクエストだけを受け付ける。 設定 (PUBSUB_PUSH_AUDIENCE / PUBSUB_PUSH_SERVICE_ACCOUNT。冒頭の「⚙️ 環境変数」を参照) が無いと、 CloudPubSubSubscriber.from_settings() の時点で例外になる。検証しないのは PUBSUB_EMULATOR_HOST があるときだけ。

受け口のルートはアプリに書く

from fastapi import FastAPI, Request
from fastship.common.port.adapter.messaging.pubsub import CloudPubSubSubscriber

from core import Backend  # 全モジュールを束ねた CompositeModule

app = FastAPI()
subscriber = CloudPubSubSubscriber.from_settings()
subscriber.add(Backend().listeners)


@app.post("/pubsub/push", status_code=204)
async def push(request: Request) -> None:
    # 2xx 以外を返すと Pub/Sub が再配送する。同じメッセージが 2 回届いても、inbox が重複を捨てる
    await subscriber.receive(await request.json(), request.headers.get("Authorization"))

• 送信元を認証できないと SystemException (COMMON_2002) になり、add_exception_handlers の例外ハンドラが 401 を返す
• 読めないメッセージも再配送され続けるため、push サブスクリプションにはデッドレタートピックを設定する

トークンの検証に加えて、Cloud Run の IAM でも入口を閉じておく (設定を誤ったときの二重の守り)。

• subscriber の Cloud Run は、未認証の呼び出しを拒否する (--no-allow-unauthenticated)
• roles/run.invoker は、push サブスクリプションのサービスアカウントにだけ付ける

PublishNotificationsMiddleware は、API が 2xx を返すリクエストで NotificationApplicationService.publish_notifications() を呼び、 outbox に溜まった未発行の通知を MQ に発行する。発行と、発行済みの記録は 1 つのトランザクションで行う (中継の動きは「📬 outbox / inbox」)。 NotificationPublisher の実装 (NotificationPublisherImpl) は Common が DI コンテナに登録する。API と、push を受ける worker の両方に入れる。

from fastship.common.middleware import PublishNotificationsMiddleware

app.add_middleware(PublishNotificationsMiddleware, excluded_paths={"/health/check"})

• 応答の最後の本文を送る直前に発行し、応答は発行を待つ。Cloud Run の cpu_idle (CPU をリクエストの処理中だけ割り当てる) のままで動く
• 応答を返した後に発行を残さない (FastAPI の BackgroundTasks を使わない)。リクエストの外では CPU が絞られ、発行が止まるため
• 発行は同時に 1 本だけ行い、同期の発行はスレッドに逃がす。イベントループを塞がず、他のリクエストの処理は止まらない
• 同じプロセスでは発行を並行させない。発行中に来たリクエストは、今の発行の後の 1 回にまとめて待つ。リクエストごとに並行に発行すると、発行ごとに DB の接続とスレッドを MQ の応答まで握り、リクエストの処理に回す分を食うため
• 発行に失敗しても応答は 2xx のまま返り、失敗はログに残る。未発行の通知は outbox に残り、次の発行で送られる
• 発行は、リクエストの文脈 (ContextVar に置いたテナントなど) を引き継がない
• 複数のインスタンスからは並行に発行しうる。未発行の通知は中継どうしで分担し、同じ通知を二重に発行しない
• MCP サーバーは PublishNotificationsMiddleware ではなく PublishNotificationsOnToolCall を使う。HTTP の 2xx ごとに発行すると、initialize・tools/list などのプロトコルのメッセージ 1 件ごとに outbox を読むため (「🤖 MCP」)

応答の遅れが気になるとき

• 発行は、溜まった通知 (100 件まで) をまとめて送り、Pub/Sub が受け取るまで最大 30 秒待つ。発行中に来たリクエストは次の 1 回も待つため、Pub/Sub の障害中は応答が最大で約 60 秒遅れる
• 発行が重くなったら、ミドルウェアを外し、Cloud Tasks のキュー (または Cloud Scheduler) から NotificationApplicationService.publish_notifications() を呼ぶエンドポイントを叩く。API の処理と発行が切り離され、再試行もキューに任せられる

登録方法:

from sqlalchemy import create_engine

from fastship.common.application import UnitOfWork
from fastship.common.port.adapter.persistence.inmem import InMemUnitOfWork
from fastship.common.port.adapter.persistence.sqlalchemy import SQLAlchemyUnitOfWork
from fastship.di import DI, DIContainer

engine = create_engine("postgresql+psycopg://...")

DIContainer.instance().register(
    DI.of(UnitOfWork, {"InMem": InMemUnitOfWork()}, SQLAlchemyUnitOfWork(engine)),
)

利用方法:

from fastship.di import DI, DIContainer

# `DI_FOR_PY=InMem` なら InMemUnitOfWork のインタンスを取得
# `DI_FOR_PY=` なら SQLAlchemyUnitOfWork のインタンスを取得
unit_of_work = DIContainer.instance().resolve(UnitOfWork)

from injector import inject


class 〇〇ApplicationService:
    @inject
    def __init__(self, 〇〇_repository: 〇〇Repository) -> None: ...

from fastship.common.domain.model import DomainRegistry

〇〇_repository = DomainRegistry.resolve(〇〇Repository)

ErrorJson は RFC 9457 の Problem Details を表す。ErrorCode から組み立てる。

• type は /errors/{コード名} になる。クライアントは type の完全一致で分岐する
• SystemException.detail はログ用で内部の情報を含みうるため、detail には渡さない

add_exception_handlers(app) は、例外を ErrorJson (application/problem+json) の応答に変える例外ハンドラを登録する。

• ValueError / AssertionError のログには、例外の型と送出した場所だけを出す (メッセージには入力値が入りうる)
• 応答を変えたい例外は、add_exception_handlers(app) の後に app.add_exception_handler で登録する (同じ例外なら後から登録したものが勝つ)。応答は ErrorJsonResponse.of(エラーコード, request) で返す

{"type": "/errors/USER_NOT_FOUND", "title": "ユーザーが見つかりません", "status": 404, "instance": "https://api.example.com/users/1"}

設定は fastship が持たない。テンプレートの pyproject.toml にある設定を、プロジェクトで自由に変える。

• fastship-py 経由で入る依存 (fastapi / injector / pydantic / pydantic-settings / sqlalchemy) も、import するならプロジェクトの dependencies に直接書く
• src 直下のモジュールは [tool.deptry] の known_first_party に並べる (並べないと自分のモジュールを未宣言の依存とみなす)

[tool.deptry.per_rule_ignores]
DEP002 = ["uvicorn"]  # import せずに使う依存

テンプレートの pyproject.toml の [tool.importlinter] が規約を有効にしている。

• 規約の中身 (層の順序や許可する import) は fastship のコード (fastship.linter.architecture) にあり、fastship-py を更新すると変わる
• root_packages には src 直下のモジュールをすべて並べる。足し忘れは lint-imports が検出する
• import-linter は root_packages が空だと動かない。テンプレートが見本の example モジュールを持つのはそのため

プロジェクト独自の設定や contract も [tool.importlinter] に足す

[tool.importlinter]
root_packages = ["billing"]
include_external_packages = true  # sqlalchemy などの外部パッケージを forbidden_modules に書くときに要る

[[tool.importlinter.contracts]]
id = "no-sqlalchemy-in-domain"
name = "ドメイン層は永続化の技術を知らない"
type = "forbidden"
source_modules = ["billing.domain"]
forbidden_modules = ["sqlalchemy"]

ヘキサゴナルアーキテクチャ (モジュール内部の依存)

モジュラモノリス (モジュール間の依存)