AIに有価証券報告書を読ませてみる【EDINET×MCPサーバー】

IT・プログラミング

はじめに

本記事では、金融庁が提供する有価証券報告書等の開示システム「EDINET」とMCP（Machine Communication Protocol）サーバーを連携させる方法について解説します。このサーバーを使えば、EDINETから簡単に企業情報や有価証券報告書データを取得し、機械学習やデータ分析に活用できます。

MCPとは

MCPは、AIと他のシステムを連携させるためのプロトコルです。特にAIモデルとツールを接続し、AIがツールを使って実際の作業を行えるようにするための仕組みです。このプロトコルを利用することで、AIモデルがEDINETのAPIを呼び出し、データを取得・処理することが可能になります。

プロジェクトの概要

今回作成するMCPサーバーは以下の機能を提供します：

  1. 特定期間内の有価証券報告書のドキュメントID取得：指定した期間内に提出された有価証券報告書の情報を一覧として取得
  2. ドキュメントIDからCSVデータ取得：特定のドキュメントIDを指定して、その有価証券報告書のデータをCSV形式で取得

必要なもの

  • EDINET API用のAPIキー（EDINETのサイトから取得）
  • Docker（オプション）

ファイル構成

プロジェクトは以下の3つの主要ファイルで構成されています：

  1. edinet.py – MCPサーバーのメインコード
  2. Dockerfile – Dockerコンテナ作成用の設定ファイル
  3. pyproject.toml – Pythonプロジェクトの依存関係定義ファイル

コードの解説

1. edinet.py

このファイルにはMCPサーバーの主要なコードが含まれています。

import os
from urllib.parse import urlencode
from mcp.server.fastmcp import FastMCP

import datetime
import requests
import time
import pandas as pd
import unicodedata
import zipfile

# Initialize FastMCP server
mcp = FastMCP("edinet")

# Constants
api_key = os.environ.get("API_KEY")

def make_day_list(start_date:str, end_date:str):
    """Generate a list of dates within the specified date range"""
    start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d').date()
    end_date = datetime.datetime.strptime(end_date, '%Y-%m-%d').date()
    period = (end_date - start_date).days
    return [start_date + datetime.timedelta(days=d) for d in range(period + 1)]


@mcp.tool()
async def get_doc_id_data(start_date:str, end_date:str):
    """
    Fetch company information from EDINET and create a list
    Parameters:
    - start_date: Start date in 'YYYY-MM-DD' format
    - end_date: End date in 'YYYY-MM-DD' format
    """

    day_list = make_day_list(start_date, end_date)
    print(day_list)
    securities_report_data = []
    url = "https://disclosure.edinet-fsa.go.jp/api/v2/documents.json"

    # Fetch data for each day in the day list
    for day in day_list:
        params = {"date": day, "type": 2}
        params["Subscription-Key"] = api_key
        response = requests.get(url, params=params)
        if response.status_code == 200:
            try:
                # Attempt to parse the JSON response
                json_data = response.json()
            except ValueError:
                raise ValueError(f"Error: Failed to parse JSON response.\nResponse: {response.text}")

        # Filter the data based on specific conditions
        for result in json_data["results"]:
            ordinance_code = result["ordinanceCode"]
            form_code = result["formCode"]

            if ordinance_code == "010" and form_code == "030000":
                securities_report_data.append(result)
        time.sleep(1)

    # Convert data into a DataFrame
    df = pd.DataFrame(securities_report_data)
    if len(df) > 0:
        # Format the security code and normalize company names
        df['Security Code'] = df['secCode'].astype(str).str[:4]
        df["Company Name"] = df["filerName"].apply(lambda x: unicodedata.normalize('NFKC', x) if 'filerName' in df and isinstance(x, str) else None)

    return df

@mcp.tool()
async def get_csv_from_doc_id(doc_id:str):
    """Download a ZIP file from EDINET and extract the specified CSV file as a pandas DataFrame."""
    url = f"https://disclosure.edinet-fsa.go.jp/api/v2/documents/{doc_id}"
    params = {"type": 5, "Subscription-Key": api_key}
    filename = f"{doc_id}.zip"
    extract_dir = filename.replace('.zip', '')
    app_dir = os.path.join(extract_dir, "XBRL_TO_CSV")

    # Fetch the file from EDINET API
    response = requests.get(url, params=params, stream=True)
    if response.status_code == 200:
        with open(filename, 'wb') as file:
            for chunk in response.iter_content(chunk_size=1024):
                file.write(chunk)
        print(f"Download of ZIP file {filename} is complete.")

        # Extract the downloaded ZIP file
        zip_obj = zipfile.ZipFile(filename, 'r')

        # Create directory for extraction if it doesn't exist
        if not os.path.exists(extract_dir):
            os.makedirs(extract_dir)

        zip_obj.extractall(extract_dir)
        zip_obj.close()
        print(f"Extracted ZIP file {filename}.")

        # Find the first CSV file containing "jpcrp030000" in the specified directory
        for file in os.listdir(app_dir):
            if file.endswith(".csv") and "jpcrp030000" in file:
                csv_filepath = os.path.join(app_dir, file)
                break
        else:
            print(f"Error: No CSV file containing 'jpcrp030000' found in {app_dir}")
            return None

        # Read the CSV file into a pandas DataFrame
        try:
            df = pd.read_csv(csv_filepath, encoding="utf-16",sep="\t")
            print(f"Successfully read CSV file: {csv_filepath}")

            # Delete the zip file and extracted directory
            os.remove(filename)
            import shutil
            shutil.rmtree(extract_dir)
            print(f"Deleted zip file {filename} and extracted directory {extract_dir}")

            return df
        except FileNotFoundError:
            print(f"Error: CSV file not found at {csv_filepath}")
            return None
        except Exception as e:
            print(f"Error reading CSV file: {e}")
            return None
    else:
        print("Failed to download the ZIP file.")
        return None

