📑 技術調査 - LLMエージェント時代の上場企業データインフラ設計

[方法論の射程]

  データ契約 4 原則          ← 安定。規制業界 API の延長線上
       ↓ 適用
  シリアライズ・届け方       ← 流動。MCP / REST / code execution が競合
       ↓ 選択
  LLM エージェント           ← 消費者

エージェント呼び出し
      │
      ├─ 社外配布 + ツール数安定 → MCP Server（契約 4 原則を outputSchema に反映）
      │
      ├─ 社内専用 / トークン効率重視 → Code Execution + Skill Loading
      │    （コミュニティ討議で 150K → 2K tokens 程度の削減が報告された例あり）
      │
      └─ 既存 REST が安定している → REST API + 応答フィールドに 4 原則を追加

{
  "company": "7203",
  "net_sales": 45095000000000,
  "period": "2025-03"
}

{
  "company": "7203",
  "net_sales": 45095000000000,
  "unit": "JPY",
  "as_of": "2025-03-31T23:59:59Z",
  "null_reason": null,
  "definition_ref": "https://edinet.fsa.go.jp/taxonomies/jpcrp_cor/NetSalesSummaryOfBusinessResults",
  "source": {
    "primary_evidence": "edinet://docID=S100RABCD#jpcrp_cor:NetSalesSummaryOfBusinessResults",
    "retrieved_at": "2026-05-11T08:00:00Z"
  }
}

from __future__ import annotations
from pydantic import BaseModel, Field, model_validator
from typing import Optional, Literal
from datetime import datetime

class SourceInfo(BaseModel):
    """データ出典メタデータ"""
    system: str = Field(
        description="取得元システム識別子 (例: 'snowflake://analytics.prod.fact_financials')"
    )
    doc_id: Optional[str] = Field(
        None,
        description="原票の一意識別子 (EDINET docID / SEC accession number 等)"
    )
    primary_evidence: Optional[str] = Field(
        None,
        description="原票への直リンク URL (EDINET / SEC EDGAR 等)"
    )
    retrieved_at: datetime = Field(
        description="取得時刻 (ISO 8601)"
    )

class FinancialMetricResponse(BaseModel):
    """財務指標応答の契約スキーマ (LLM エージェント向け)

    null_reason の扱い:
      - value != null の場合: null_reason は None（フィールドは常に存在する）
      - value == null の場合: null_reason は enum 必須（None は不可）
    """
    metric: str = Field(
        description=(
            "指標の canonical 名（定義を含める命名推奨: 例 'roe_consolidated'）。"
            "dbt Semantic Layer 等で単一定義されている識別子を使う"
        )
    )
    value: Optional[float] = Field(
        None,
        description="指標値。データ取得不能な場合は null を返す (sentinel value 禁止)"
    )
    unit: str = Field(
        description="単位 (例: 'JPY', '%', 'shares')"
    )
    as_of: datetime = Field(
        description="データの観測時点 (ISO 8601)。FRED の observation_date に相当"
    )
    null_reason: Optional[Literal[
        "not_disclosed",
        "definition_changed",
        "out_of_scope",
        "source_unavailable"
    ]] = Field(
        None,
        description=(
            "フィールドは常に存在する。"
            "value != null のときは None。"
            "value == null のときは enum 必須（省略不可）。"
            "エージェントへの補完禁止シグナル"
        )
    )
    definition_ref: Optional[str] = Field(
        None,
        description="指標定義 URL。エージェントが定義を照合するためのアンカー"
    )
    compiled_sql: Optional[str] = Field(
        None,
        description="計算に使用した SQL (dbt-mcp get_metrics_compiled_sql 方式)"
    )
    source: SourceInfo
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="値の信頼度スコア (Vertex AI groundingSupports.confidenceScores に相当)"
    )

    @model_validator(mode="after")
    def null_reason_required_when_value_is_null(self) -> "FinancialMetricResponse":
        """value が null のとき null_reason は必須"""
        if self.value is None and self.null_reason is None:
            raise ValueError(
                "null_reason は value=null のとき必須です。"
                "not_disclosed / definition_changed / out_of_scope / source_unavailable のいずれかを指定してください。"
            )
        return self

import json
print(json.dumps(FinancialMetricResponse.model_json_schema(), indent=2))
# → MCP Tool の outputSchema にそのまま渡せる

uvx dbt-mcp

{
  "mcpServers": {
    "dbt": {
      "command": "uvx",
      "args": ["dbt-mcp"],
      "env": {
        "DBT_HOST": "your-dbt-cloud-host",
        "DBT_TOKEN": "${DBT_SERVICE_TOKEN}",
        "DBT_ENVIRONMENT_ID": "12345"
      }
    }
  }
}

# 1. メトリック一覧を取得 (エージェントが最初に呼ぶ)
#    Tool: list_metrics
#    → MetricToolResponse[] { name, type, label, description, metadata }

