ツール実行

# python/llm-service/llm_service/tools/base.py
class ToolMetadata:
    name: str                    # 一意の識別子
    version: str                 # セマンティックバージョン
    description: str             # LLM可視の説明
    category: str                # 検索、データ、分析、コードなど
    parameters: List[ToolParameter]  # JSONスキーマ

    # 実行制約
    timeout_seconds: int = 30
    memory_limit_mb: int = 128
    rate_limit: int = 60        # 分あたりのリクエスト数
    cost_per_use: float = 0.0   # USD

    # セキュリティフラグ
    dangerous: bool = False      # ファイル書き込み、コード実行
    requires_auth: bool = False
    sandboxed: bool = False
    session_aware: bool = False

# すべてのツールをリスト
GET /tools/list

# ツールスキーマを取得
GET /tools/{name}/schema

# ツールメタデータを取得
GET /tools/{name}/metadata

# カテゴリ別にリスト
GET /tools/list?category=data

# Python WASIインタープリタをダウンロード (20MB)
./scripts/setup_python_wasi.sh

# インストールを確認
ls -lh wasm-interpreters/python-3.11.4.wasm

# 設定
PYTHON_WASI_WASM_PATH=./wasm-interpreters/python-3.11.4.wasm
WASI_MEMORY_LIMIT_MB=512
WASI_TIMEOUT_SECONDS=60

# ローカル $ref 参照を解決
# #/components/schemas/Pet → Pet スキーマ定義
def resolve_ref(ref: str, spec: dict) -> dict:
    path = ref.lstrip("#/").split("/")
    schema = spec
    for key in path:
        schema = schema[key]
    return schema

# python/llm-service/llm_service/tools/vendor_adapters/__init__.py
def get_vendor_adapter(name: str):
    """名前によってアダプターインスタンスを返す。利用できない場合は None を返す。"""
    if not name:
        return None

    try:
        if name.lower() == "datainsight":
            from .datainsight import DataInsightAdapter
            return DataInsightAdapter()
        # ベンダーモジュールが欠落している場合の優雅なフォールバック
    except ImportError:
        logger.warning(f"ベンダーアダプター '{name}' は利用できません")
        return None
    except Exception as e:
        logger.error(f"ベンダーアダプター '{name}' の読み込みに失敗しました: {e}")
        return None

    return None

# 変換: "users" → "di:unique_users"
metric_aliases = {
    "users": "di:unique_users",
    "sessions": "di:total_sessions",
}
body["metrics"] = [metric_aliases.get(m, m) for m in body["metrics"]]

# prompt_params から account_id を注入
if prompt_params and "account_id" in prompt_params:
    body["account_id"] = prompt_params["account_id"]

# 変換: {start, end} → {startTime, endTime}
if "start" in body["timeRange"]:
    body["timeRange"]["startTime"] = body["timeRange"].pop("start")
if "end" in body["timeRange"]:
    body["timeRange"]["endTime"] = body["timeRange"].pop("end")

class ToolMetadata:
    rate_limit: int = 60  # 分あたりのリクエスト数（デフォルト）

class RateLimiter:
    def __init__(self, rate_limit: int):
        self.rate = rate_limit
        self.tokens = rate_limit
        self.last_update = time.time()

    def allow(self) -> bool:
        now = time.time()
        elapsed = now - self.last_update

        # トークンを補充
        self.tokens = min(
            self.rate,
            self.tokens + elapsed * (self.rate / 60.0)
        )
        self.last_update = now

        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

# 環境変数
MCP_RATE_LIMIT_DEFAULT=60              # MCP ツールのデフォルト
WEB_SEARCH_RATE_LIMIT=120              # ウェブ検索特有
CALCULATOR_RATE_LIMIT=2000             # 計算機特有
PYTHON_EXECUTOR_RATE_LIMIT=60          # Python 実行

# MCP サーキットブレーカー
MCP_CB_FAILURES=5                      # 5回の失敗後にオープン
MCP_CB_RECOVERY_SECONDS=60             # 60秒間オープンのまま

