技術資料

以前、Pythonで読み取り専用MCPサーバーを実装し、Codexから別プロジェクトの情報を取得するという記事で、Pythonの標準ライブラリを使ったMCPサーバーを作成しました。

ただ、Pythonに触れたことがない場合、MCPの処理以前に、変数や関数、辞書、JSON、標準入出力などの部分でコードを読みづらく感じるかもしれません。

今回は、MCPサーバーのコードを読むために必要なPythonの基本だけを確認します。Python全般を網羅するのではなく、WindowsのPowerShellから短いサンプルを実行しながら、標準ライブラリ、JSON、標準入出力、例外処理を見ていきます。

前提

今回の確認環境は以下の通りです。

• OS: Windows 11
• Python: 3.9.7
• ターミナル: PowerShell
• 外部パッケージは使用しない

Pythonのインストール方法や仮想環境、pip、テストコードはこの記事では扱いません。Pythonが実行できる状態から始めます。

Pythonの基本文法については、Python公式ドキュメントのチュートリアルにも説明があります。この記事では、その中からMCPのコードを読むために必要な範囲だけを取り上げます。

また、記事中のサンプルはMCPサーバーの処理を理解するための教材です。完全なMCPサーバーを作成するものではありません。実際のMCPサーバーを実装してCodexから呼び出す手順は、先ほどの記事を参照してください。

Pythonスクリプトを実行する

まず、PowerShellでPythonのバージョンを確認します。

python --version

今回の環境では、次のように表示されました。

Python 3.9.7

Pythonのコードは、拡張子が.pyのファイルへ保存して実行できます。例えば、hello.pyを作成して次の内容を記述します。

message = 'Hello, Python'
print(message)

PowerShellから実行します。

python .\hello.py

PHPなどの他の言語と同じように、Pythonでも上から順番に処理が実行されます。ただし、Pythonではインデントが処理のまとまりを表すため、後で出てくるコードでは空白の位置にも注意します。

MCPコードを読むためのPython基本文法

まず、MCPとは関係のない短いコードで、変数、辞書、リスト、関数を確認します。

document = {
    'title': 'MCP入門',
    'tags': ['MCP', 'Python']
}


def describe(document):
    title = document['title']
    tags = document['tags']
    return f'{title}: {len(tags)}個のタグ'


print(describe(document))

documentは辞書です。titleやtagsというキーに対して、それぞれ文字列とリストを保存しています。他の言語では「連想配列」とか呼ばれるやつです。

describeは関数です。引数としてdocumentを受け取り、辞書から値を取り出して、returnで結果を返しています。

forやifも、MCPサーバーのコードでよく使います。例えば、リストの各要素を順番に確認する処理は次のように書けます。

documents = [
    {'title': 'MCPの概要', 'text': 'AIと外部のデータを接続します。'},
    {'title': 'Pythonの基本', 'text': '標準ライブラリを使います。'}
]

for document in documents:
    if 'Python' in document['text']:
        print(document['title'])

MCPサーバーでは、検索対象の配列をforで繰り返し、検索語を含む項目だけを結果へ追加する、といった処理を行います。

標準ライブラリのJSONを使う

Pythonには、追加インストールなしで利用できる標準ライブラリが用意されています。JSONを扱うときは、標準ライブラリのjsonをimportして使います。詳しい仕様はPython公式ドキュメントのjsonモジュールを参照してください。

import json

request_text = '{"method": "search", "query": "Python"}'
request = json.loads(request_text)

response = {
    'ok': True,
    'query': request['query'],
    'results': []
}

print(json.dumps(response, ensure_ascii=False))

json.loadsは、JSON形式の文字列をPythonの値へ変換します。JSONのオブジェクトはPythonの辞書、JSONの配列はPythonのリストとして扱われます。

反対に、json.dumpsはPythonの辞書やリストをJSON形式の文字列へ変換します。ensure_ascii=Falseを指定すると、日本語をエスケープせずに出力できます。

ファイルに保存されたJSONを読む場合はjson.load、ファイルへ書き込む場合はjson.dumpを使います。MCPサーバーの実装では、設定ファイルやインデックスを読む処理で使われることがあります。

