"""MCP server exposing EDINET tools to LLMs via FastMCP.
This module defines the MCP (Model Context Protocol) server that allows
AI assistants like Claude to search Japanese companies, retrieve financial
filings, and parse XBRL financial statements through EDINET.
Usage with Claude Desktop (add to ``claude_desktop_config.json``)::
{
"mcpServers": {
"edinet": {
"command": "uvx",
"args": ["edinet-mcp", "serve"]
}
}
}
"""
from __future__ import annotations
import asyncio
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Annotated, Any, cast
from pydantic import BeforeValidator, Field
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from edinet_mcp.models import StatementType
from fastmcp import FastMCP
from edinet_mcp._diff import diff_statements as _diff_statements
from edinet_mcp._metrics import calculate_metrics, compare_periods
from edinet_mcp._normalize import get_taxonomy_labels
from edinet_mcp._screening import screen_companies as _screen_companies
from edinet_mcp.client import EdinetClient
def _coerce_str(v: Any) -> str | None:
"""Coerce int to str for period parameters (MCP clients may send int)."""
if v is None:
return None
return str(v)
# Period type that accepts both str and int from MCP clients
CoercedStr = Annotated[str | None, BeforeValidator(_coerce_str)]
# Lazily initialized client with lock for concurrent-safe access
_client: EdinetClient | None = None
_client_lock = asyncio.Lock()
async def _get_client() -> EdinetClient:
"""Return the shared EdinetClient, creating it on first call.
Uses double-checked locking to avoid race conditions when
multiple MCP tool calls arrive concurrently.
"""
global _client
if _client is not None:
return _client
async with _client_lock:
if _client is None:
_client = EdinetClient()
return _client
@asynccontextmanager
async def _lifespan(server: FastMCP[dict[str, Any]]) -> AsyncIterator[dict[str, Any]]:
"""Manage EdinetClient lifecycle — close httpx.AsyncClient on shutdown."""
yield {}
if _client is not None:
await _client.close()
mcp = FastMCP(
name="EDINET",
lifespan=_lifespan,
instructions=(
"EDINET MCP server provides tools for accessing Japanese financial "
"disclosure data. You can search for companies listed on the Tokyo "
"Stock Exchange, retrieve financial filings (有価証券報告書, 四半期報告書), "
"and parse XBRL financial statements (BS, PL, CF) into structured data.\n\n"
"All financial data comes from EDINET, the Electronic Disclosure for "
"Investors' NETwork operated by Japan's Financial Services Agency (FSA).\n\n"
"Key tools:\n"
"- search_companies: Find companies by name/ticker/code\n"
"- get_financial_statements: Get normalized BS/PL/CF data\n"
"- get_financial_metrics: Get calculated ratios (ROE, ROA, margins)\n"
"- compare_financial_periods: Get year-over-year changes\n"
"- screen_companies: Compare metrics across multiple companies\n"
"- list_available_labels: See which labels are available\n\n"
"IMPORTANT: The 'period' parameter is the FILING year, not fiscal year. "
"Japanese companies with March fiscal year-end file annual reports in "
"June of the following year (e.g., FY2024 → filed 2025 → period='2025')."
),
)
@mcp.tool()
async def search_companies(
query: Annotated[
str,
Field(description="企業名(日本語 or 英語)、証券コード、またはEDINETコードで検索"),
],
) -> list[dict[str, Any]]:
"""Search for Japanese companies registered in EDINET.
Examples:
- search_companies("トヨタ") → Toyota Motor Corporation
- search_companies("7203") → Toyota (by ticker)
- search_companies("E02144") → Toyota (by EDINET code)
"""
client = await _get_client()
companies = await client.search_companies(query)
return [c.model_dump() for c in companies[:20]]
@mcp.tool()
async def get_filings(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
start_date: Annotated[
str,
Field(description="検索開始日 (YYYY-MM-DD形式)"),
],
end_date: Annotated[
str | None,
Field(description="検索終了日 (YYYY-MM-DD形式、省略時は今日)"),
] = None,
doc_type: Annotated[
str | None,
Field(
description=(
"書類タイプでフィルタ: 'annual_report' (有価証券報告書), "
"'quarterly_report' (四半期報告書), または省略で全件"
)
),
] = None,
) -> list[dict[str, Any]]:
"""List financial filings for a company within a date range.
Returns filing metadata including doc_id, filing date, and document type.
Use the doc_id with get_financial_statements to retrieve actual data.
"""
client = await _get_client()
filings = await client.get_filings(
start_date=start_date,
end_date=end_date,
edinet_code=edinet_code,
doc_type=doc_type,
)
return [f.model_dump(mode="json") for f in filings[:50]]
@mcp.tool()
async def get_financial_statements(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
period: Annotated[
CoercedStr,
Field(
description=(
"書類が提出された年 (例: '2024')。"
"3月決算企業のFY2024報告書は2025年提出のため period='2025' を指定。"
"省略時は直近1年間の最新報告書を取得"
)
),
] = None,
doc_type: Annotated[
str,
Field(description="'annual_report' (有価証券報告書) or 'quarterly_report' (四半期報告書)"),
] = "annual_report",
) -> dict[str, Any]:
"""Retrieve and parse financial statements (BS, PL, CF) for a company.
Returns normalized financial data with Japanese labels.
Each line item has 当期 (current) and 前期 (previous) values.
Example response:
income_statement: [{"科目": "売上高", "当期": 45095325, "前期": 37154298}, ...]
"""
client = await _get_client()
stmt = await client.get_financial_statements(
edinet_code=edinet_code,
doc_type=doc_type,
period=period,
)
result: dict[str, Any] = {
"filing": stmt.filing.model_dump(mode="json"),
"accounting_standard": stmt.accounting_standard.value,
}
for name, data in stmt.all_statements.items():
result[name] = data.to_dicts()
return result
@mcp.tool()
async def get_financial_metrics(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
period: Annotated[
CoercedStr,
Field(
description=(
"書類が提出された年 (例: '2024')。"
"3月決算企業のFY2024報告書は2025年提出のため period='2025' を指定。"
"省略時は最新"
)
),
] = None,
doc_type: Annotated[
str,
Field(description="'annual_report' or 'quarterly_report'"),
] = "annual_report",
) -> dict[str, Any]:
"""Calculate key financial metrics (ROE, ROA, profit margins, etc.).
Returns profitability ratios, stability ratios, and cash flow summary
computed from the company's financial statements.
Example: get_financial_metrics("E02144") → {
"profitability": {"営業利益率": "11.87%", "ROE": "12.50%", ...},
"stability": {"自己資本比率": "41.60%", ...},
"cash_flow": {"営業CF": 5000000, "フリーCF": 3000000, ...}
}
"""
client = await _get_client()
stmt = await client.get_financial_statements(
edinet_code=edinet_code,
doc_type=doc_type,
period=period,
)
metrics = calculate_metrics(stmt)
result: dict[str, Any] = dict(metrics)
result["filing"] = {
"doc_id": stmt.filing.doc_id,
"company_name": stmt.filing.company_name,
"period_end": stmt.filing.period_end.isoformat() if stmt.filing.period_end else None,
}
result["accounting_standard"] = stmt.accounting_standard.value
return result
@mcp.tool()
async def compare_financial_periods(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
period: Annotated[
CoercedStr,
Field(
description=(
"書類が提出された年 (例: '2024')。"
"3月決算企業のFY2024報告書は2025年提出のため period='2025' を指定。"
"省略時は最新"
)
),
] = None,
doc_type: Annotated[
str,
Field(description="'annual_report' or 'quarterly_report'"),
] = "annual_report",
) -> dict[str, Any]:
"""Compare financial data between current and previous periods.
Returns year-over-year changes (増減額 and 増減率) for all items
that have both 当期 and 前期 data.
Example: compare_financial_periods("E02144") → {
"changes": [
{"科目": "売上高", "当期": 45095325, "前期": 37154298, "増減率": "+21.38%"},
...
]
}
"""
client = await _get_client()
stmt = await client.get_financial_statements(
edinet_code=edinet_code,
doc_type=doc_type,
period=period,
)
changes = compare_periods(stmt)
return {
"filing": {
"doc_id": stmt.filing.doc_id,
"company_name": stmt.filing.company_name,
"period_end": stmt.filing.period_end.isoformat() if stmt.filing.period_end else None,
},
"accounting_standard": stmt.accounting_standard.value,
"changes": changes,
}
@mcp.tool()
async def list_available_labels(
statement_type: Annotated[
str,
Field(
description=(
"財務諸表の種類: 'income_statement' (PL), 'balance_sheet' (BS), 'cash_flow' (CF)"
)
),
] = "income_statement",
) -> list[dict[str, str]]:
"""List all available financial line item labels for a statement type.
Use this to discover which labels (科目) are available when
accessing financial data via get_financial_statements.
Returns labels in display order with Japanese and English names.
"""
valid_types = ("income_statement", "balance_sheet", "cash_flow", "cash_flow_statement")
if statement_type not in valid_types:
msg = f"Invalid statement_type: {statement_type!r}. Must be one of {valid_types}"
raise ValueError(msg)
return get_taxonomy_labels(cast("StatementType", statement_type))
@mcp.tool()
async def get_company_info(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
) -> dict[str, Any]:
"""Get detailed information about a company by EDINET code."""
client = await _get_client()
company = await client.get_company(edinet_code)
return company.model_dump()
@mcp.tool()
async def screen_companies(
edinet_codes: Annotated[
list[str],
Field(
description=(
"比較対象の企業EDINETコードのリスト (最大20社)。例: ['E02144', 'E01777', 'E02529']"
)
),
],
period: Annotated[
CoercedStr,
Field(description=("書類が提出された年 (例: '2025')。省略時は最新。")),
] = None,
doc_type: Annotated[
str,
Field(description="'annual_report' or 'quarterly_report'"),
] = "annual_report",
sort_by: Annotated[
str | None,
Field(
description=(
"ソート基準の指標名 (例: 'ROE', '営業利益率', '売上高')。省略時はソートなし"
)
),
] = None,
) -> dict[str, Any]:
"""Compare financial metrics across multiple companies.
Fetches financial statements for each company, calculates key metrics
(ROE, ROA, profit margins, etc.), and returns a comparison table.
Useful for sector analysis, portfolio comparison, and screening.
Maximum 20 companies per request (rate limit constraint).
Example: screen_companies(["E02144", "E01777"]) → {
"results": [
{"edinet_code": "E02144", "company_name": "トヨタ...", "profitability": {...}, ...},
{"edinet_code": "E01777", "company_name": "ソニー...", "profitability": {...}, ...}
],
"errors": [],
"count": 2
}
"""
client = await _get_client()
return await _screen_companies(
client,
edinet_codes,
period=period,
doc_type=doc_type,
sort_by=sort_by,
)
@mcp.tool()
async def diff_financial_statements(
edinet_code: Annotated[
str,
Field(description="企業のEDINETコード (例: 'E02144')"),
],
period1: Annotated[
Annotated[str, BeforeValidator(_coerce_str)],
Field(description="比較元の期間 (例: '2024')"),
],
period2: Annotated[
Annotated[str, BeforeValidator(_coerce_str)],
Field(description="比較先の期間 (例: '2025')"),
],
doc_type: Annotated[
str,
Field(description="'annual_report' (有価証券報告書) or 'quarterly_report' (四半期報告書)"),
] = "annual_report",
) -> dict[str, Any]:
"""Compare financial statements for the same company across two periods.
Calculates changes (増減額) and growth rates (増減率) for each line item
in the income statement, balance sheet, and cash flow statement.
Useful for analyzing year-over-year performance changes.
Example: diff_financial_statements("E02144", "2024", "2025") → {
"edinet_code": "E02144",
"company_name": "トヨタ自動車株式会社",
"period1": "2024",
"period2": "2025",
"diffs": [
{"statement": "income_statement", "科目": "売上高",
"増減額": 7941025000000, "増減率": "+21.38%"},
...
],
"summary": {"total_items": 45, "increased": 30, "decreased": 12, ...}
}
"""
client = await _get_client()
result = await _diff_statements(
client,
edinet_code=edinet_code,
period1=period1,
period2=period2,
doc_type=doc_type,
)
return {
"edinet_code": result["edinet_code"],
"company_name": result["company_name"],
"period1": result["period1"],
"period2": result["period2"],
"accounting_standard": result["accounting_standard"],
"diffs": result["diffs"][:50], # Limit output size for MCP
"summary": result["summary"],
}