# Agent Core の強制
ENFORCE_CB_ERROR_THRESHOLD=0.5         # オープンするエラー率 (50%)
ENFORCE_CB_WINDOW_SECONDS=30           # スライディングウィンドウ
ENFORCE_CB_MIN_REQUESTS=20             # アクティベーション前の最小リクエスト数

class ToolMetadata:
    cost_per_use: float = 0.001  # 呼び出しあたりの USD

// go/orchestrator/internal/budget/manager.go
type BudgetManager struct {
    maxCostPerRequest float64
    maxTokensPerRequest int
}

func (bm *BudgetManager) CheckBudget(currentCost, currentTokens float64) error {
    if currentCost >= bm.maxCostPerRequest {
        return ErrBudgetExceeded
    }
    if currentTokens >= bm.maxTokensPerRequest {
        return ErrTokenLimitExceeded
    }
    return nil
}

MAX_COST_PER_REQUEST=0.50              # USD
MAX_TOKENS_PER_REQUEST=10000
TOKEN_BUDGET_PER_AGENT=5000
TOKEN_BUDGET_PER_TASK=20000

# ジッターを伴う指数バックオフ
@retry(
    retries=3,
    backoff=exponential_backoff(base=1.0, max_delay=10.0),
    on_exception=(NetworkError, TimeoutError)
)
def execute_tool(tool_name, parameters):
    return registry.execute(tool_name, parameters)

# 外部API用のHTTPコネクションプール
session = aiohttp.ClientSession(
    connector=aiohttp.TCPConnector(
        limit=100,              # 最大接続数
        limit_per_host=10,      # ホストごとの最大接続数
        ttl_dns_cache=300,      # DNSキャッシュのTTL
    )
)

# OpenAPI仕様をキャッシュ
@lru_cache(maxsize=100)
def load_openapi_spec(spec_url: str) -> dict:
    return fetch_and_parse(spec_url)

// 複数のツールを並列に実行
func ExecuteToolsParallel(tools []Tool, parallelism int) []Result {
    sem := make(chan struct{}, parallelism)  // セマフォ
    results := make([]Result, len(tools))

    var wg sync.WaitGroup
    for i, tool := range tools {
        wg.Add(1)
        go func(idx int, t Tool) {
            defer wg.Done()
            sem <- struct{}{}        // 獲得
            defer func() { <-sem }() // 解放

            results[idx] = t.Execute()
        }(i, tool)
    }
    wg.Wait()
    return results
}

TOOL_PARALLELISM=5  # 最大5つのツールを同時に実行

# 外部APIアクセスを制限
MCP_ALLOWED_DOMAINS=api.example.com,api.partner.com
OPENAPI_ALLOWED_DOMAINS=api.github.com,api.stripe.com

# 開発専用
MCP_ALLOWED_DOMAINS=*  # すべてを許可 (⚠️ 本番環境では使用しない)

class ToolMetadata:
    dangerous: bool = False      # ファイル書き込み、コード実行
    requires_auth: bool = False  # ユーザー認証が必要

# config/opa/policies/tools.rego
package tools

deny[msg] {
    input.tool == "file_write"
    not is_admin(input.user)
    msg := "file_writeには管理者ロールが必要です"
}

# ツール実行メトリクス
shannon_tool_executions_total{tool_name, status}
shannon_tool_execution_duration_seconds{tool_name}
shannon_tool_errors_total{tool_name, error_type}
shannon_tool_cost_usd{tool_name}

# サーキットブレーカーのメトリクス
shannon_circuit_breaker_state{base_url}
shannon_circuit_breaker_failures_total{base_url}

# レートリミッターのメトリクス
shannon_rate_limit_exceeded_total{tool_name}
shannon_rate_limit_tokens_available{tool_name}

# ツール実行のための構造化ロギング
logger.info(
    "ツール実行が完了しました",
    extra={
        "tool_name": tool_name,
        "duration_ms": duration,
        "cost_usd": cost,
        "success": success,
        "error": error_message,
    }
)

# OpenTelemetryトレーシング
with tracer.start_as_current_span("tool_execution") as span:
    span.set_attribute("tool.name", tool_name)
    span.set_attribute("tool.category", category)

    result = execute_tool(tool_name, parameters)

    span.set_attribute("tool.success", result.success)
    span.set_attribute("tool.cost_usd", result.cost)