なお、JSONは外部から受け取ったデータを無条件に読み込んでよいという意味ではありません。入力サイズを制限するなど、受け取るデータの大きさも確認する必要があります。

標準入力と標準出力を使う

MCPのSTDIOトランスポートでは、クライアントがMCPサーバーを子プロセスとして起動します。サーバーは標準入力からJSON-RPCメッセージを読み取り、標準出力へ応答を返します。詳しい仕様はMCP公式仕様のTransportsを確認してください。

Pythonでは、標準入力と標準出力をsys.stdinとsys.stdoutで扱えます。これらの機能はPython公式ドキュメントのsysモジュールで説明されています。次の例では、標準入力から1行ずつ読み取り、受け取った値をJSONで返します。

import json
import sys


for line in sys.stdin:
    if not line.strip():
        continue

    request = json.loads(line)
    response = {
        'ok': True,
        'query': request.get('query', '')
    }
    print(json.dumps(response, ensure_ascii=False), flush=True)

for line in sys.stdinは、標準入力を1行ずつ処理します。line.strip()は、前後の空白や改行を取り除いた結果を確認しています。

printは標準出力へ文字列を出力します。flush=Trueを付けると、出力をバッファーへ残さず、すぐに書き出すよう指示できます。別のプロセスと1行ずつ通信する処理では、応答を相手へ渡すタイミングが重要になります。

標準エラー出力へメッセージを出す場合は、printにfile=sys.stderrを指定します。

print('デバッグ情報', file=sys.stderr)

MCPのSTDIOでは、標準出力へ有効なMCPメッセージ以外を出力してはいけません。デバッグログを標準出力へ出すと、クライアントがログをMCPメッセージとして解釈しようとして通信に失敗する可能性があります。ログは標準エラー出力へ出します。

例外処理で入力エラーを扱う

JSONが壊れている場合、json.loadsはjson.JSONDecodeErrorを発生させます。Pythonの例外についてはPython公式ドキュメントの組み込み例外を参照してください。例外が処理されないままスクリプトが終了すると、MCPクライアントへ適切な応答を返せません。

tryでエラーが発生する可能性のある処理を囲み、exceptで例外を処理します。

import json

request_text = '{"method": "search"}'

try:
    request = json.loads(request_text)
    print(request['query'])
except json.JSONDecodeError:
    print('JSONの形式が正しくありません')
except KeyError:
    print('queryが指定されていません')

この例では、JSONの形式が正しくない場合と、JSONは正しいもののqueryキーがない場合を分けて扱っています。エラーの種類を分けると、原因に応じたメッセージや応答を返しやすくなります。

短いサンプルコードでMCPの処理を読む

ここまでの内容を使って、MCPのtools/callを簡略化したサンプルコードを作ります。このコードは検索処理の代わりに、受け取った検索語を応答へ入れて返します。

import json
import sys


def error_response(request_id, message):
    return {
        'jsonrpc': '2.0',
        'id': request_id,
        'error': {'code': -32600, 'message': message}
    }


def handle_request(request):
    if not isinstance(request, dict):
        raise ValueError('request must be an object')

    request_id = request.get('id')
    if request.get('method') != 'tools/call':
        raise ValueError('method must be tools/call')

    params = request.get('params', {})
    if not isinstance(params, dict):
        raise ValueError('params must be an object')
    if params.get('name') != 'search_files':
        raise ValueError('tool must be search_files')

    arguments = params.get('arguments', {})
    if not isinstance(arguments, dict):
        raise ValueError('arguments must be an object')

    query = arguments.get('query')
    if not isinstance(query, str) or not query.strip():
        raise ValueError('query is required')

    return {
        'jsonrpc': '2.0',
        'id': request_id,
        'result': {
            'content': [
                {'type': 'text', 'text': f'検索語: {query.strip()}'}
            ]
        }
    }


