0Pricing
AI Agents · レッスン

エージェントアクションへのポリシー適用

実行前のポリシーチェック、許可リスト・拒否リスト、動的なポリシールールを扱います。

「エージェントアクションへのポリシー適用」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

エージェントのポリシー適用とは

ポリシー適用とは、すべてのエージェントアクションの前に実行され、そのアクションが許可されているかどうかを判断する実行時ゲートです。これがなければ、エージェントを制約するものはLLMの指示追従だけになり、それは回避されたり誤解釈されたりする可能性があります。

適用処理はLLMの外側、つまり利用するインフラストラクチャ内に置く必要があります。

アクション前チェックパターン

ツールを実行する前に、can_agent_do(action, context)を呼び出します。この関数を唯一の適用ポイントとし、アクション実行に至るすべての経路をここに通します。

def can_agent_do(action: str, context: dict) -> tuple[bool, str]:
    '''
    Returns (allowed: bool, reason: str).
    Context includes: user_id, agent_id, session_id, parameters, timestamp.
    '''
    # 1. Check denylist first (fast path for obvious violations)
    if action in DENIED_ACTIONS:
        return False, f'Action "{action}" is on the global denylist'

    # 2. Check allowlist
    if action not in ALLOWED_ACTIONS:
        return False, f'Action "{action}" is not on the allowlist'

    # 3. Context-specific checks
    return check_context_policy(action, context)

AllowlistとDenylistの定義

allowlistには、エージェントが実行を許可されているすべてのアクションを列挙します。リストにないものは、デフォルトでブロックされます。denylistは、コンテキストに関係なく決して許可してはならないアクションに対する追加の安全網です。

# Allowlist: tools the agent can use
ALLOWED_ACTIONS = {
    'web_search',
    'read_file',
    'write_file',
    'send_email',
    'create_calendar_event',
    'query_database',
    'execute_python_sandbox',
    'fetch_url',
    'create_ticket'
}

# Denylist: actions that are always blocked, regardless of context
DENIED_ACTIONS = {
    'delete_all_records',
    'export_entire_database',
    'send_mass_email',
    'modify_system_config',
    'create_admin_user',
    'disable_audit_logging'
}

if __name__ == '__main__':
    for action in ('web_search', 'send_mass_email'):
        print(f"{action}: allowed={action in ALLOWED_ACTIONS} denied={action in DENIED_ACTIONS}")

コンテキスト固有のポリシーチェック

単純な許可・拒否リストだけでなく、ポリシーはコンテキストに依存することがよくあります。ユーザーは誰か、役割は何か、現在の時刻はいつか、対象リソースは何か、といった情報です。これらがコンテキスト固有のチェックです。

from datetime import datetime, timezone

def check_context_policy(action: str, context: dict) -> tuple[bool, str]:
    user_id   = context.get('user_id', '')
    params    = context.get('parameters', {})
    user_role = context.get('user_role', 'user')

    # send_email: only agents with email_sender role
    if action == 'send_email':
        if user_role not in ('email_agent', 'admin'):
            return False, f'Role "{user_role}" cannot send emails'
        recipient = params.get('to', '')
        if not recipient.endswith('@trusted-domain.com'):
            return False, 'Email recipient must be in @trusted-domain.com'

    # write_file: path restrictions
    if action == 'write_file':
        path = params.get('path', '')
        if not path.startswith('/tmp/') and not path.startswith('/workspace/'):
            return False, f'File writes outside /tmp/ and /workspace/ are not allowed'

    return True, 'Permitted'

if __name__ == '__main__':
    ctx = {'user_id': 'u1', 'user_role': 'user',
           'parameters': {'to': 'someone@gmail.com'}}
    print('send_email as user:', check_context_policy('send_email', ctx))

    ctx2 = {'user_id': 'u1', 'user_role': 'user',
            'parameters': {'path': '/etc/passwd'}}
    print('write_file outside sandbox:', check_context_policy('write_file', ctx2))

ポリシーエンジンによる動的ポリシー

ハードコードされたポリシーは、実運用で更新するのが困難です。OPA（Open Policy Agent）のようなポリシーエンジンを使用し、コードではなくデータとして定義されたポリシーを評価してください。エージェントを再デプロイせずにポリシーを更新できます。

import requests

OPA_URL = 'http://localhost:8181/v1/data/agent/allow'