# 2. 特定メトリックのコンパイル済み SQL を取得
#    Tool: get_metrics_compiled_sql
#    Input: { metrics: ["net_sales"], group_by: ["company_id", "fiscal_year"] }
#    → { sql: "SELECT company_id, fiscal_year, SUM(...) FROM ..." }

# 3. メトリックの次元を確認
#    Tool: get_dimensions
#    Input: { metrics: ["net_sales"] }
#    → ["company_id", "fiscal_year", "segment", ...]

# 4. 実際のクエリ実行 (フィルタ付き)
#    Tool: query_metrics
#    Input: {
#      metrics: ["net_sales"],
#      group_by: ["company_id"],
#      where: "{{ Entity('company_id') }} = '7203'",
#      order_by: ["-net_sales"]
#    }

pip install fastmcp pydantic

from fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime

mcp = FastMCP(
    name="financial-data-mcp",
    instructions=(
        "上場企業の財務データを返す MCP サーバ。"
        "値が取得できない場合は value=null + null_reason を返す。"
        "エージェントは null_reason があるフィールドを補完してはならない。"
    )
)

class GetFinancialInput(BaseModel):
    company_id: str = Field(
        description="企業識別子 (EDINET 証券コード or CIK)"
    )
    metric: Literal["net_sales", "operating_income", "total_assets"] = Field(
        description="取得する財務指標名"
    )
    fiscal_year: int = Field(
        ge=2000, le=2030,
        description="会計年度 (例: 2025 = 2025年3月期)"
    )

@mcp.tool(
    name="get_financial_metric",
    description=(
        "上場企業の単一財務指標を取得する。"
        "返値は契約スキーマ準拠: value (null 可) + null_reason + source.primary_evidence + definition_ref + compiled_sql を含む。"
        "null_reason が非 null の場合、エージェントは推測による補完を禁止する。"
        "boundary: 四半期データは含まない。連結財務のみ。"
    )
)
def get_financial_metric(params: GetFinancialInput) -> FinancialMetricResponse:
    try:
        value, sql = _fetch_from_dbt_semantic_layer(
            company_id=params.company_id,
            metric=params.metric,
            fiscal_year=params.fiscal_year
        )
        return FinancialMetricResponse(
            metric=params.metric,
            value=value,
            unit="JPY",
            as_of=datetime(params.fiscal_year, 3, 31),
            null_reason=None,
            definition_ref=f"dbt://semantic_models/financials.yml#{params.metric}",
            compiled_sql=sql,
            source=SourceInfo(
                system="snowflake://analytics.prod.fact_financials",
                primary_evidence=_get_edinet_doc_id(params.company_id, params.fiscal_year),
                retrieved_at=datetime.utcnow()
            ),
            confidence=0.98
        )
    except DataUnavailableError as e:
        return FinancialMetricResponse(
            metric=params.metric,
            value=None,
            unit="JPY",
            as_of=datetime(params.fiscal_year, 3, 31),
            null_reason="source_unavailable",
            source=SourceInfo(
                system="snowflake://analytics.prod.fact_financials",
                retrieved_at=datetime.utcnow()
            ),
            confidence=0.0
        )

if __name__ == "__main__":
    mcp.run()

import anthropic

client = anthropic.Anthropic()

response = client.beta.messages.create(
    model="claude-opus-4-5",
    max_tokens=4096,
    betas=[
        "code-execution-2025-08-25",
        "skills-2025-10-02",
        "files-api-2025-04-14"
    ],
    container={
        "type": "persistent",
        "skill_ids": ["financial-data-fetcher"]
    },
    tools=[{"type": "code_execution_20250825"}],
    messages=[{
        "role": "user",
        "content": "トヨタ自動車 (7203) の FY2025 売上高を EDINET から取得してください"
    }]
)

あなたは上場企業の財務データを分析するアナリストエージェントです。

## データ取得ルール
1. 財務指標は必ず get_financial_metric ツールを使って取得すること
2. ツール応答に `null_reason` が含まれる場合:
   - そのフィールドを推測・補完してはならない
   - ユーザに「{metric} は {null_reason} のため取得できませんでした」と正確に伝える
   - 代替データソースを提案する場合は、そのソースも同様に出典を明示する
3. `definition_ref` が存在する場合:
   - 応答に指標の定義 URL を引用する
   - 「{指標名}」の定義は {definition_ref} に基づく、と明示する
4. `compiled_sql` が存在する場合:
   - 数値の根拠として SQL ロジックをユーザに開示できる
   - 再現性が必要なユーザには SQL を提示する

## ハルシネーション防止
- ツール外から財務数値を生成することを厳禁とする
- "approximately", "roughly" などの曖昧表現で数値を出力しない
- 不確かな場合は「データが確認できません。ツールを再試行してください」と返す

tool_result = call_tool("get_financial_metric", {
    "company_id": "7203",
    "metric": "net_sales",
    "fiscal_year": 2025
})