def serve():
    for line in sys.stdin:
        if not line.strip():
            continue

        request = {}
        try:
            request = json.loads(line)
            response = handle_request(request)
        except json.JSONDecodeError:
            response = error_response(None, 'invalid JSON')
        except ValueError as error:
            print(str(error), file=sys.stderr)
            request_id = request.get('id') if isinstance(request, dict) else None
            response = error_response(request_id, str(error))

        print(json.dumps(response, ensure_ascii=False), flush=True)


if __name__ == '__main__':
    serve()

このコードは、次のような処理に分かれています。

1. importでJSONと標準入出力の機能を読み込む。
2. error_responseでエラー応答の辞書を作る。
3. handle_requestで辞書のキーを確認し、正常時の応答を返す。
4. serveで標準入力を1行ずつ読み取る。
5. try／exceptで不正なJSONや入力値のエラーを処理する。
6. json.dumpsで応答をJSONへ変換し、標準出力へ返す。

if __name__ == '__main__'は、このファイルを直接実行した場合だけserve()を呼び出すための記述です。別のPythonファイルからimportされた場合は、読み込んだだけでサーバー処理が始まりません。

実際のMCPサーバーでは、初期化、ツール一覧、ツール呼び出し、JSON-RPCのエラーなど、さらに多くの処理が必要です。ただし、関数、辞書、リスト、JSON、標準入出力、例外処理という基本的な構造は、この短いコードと共通しています。

Windowsで動作を確認する

サンプルコードをmcp_learning.pyという名前で保存し、PowerShellから実行します。

python .\mcp_learning.py

このプログラムは標準入力を待ち続けるため、別のPowerShellから入力をパイプで渡します。

@'
{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"search_files","arguments":{"query":"Python"}}}
'@ | python .\mcp_learning.py

次のようなJSONが標準出力へ返ります。

{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", "text": "検索語: Python"}]}}

検索語を指定しない入力も確認します。

@'
{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"search_files","arguments":{}}}
'@ | python .\mcp_learning.py

この場合は、標準出力へエラー応答が返り、標準エラー出力へquery is requiredが出力されます。標準出力と標準エラー出力を分けているため、MCPの通信データへデバッグメッセージを混ぜずに確認できます。

JSONの途中で入力を終わらせると、json.JSONDecodeErrorの処理が使われます。

'{"jsonrpc":"2.0",' | python .\mcp_learning.py

この場合も、標準出力へinvalid JSONのエラー応答が返ります。

既存のMCPサーバー記事との関係

今回のサンプルコードでは、実際のファイル検索やMCPサーバーの登録処理は省略しています。Pythonの構文とデータの流れを確認することを優先しました。

実際のMCPサーバーでは、index.jsonを読み込む関数、検索結果を作る関数、ツール一覧を定義する辞書、Codexから起動するための設定などが加わります。今回確認した内容を踏まえて、既存記事のload_index、search_files、handle、serveの処理を順番に読むと理解しやすくなります。

注意点

• この記事のサンプルコードは、MCPサーバーの完全な実装ではない。
• 標準出力へデバッグログを出すと、STDIO通信を壊す可能性がある。
• 外部から受け取ったJSONを処理する場合は、入力サイズや値の形式を確認する。
• ファイル読み取りやコマンド実行などの機能をMCPへ追加する場合は、公開する権限と対象パスを限定する。
• PythonのバージョンやCodex、MCPの仕様が変わる可能性があるため、実装時は公式資料も確認する。

まとめ

今回は、MCPサーバーのコードを読むために、Pythonの変数、辞書、リスト、関数、標準ライブラリ、JSON、標準入出力、例外処理を確認しました。

MCPのSTDIOサーバーでは、標準入力からJSONを読み取り、処理結果を標準出力へJSONとして返します。標準出力へ通信以外の文字列を出さず、ログは標準エラー出力へ出すという点も重要です。

Python全体を一度に学ぼうとせず、実際に読みたいMCPサーバーのコードに登場する文法から確認すると、処理の流れを追いやすくなります。今回の内容を確認したら、次はPythonで読み取り専用MCPサーバーを実装し、Codexから別プロジェクトの情報を取得するの記事へ進むと、より実際の構成を確認できます。