if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport="stdio")

基本設定とライブラリのインポート

import os
from urllib.parse import urlencode
from mcp.server.fastmcp import FastMCP

import datetime
import requests
import time
import pandas as pd
import unicodedata
import zipfile

# MCPサーバーの初期化
mcp = FastMCP("edinet")

# 定数
api_key = os.environ.get("API_KEY")

ここでは必要なライブラリをインポートし、FastMCPサーバーをインスタンス化しています。APIキーは環境変数から取得します。

日付リスト生成関数

def make_day_list(start_date:str, end_date:str):
    """指定された日付範囲内の日付リストを生成する"""
    start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d').date()
    end_date = datetime.datetime.strptime(end_date, '%Y-%m-%d').date()
    period = (end_date - start_date).days
    return [start_date + datetime.timedelta(days=d) for d in range(period + 1)]

この関数は開始日と終了日を受け取り、その期間内のすべての日付をリストとして返します。

ドキュメントID取得ツール

@mcp.tool()
async def get_doc_id_data(start_date:str, end_date:str):
    """
    EDINETから企業情報を取得してリストを作成する
    パラメータ:
    - start_date: 開始日（'YYYY-MM-DD'形式）
    - end_date: 終了日（'YYYY-MM-DD'形式）
    """
    # 省略（詳細はコード参照）

この関数は指定期間内の有価証券報告書情報をEDINET APIから取得し、pandas DataFrameとして返します。特に「有価証券報告書」（ordinanceCode=”010″ かつ formCode=”030000″）をフィルタリングしています。

CSVデータ取得ツール

@mcp.tool()
async def get_csv_from_doc_id(doc_id:str):
    """EDINETからZIPファイルをダウンロードし、指定されたCSVファイルをpandas DataFrameとして抽出する"""
    # 省略（詳細はコード参照）

この関数は特定のドキュメントIDを使用してEDINETからZIPファイルをダウンロードし、その中から「jpcrp030000」を含むCSVファイルを抽出して、pandas DataFrameとして読み込みます。

メイン関数

if __name__ == "__main__":
    # サーバーの初期化と実行
    mcp.run(transport="stdio")

このコードはスクリプトが直接実行された場合に、MCPサーバーを標準入出力（stdio）トランスポートで起動します。

2. Dockerfile

FROM python:3.12-slim-bookworm

WORKDIR /app

COPY pyproject.toml ./
COPY edinet.py ./
COPY README.md ./

RUN pip install uv
RUN uv venv
RUN . .venv/bin/activate
RUN uv add "mcp[cli]" requests pandas 

CMD ["uv", "run", "edinet.py"]

このDockerfileはPython 3.12のSlimイメージをベースにして、必要なファイルをコピーし、uvを使って仮想環境をセットアップ、依存関係をインストールします。コンテナ起動時にはedinet.pyを実行します。

3. pyproject.toml

[project]
name = "edinet"
version = "0.1.0"
description = "MCP financial data from edinet"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.2.0",
    "requests",
    "pandas",
]

[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"

[project.scripts]
edinet = "edinet:main"

このファイルはプロジェクトの依存関係を定義しています。MCPクライアント、requests、pandasが主な依存パッケージです。

使い方

Dockerイメージのビルド

# イメージをビルド
docker build -t edinet-mcp-server .

MCPサーバーの設定

以下のように設定を書きます。

"edinet": {
      "autoApprove": [],
      "disabled": true,
      "timeout": 60,
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-e",
        "API_KEY",
        "edinet-mcp-server"
      ],
      "env": {
        "API_KEY": "APIキー"
      },
      "transportType": "stdio"
    },

Dockerでコンテナ起動していろいろやりたい場合は、

# コンテナを実行（APIキーを環境変数として渡す）
docker run -e API_KEY=あなたのEDINET_APIキー edinet-mcp-server

MCPサーバーの利用方法

このMCPサーバーをClaudeなどのAIと連携させると、以下のような質問に対応できるようになります：

  • 「edinetで2023年1月から2023年3月までの書類一覧（doc_id）を取得してください」
  • 「[doc_id]の財務情報を抽出してください」

発展的な使い方

  1. データ分析パイプラインの構築：取得したデータを自動的に分析し、レポートを生成するパイプライン
  2. 定期的なデータ収集：cronなどを使って定期的にデータを収集・蓄積
  3. 他のデータソースとの連携：株価データなどと組み合わせた総合的な分析
タイトルとURLをコピーしました