def opa_policy_check(action: str, context: dict) -> tuple[bool, str]:
    payload = {
        'input': {
            'action':   action,
            'user_id':  context.get('user_id'),
            'role':     context.get('user_role', 'user'),
            'params':   context.get('parameters', {}),
            'time_utc': datetime.now(timezone.utc).isoformat()
        }
    }
    try:
        resp = requests.post(OPA_URL, json=payload, timeout=0.5)
        result = resp.json().get('result', {})
        allowed = result.get('allow', False)
        reason  = result.get('reason', 'Policy decision')
        return allowed, reason
    except Exception as e:
        # Fail closed: deny if policy engine is unreachable
        return False, f'Policy engine unavailable: {e}'

Fail-ClosedとFail-Open

ポリシーエンジンを利用できない場合、2つの選択肢があります。

  • Fail-closed: すべてのアクションを拒否します。安全ですが、エージェントは動作を停止します。
  • Fail-open: すべてのアクションを許可します。エージェントは動作を続けますが、ポリシーは適用されません。

セキュリティが重要なエージェントでは、必ずfail-closedにしてください。低リスクのアクションを実行する生産性向上エージェントでは、fail-openが許容される場合があります。

FAIL_CLOSED = True  # Configure per agent

def safe_policy_check(action: str, context: dict) -> tuple[bool, str]:
    try:
        return can_agent_do(action, context)
    except Exception as e:
        if FAIL_CLOSED:
            return False, f'Policy check failed (fail-closed): {e}'
        else:
            # Log the failure but allow the action
            import logging
            logging.warning('Policy check error (fail-open): %s', e)
            return True, 'Policy check bypassed due to error (fail-open)'

アクションのレート制限

ポリシー適用にはレート制限を含めることができます。たとえば、エージェントにメール送信を許可しつつ、1セッションあたり5通までに制限できます。上限を超えると、アクションは拒否されます。

from collections import defaultdict
import time

action_counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
action_window_start: dict[str, float] = defaultdict(float)

ACTION_RATE_LIMITS = {
    'send_email':   {'limit': 5,  'window_secs': 3600},  # 5/hour
    'write_file':   {'limit': 50, 'window_secs': 300},
    'web_search':   {'limit': 20, 'window_secs': 60}
}

def check_rate_limit(action: str, session_id: str) -> tuple[bool, str]:
    limit_config = ACTION_RATE_LIMITS.get(action)
    if not limit_config:
        return True, 'No rate limit defined'

    window = limit_config['window_secs']
    now = time.time()
    key = f'{session_id}:{action}'

    if now - action_window_start[key] > window:
        action_counts[key] = defaultdict(int)
        action_window_start[key] = now

    action_counts[key]['count'] += 1
    if action_counts[key]['count'] > limit_config['limit']:
        return False, f'Rate limit exceeded: {action} ({limit_config["limit"]}/{window}s)'
    return True, 'Within rate limit'

if __name__ == '__main__':
    for i in range(6):
        allowed, reason = check_rate_limit('send_email', 'session-1')
    print(f'After 6 send_email calls: allowed={allowed}, reason={reason}')

ポリシー違反のログ記録

ポリシーによる拒否はすべて、完全なコンテキストとともに記録する必要があります。予期しないエージェントの動作をデバッグしたり、セキュリティインシデントを調査したりするときは、まずこれらのログを確認します。

import logging, json, time

policy_logger = logging.getLogger('agent.policy')

def enforced_action(agent_id: str, user_id: str, action: str,
                    context: dict, audit_log) -> tuple[bool, str]:
    allowed, reason = safe_policy_check(action, context)

    log_entry = {
        'ts':       time.time(),
        'agent_id': agent_id,
        'user_id':  user_id,
        'action':   action,
        'allowed':  allowed,
        'reason':   reason,
        'params':   context.get('parameters', {})
    }

    if allowed:
        policy_logger.info('ALLOWED %s', json.dumps(log_entry))
    else:
        policy_logger.warning('DENIED %s', json.dumps(log_entry))

    audit_log.append(
        agent_id, user_id,
        f'POLICY_{"ALLOW" if allowed else "DENY"}',
        {'action': action},
        {'allowed': allowed, 'reason': reason},
        context.get('session_id', '')
    )
    return allowed, reason

リソースのスコープ設定

許可されたアクションであっても、エージェントがアクセスできるリソースの範囲を制限します。ユーザーAのドキュメントを扱うエージェントは、read_fileがallowlistに含まれていても、ユーザーBのファイルを読み取れないようにする必要があります。

