第2回: Pythonで自作MCPサーバを実装する
前回は、MCPサーバを「AIエージェントと外部サービスをつなぐ翻訳レイヤー」として整理しました。
概念としては分かったものの、まだ少しつかみどころがありません。
普通のPythonプログラムと何が違うのでしょうか。関数を1つ用意すれば、それだけでMCPのtoolになるのでしょうか。そもそも、Codexはどうやってその関数を見つけるのでしょうか。
この辺の理解について実際に開発を進めていきながら理解を深めていきます。
今回はPythonで get_issue_context というtoolを公開します。最初からBacklog APIにつなぐと、MCPの話とAPI連携の話が混ざってしまうので、まずは固定値を返すだけのtoolを作ります。
固定値がMCP経由で返ってくることを確認できたら、次の順番で処理を足していきます。
Backlog URLを受け取る
↓
許可されたBacklogスペースのURLか検証する
↓
URLから課題キーを取り出す
↓
Backlog APIから課題を取得する
↓
PRレビューで使いやすい形にして返す
今回のゴールは、許可したBacklogスペースの課題URLから、課題の件名と本文を読み取り専用で取得できるところまでです。
一応すでにこの連載の開発は完了してまして、実際に出来上がったMCPサーバについて下記のgithubより確認できます。
今回使う構成
MCPサーバはPythonで実装し、Docker Composeから起動します。
主に使うライブラリは次の2つです。
dependencies = [
"httpx==0.28.1",
"mcp==2.0.0",
]
mcp はPython向けのMCP SDK、httpx はBacklog APIへのHTTPリクエストに使います。
ディレクトリは次のように分けました。
mcp-server-backlog/
├── src/backlog_mcp/
│ ├── __main__.py
│ ├── config.py
│ ├── mcp/
│ │ └── server.py
│ ├── application/
│ │ └── get_issue_context.py
│ └── backlog/
│ ├── client.py
│ ├── dto.py
│ ├── errors.py
│ └── url.py
├── tests/
├── Dockerfile
├── compose.yaml
└── pyproject.toml
ファイルが少し多く見えますが、最初から全部を理解する必要はありません。
まず見るのは mcp/server.py です。ここが「Pythonの処理をMCPのtoolとして外へ見せる」部分になります。
まずは固定値を返してみる
理解のために、実装をかなり小さくすると次のようになります。
from typing import Any
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
server = MCPServer(
name="backlog-pr-review",
title="Backlog PR Review",
description="Read-only Backlog context provider for pull request review",
version="0.1.0",
)
@server.tool(
name="get_issue_context",
description="Read and normalize review context for a Backlog issue URL.",
annotations=ToolAnnotations(
readOnlyHint=True,
destructiveHint=False,
idempotentHint=True,
openWorldHint=False,
),
structured_output=True,
)
async def get_issue_context(backlog_url: str) -> dict[str, Any]:
return {
"issue": {
"key": "PROJECT-123",
"summary": "固定値で返した課題",
"description": "まだBacklog APIには接続していません。",
},
"comments": [],
"change_logs": [],
"relationships": {
"parent": None,
"children": [],
"related": [],
},
"retrieval": {
"source_url": backlog_url,
"partial": False,
"warnings": [],
},
}
if __name__ == "__main__":
server.run(transport="stdio")
普通のPython関数との違いは、@server.tool が付いていることです。
このデコレータによって、get_issue_context がMCPのtoolとして公開されます。関数名だけではなく、toolの名前、説明、入力schema、出力形式もMCP clientから確認できるようになります。
入力schemaはどこに書かれているのでしょうか。
この例では、関数の引数にある backlog_url: str からSDKがschemaを組み立てます。そのため、MCP clientから見ると backlog_url は必須の文字列として扱われます。
structured_output=True を指定しているのは、戻り値の dict を構造化された結果として返したいからです。後でPRレビュースキルが issue や comments を取り出すので、長い文章を1つ返すよりも都合がよさそうです。
ToolAnnotations では、このtoolの性質も伝えています。
- 読み取り専用である
- 破壊的な操作をしない
- 繰り返し呼んでも、追加の副作用を起こさない
- 接続先を設定済みのBacklogスペースに限定する
ただし、ここには少し注意が必要です。
readOnlyHint=True と書くだけで、書き込み処理が技術的に禁止されるわけではありません。これはclientへ渡すヒントです。実際に読み取り専用にするには、MCPサーバの実装から更新系APIを呼ばない設計にしておく必要があります。
Codexはtoolをどうやって見つけるのか
toolを1つ作りましたが、CodexはPythonのソースコードを直接読んで関数を探しているわけではありません。
MCP clientはサーバへ接続したあと、利用できるtoolの一覧を問い合わせます。今回のPython SDKを使うと、client側は次のように確認できます。
import asyncio
import os
import sys
from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main() -> None:
parameters = StdioServerParameters(
command=sys.executable,
args=["-m", "backlog_mcp"],
env=dict(os.environ),
)
async with Client(stdio_client(parameters)) as client:
tools = await client.list_tools()
print([tool.name for tool in tools.tools])
result = await client.call_tool(
"get_issue_context",
{
"backlog_url": (
"https://your-space.backlog.jp/view/PROJECT-123"
)
},
)
print(result.structured_content)
asyncio.run(main())
ここでは、接続時の初期化を済ませたあとに list_tools() でtool一覧を取得し、call_tool() で get_issue_context を呼んでいます。
MCPのメッセージとして見ると、initialize、tools/list、tools/call というやり取りです。SDKを使うとJSON-RPCのメッセージを自分で組み立てる必要はありません。
これで固定値が返ってくれば、少なくとも次の部分は動いています。
MCP client
↓ tools/call
Python MCPサーバ
↓
get_issue_context
↓ structured result
MCP client
Backlog APIにはまだ接続していません。それでも、MCPサーバとしてtoolを公開し、clientから発見して呼び出すところまでは確認できます。
最初に固定値を返したのは、この境目を確認したかったからです。いきなりAPIへ接続して失敗すると、MCPの設定が悪いのか、URLが悪いのか、API keyが悪いのかが分かりにくくなります。
stdioで通信するときの注意
今回は、前回の記事で触れたstdio transportを使います。
server.run(transport="stdio") を実行すると、MCPサーバは標準入力からJSON-RPCメッセージを受け取り、標準出力へ結果を返します。
ここで、つい次のような確認をしたくなります。
print("MCP server started")
しかし、この print をMCPサーバ側の標準出力へ出すと、JSON-RPCの通信に関係のない文字列が混ざってしまいます。clientからすれば、通信データの途中に突然別の文字列が現れることになります。
では、デバッグログはどこへ出せばよいのでしょうか。
stdio transportを使う場合、通常のログは標準エラー出力へ流します。標準出力はMCP通信用として空けておく、と覚えておくとよさそうです。
固定値をBacklogの課題に置き換える
MCPとして呼び出せることは確認できました。次は固定値ではなく、実際のBacklog課題を返します。
toolへ渡すのは課題キーではなく、PR本文に貼られているBacklogの課題URLです。
{
"backlog_url": "https://your-space.backlog.jp/view/PROJECT-123"
}
URLを受け取ったのなら、そのURLへそのままHTTPリクエストを送ればよいのでしょうか。
これは避けます。
PR本文はレビュー対象のデータであり、信頼できる設定値ではありません。そこに書かれたURLを無条件に信用すると、想定していないhostへMCPサーバから接続できてしまいます。
そこで、受け取ったURLは接続先として使わず、まず extract_issue_key へ渡します。
from urllib.parse import urlsplit
def extract_issue_key(backlog_url: str, base_url: str) -> str:
target = urlsplit(backlog_url.strip())
allowed = urlsplit(base_url)
target_origin = (
target.scheme,
target.hostname,
target.port or 443,
)
allowed_origin = (
allowed.scheme,
allowed.hostname,
allowed.port or 443,
)
if target_origin != allowed_origin:
raise BacklogUrlError(
"Backlog issue URL is outside the configured space"
)
path_parts = target.path.rstrip("/").split("/")
if (
len(path_parts) != 3
or path_parts[0] != ""
or path_parts[1] != "view"
):
raise BacklogUrlError(
"Backlog issue URL must use /view/<issue-key>"
)
return validate_issue_key(path_parts[2])
実際のコードでは、これに加えてURL内のuser infoや不正なportも検証しています。
許可するのは、環境変数 BACKLOG_BASE_URL に設定したoriginと一致し、pathが /view/<issue-key> になっているURLだけです。
たとえば、次のURLは受け付けます。
https://your-space.backlog.jp/view/PROJECT-123
一方で、次のようなURLは拒否します。
http://your-space.backlog.jp/view/PROJECT-123
https://attacker.example/view/PROJECT-123
https://your-space.backlog.jp.evil.example/view/PROJECT-123
https://your-space.backlog.jp/issues/PROJECT-123
queryとfragmentが付いている場合は、そこを接続先へ引き継がず、課題キーだけを取り出します。
入力:
https://your-space.backlog.jp/view/PROJECT-123?from=pr#comment
API clientへ渡す値:
PROJECT-123
つまり、MCPサーバが実際にアクセスするURLを、PR本文の値に決めさせないようにしています。
Backlog APIを呼び出す
検証済みの課題キーを取得できたら、ようやくBacklog APIを呼び出します。
課題取得に必要な設定は環境変数から読み込みます。
BACKLOG_BASE_URL=https://your-space.backlog.jp
BACKLOG_API_KEY=your-api-key
API keyをソースコードやPR本文へ書かないのはもちろんですが、ログへURLを出すときにも注意が必要です。Backlog APIではAPI keyをquery parameterへ付けるため、リクエストURLをそのままログへ出すとkeyまで残る可能性があります。
今回の BacklogClient では、httpx.AsyncClient を使って課題を取得します。
async def get_issue(self, issue_key: str) -> Issue:
normalized_key = validate_issue_key(issue_key)
payload = await self._get_json(
f"/api/v2/issues/{quote(normalized_key, safe='')}",
params={},
)
return parse_issue(payload)
_get_json の中では、設定済みのbase URLと、MCPサーバ側で組み立てたAPI pathを使います。
response = await self._client.get(
f"{self._settings.base_url}{path}",
params={**params, "apiKey": self._settings.api_key},
follow_redirects=False,
)
ここでも、入力されたBacklog URLへ直接アクセスしていないことが分かります。
httpx.AsyncClient を使っているのは、tool自体を async def で実装し、HTTP通信の待ち時間に処理を占有しないようにするためです。今回のようなI/O中心の処理とは相性がよさそうです。
また、実際のコードではtimeoutを設定し、HTTPステータスごとに例外を分けています。
-
401: API認証の失敗 -
403: アクセス権限の不足 -
404: 課題が見つからない -
429: rate limit - その他の通信失敗や不正なJSON
エラーを分けておくと、「課題が存在しない」のか「一時的に取得できない」のかを、後続の処理で区別できます。
MCPのtoolとBacklog APIをつなぐ
ここまでで、MCPのtool、URL検証、Backlog API clientができました。
最後に、それぞれを __main__.py で組み立てます。
from backlog_mcp.application.get_issue_context import GetIssueContext
from backlog_mcp.backlog.client import BacklogClient
from backlog_mcp.config import Settings
from backlog_mcp.mcp.server import build_server
def main() -> None:
settings = Settings.from_env()
use_case = GetIssueContext(
settings,
lambda: BacklogClient(settings),
)
build_server(use_case).run(transport="stdio")
if __name__ == "__main__":
main()
toolの関数から直接 httpx を呼ぶ形にもできます。それでも動きます。
では、なぜ GetIssueContext と BacklogClient を分けたのでしょうか。
MCPの都合とBacklog APIの都合を同じ場所へ詰め込むと、あとから処理を追いにくくなると考えたためです。
mcp/server.py
MCPのtool名、説明、schemaを扱う
application/get_issue_context.py
課題情報をどの順番で集め、どう返すかを扱う
backlog/client.py
Backlog APIのendpoint、認証、HTTPエラーを扱う
backlog/url.py
PR本文から渡されたURLの検証を扱う
この記事では課題本文の取得までを追いましたが、リポジトリの実装はすでにコメント、変更履歴、親課題、子課題、関連課題も同じtoolから返します。なぜそれらをMCPサーバでまとめるのか、どこからをPRレビュースキルの判断にするのかは次回整理します。
Docker Composeで動かしてみる
まず .env.example をもとに .env を用意し、自分のBacklogスペースとAPI keyを設定します。すでに .env がある場合は、上書きしないように内容を確認します。
BACKLOG_BASE_URL=https://your-space.backlog.jp
BACKLOG_API_KEY=your-api-key
Docker imageをbuildします。
docker compose build backlog-mcp
MCPサーバを起動し、tools/call まで確認するスクリプトも用意しました。
docker compose run --rm --no-deps backlog-mcp \
python scripts/probe_mcp.py \
"https://your-space.backlog.jp/view/PROJECT-123"
現在の実装はコメントや関連課題の取得まで含んでいるため、成功すると次のように取得件数や状態が表示されます。課題本文そのものは表示しません。
issue_key=PROJECT-123
comments=3
change_logs=1
children=0
related=0
partial=False
comments_truncated=False
本文を確認用スクリプトの標準出力へ出さないのは、動作確認のログへ業務上の情報を必要以上に残さないためです。
テストでは、実際に子プロセスとしてstdioサーバを起動し、get_issue_context がtool一覧に出てくることも確認しています。
docker compose run --rm --no-deps backlog-mcp pytest
固定値を返す段階では「MCPとしてつながるか」を確認し、Backlog APIへ接続したあとは「許可したURLだけを受理するか」「APIの結果を期待した形で返せるか」を確認する。この順番にしたことで、問題が起きた場所を切り分けやすくなりました。
作ってみて分かったこと
MCPサーバの最小部分は、思っていたより小さいものでした。
Python関数へデコレータを付け、stdioで起動すればtoolとして公開できます。MCP SDKがschema生成やJSON-RPCのやり取りを担当してくれるため、自分でプロトコルのメッセージを書く必要もありません。
一方で、実際の外部サービスへつなごうとすると、MCPとは別の難しさが出てきます。
受け取ったURLをどこまで信用するのか。API keyをどう扱うのか。timeoutやrate limitをどう表現するのか。取得したJSONを、そのまま返してよいのか。
今回コード量が増えたのは、toolを公開する部分よりも、Backlogとの境界を安全に作る部分でした。
この時点では、MCPサーバはBacklog課題を取得できるようになっただけです。取得した課題をどう読んで、PRレビューの判断へ使うかはまだ決めていません。
その判断までMCPサーバへ入れるべきなのでしょうか。それともPRレビュースキルに持たせるべきなのでしょうか。
次回は、このMCPサーバとPRレビュースキルの責務をどこで分けるか考えます。