if tool_result["null_reason"] is not None:
    return f"""
net_sales は取得できませんでした。
理由: {tool_result['null_reason']}
定義参照: {tool_result.get('definition_ref', 'N/A')}
代替手段: EDINET API (https://disclosure2dl.edinet-fsa.go.jp/) で直接検索することを推奨します。
"""
else:
    return f"""
{tool_result['metric']}: {tool_result['value']:,} {tool_result['unit']}
観測時点: {tool_result['as_of']}
定義: {tool_result['definition_ref']}
出典: {tool_result['source']['primary_evidence']}
"""

from pydantic import BaseModel, Field
from typing import Optional

class BulkFinancialRequest(BaseModel):
    company_id: str = Field(
        description="企業識別子 (証券コード)"
    )
    metrics: list[Literal["net_sales", "operating_income", "total_assets", "net_income"]] = Field(
        description="取得する指標リスト (最大 10 指標)"
    )
    fiscal_years: list[int] = Field(
        description="取得する会計年度リスト (最大 5 年)",
        min_length=1,
        max_length=5
    )

class BulkFinancialResponse(BaseModel):
    company_id: str
    results: list[FinancialMetricResponse]
    partial_failure: bool = Field(
        description="一部取得失敗がある場合 True。results 内の null_reason を確認"
    )

@mcp.tool(
    name="get_financials_bulk",
    description=(
        "上場企業の財務指標を複数年・複数指標で一括取得する。"
        "単発呼び出しの繰り返しを避け、ツール呼び出し回数を削減する。"
        "取得できない指標は value=null + null_reason で返す (補完禁止)。"
        "boundary: 最大 10 指標 × 5 年 = 50 行。超過する場合は分割して呼ぶ。"
    )
)
def get_financials_bulk(params: BulkFinancialRequest) -> BulkFinancialResponse:
    results = []
    has_failure = False
    for year in params.fiscal_years:
        for metric in params.metrics:
            result = _fetch_single(params.company_id, metric, year)
            if result.null_reason is not None:
                has_failure = True
            results.append(result)
    return BulkFinancialResponse(
        company_id=params.company_id,
        results=results,
        partial_failure=has_failure
    )

# 3 年分 × 3 指標 = 9 データポイントを 1 ツール呼び出しで取得
result = call_tool("get_financials_bulk", {
    "company_id": "7203",
    "metrics": ["net_sales", "operating_income", "net_income"],
    "fiscal_years": [2023, 2024, 2025]
})
# → results[9], partial_failure: false

import duckdb

def verify_metric_response(response: FinancialMetricResponse) -> dict:
    if response.compiled_sql is None:
        return {"verified": False, "reason": "compiled_sql not provided"}

    conn = duckdb.connect()
    result = conn.execute(response.compiled_sql).fetchone()
    recalculated_value = result[0] if result else None

    if recalculated_value is None:
        return {"verified": False, "reason": "SQL returned no rows"}

    tolerance = 0.0001
    delta = abs(recalculated_value - response.value) / abs(response.value)

    if delta > tolerance:
        return {
            "verified": False,
            "reported": response.value,
            "recalculated": recalculated_value,
            "delta_pct": delta * 100,
            "action": "数値差異を検出。source.primary_evidence を直接参照することを推奨"
        }

    return {"verified": True, "delta_pct": delta * 100}

User Query
    │
    ▼
[1] dbt-mcp: list_metrics + get_metrics_compiled_sql
    │  → metric 定義・計算 SQL を取得
    ▼
[2] MCP Tool: get_financials_bulk
    │  → value + null_reason + compiled_sql + source.primary_evidence
    ▼
[3] Vertex AI Grounding (オプション)
    │  → groundingChunks (EDINET URL) + groundingSupports (テキスト位置 + confidence)
    ▼
[4] enrich_with_grounding()
    │  → MCP response + grounding evidence を統合
    ▼
[5] エージェント応答生成
       「FY2025 売上高: 45,095,000 百万円
        出典: EDINET S100RABCD (信頼度 94%)
        定義: dbt://semantic_models/financials.yml#net_sales
        SQL: SELECT SUM(net_sales_amount) FROM ...」

null_reason_coverage_rate = (null_reason が付与された null 件数) / (value == null の全件数)

definition_ref ヒット率 = (HTTP 200 が返ったリクエスト数) / (definition_ref を含む全応答数)

hallucination 率 = (検算不一致の応答件数) / (検算対象の全応答件数)

"contract": {
  "id": "monthly_revenue",
  "version": "3.1.0",
  "ref": "https://contracts.example.com/monthly_revenue/v3.1.0"
}

import hashlib, json

def check_tool_integrity(tool_list: list[dict], stored_hash: str) -> bool:
    current = hashlib.sha256(
        json.dumps(tool_list, sort_keys=True).encode()
    ).hexdigest()
    if current != stored_hash:
        audit_log.warning("tool definition changed", hash=current, prev=stored_hash)
        return False
    return True

POST /token HTTP/1.1
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code
&code=...
&resource=https://data-api.example.com/mcp