def check_resource_scope(action: str, context: dict) -> tuple[bool, str]:
    user_id  = context.get('user_id', '')
    params   = context.get('parameters', {})

    if action == 'read_file':
        path = params.get('path', '')
        # Each user's files must be under their own namespace
        if not path.startswith(f'/workspace/{user_id}/'):
            return False, (
                f'User {user_id} cannot read files outside '
                f'/workspace/{user_id}/'
            )

    if action == 'query_database':
        table = params.get('table', '')
        allowed_tables = {'products', 'public_docs', f'user_{user_id}_data'}
        if table not in allowed_tables:
            return False, f'Table "{table}" not in scope for user {user_id}'

    return True, 'Resource scope check passed'

if __name__ == '__main__':
    ctx = {'user_id': 'u1', 'parameters': {'path': '/workspace/u2/secret.txt'}}
    print('Cross-user file read:', check_resource_scope('read_file', ctx))

    ctx2 = {'user_id': 'u1', 'parameters': {'path': '/workspace/u1/notes.txt'}}
    print('Own file read:      ', check_resource_scope('read_file', ctx2))

ポリシールールのテスト

ポリシールールはコードであるため、テストが必要です。各ルールの単体テストを作成し、拒否と許可が正しく機能すること、エッジケースによって意図しないポリシーの回避が発生しないことを確認してください。

def test_policy_rules():
    # Denylist blocks unconditionally
    ok, msg = can_agent_do('delete_all_records', {'user_role': 'admin'})
    assert not ok, 'Denylist should block even for admin'

    # Email requires trusted domain
    ok, msg = can_agent_do('send_email', {
        'user_role': 'email_agent',
        'parameters': {'to': 'attacker@evil.com'}
    })
    assert not ok, 'Should block untrusted email recipient'

    # File write outside allowed paths
    ok, msg = can_agent_do('write_file', {
        'user_role': 'user',
        'parameters': {'path': '/etc/crontab'}
    })
    assert not ok, 'Should block write to /etc/'

    print('All policy tests passed')

test_policy_rules()

ポリシー判断のキャッシュ

すべてのアクションでポリシーエンジンを呼び出すと、特に外部のOPAサービスを使用している場合に遅延が増加します。ラウンドトリップを減らすため、(action, context_hash)の組み合わせごとに、直近の判断を短いTTLでキャッシュしてください。

import hashlib, time

policy_cache: dict[str, dict] = {}
POLICY_CACHE_TTL = 10  # seconds — short TTL so policy updates take effect quickly

def cached_policy_check(action: str, context: dict) -> tuple[bool, str]:
    ctx_hash = hashlib.md5(
        f'{action}:{context.get("user_id")}:{context.get("user_role")}'
        .encode()
    ).hexdigest()
    key = f'{action}:{ctx_hash}'
    entry = policy_cache.get(key)
    if entry and time.time() - entry['ts'] < POLICY_CACHE_TTL:
        return entry['result']
    result = can_agent_do(action, context)
    policy_cache[key] = {'result': result, 'ts': time.time()}
    return result

ポリシー適用における「fail-closed」方式とは

fail-closedとfail-openのどちらを選ぶかは、あらゆるポリシー適用システムにおける基本的なセキュリティ上のトレードオフです。それぞれの方式が適切な場面を知ることは、ガバナンスの基本概念です。

ポリシー適用まとめ

エージェントのポリシー適用では、単一のゲートとしてのアクション前チェック、基本的な判断のためのallowlist + denylist、リソースのスコープ設定や役割チェックのためのコンテキスト固有のルール、更新可能なルールのための動的ポリシーエンジン（OPA）、セッションごとのアクション単位のレート制限、およびセキュリティが重要なエージェント向けのfail-closedデフォルトを使用します。

よくある質問

「エージェントアクションへのポリシー適用」レッスンは無料ですか？

はい。「エージェントアクションへのポリシー適用」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「エージェントアクションへのポリシー適用」で何を学びますか？

実行前のポリシーチェック、許可リスト・拒否リスト、動的なポリシールールを扱います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「エージェントアクションへのポリシー適用」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. エージェントの改ざん不能なアクションログ
  2. エージェントアクションへのポリシー適用
  3. 規制コンプライアンス：GDPR と SOC2
  4. Human-in-the-Loop 承認ゲート
← AI Agentsに戻る