エンドツーエンドプロジェクト：セキュアでデータベース連携型のエージェント構築

このレッスンのロックを解除するには、まず「認証、トークン、安全な資格情報の取り扱い」を完了してください。前のレッスンに戻る

プロジェクト概要

この集大成レッスンでは、認証されたユーザーに代わって4つの実世界タスクを実行する完全なAIエージェントを構築します：認証による身元確認、データベースへのレコード読み書き、外部API呼び出しによるデータの取得や送信、そしてユーザーが閲覧可能な範囲を制限するロールベースアクセス制御（RBAC）の実施です。これは、本コースのこれまでのすべての概念がつながる統合ポイントです。

構築するエージェントはリクエスト・レスポンスパターンに従います：ユーザーがリクエストを送信し、エージェントはユーザーの身元を確認し、ユーザーに許可されている操作を決定し、データベース照会、API呼び出し、またはその両方を行い、結果を返します。エージェントロジックにはPython、永続化にはSQLデータベース、外部データにはモック外部API、認証にはシンプルなJWT（JSON Web Token）メカニズムを使用します。

アーキテクチャとセットアップ

コードを書く前に、4つの構成要素とそれらの接続方法を概説します。ユーザーがJWTトークン付きでリクエストを送信します。エージェントはトークンを検証してユーザーIDとロールを抽出し、ユーザーの意図に応じてデータベースハンドラまたはAPI呼び出し部にルーティングします。データベースハンドラはインジェクション攻撃を防ぐためパラメータ化されたSQLクエリを使用します。API呼び出し部は同じJWTを外部リクエストに添付するため、外部サービスが呼び出し元を独立して検証できます。

python

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: int
    role: str
    token: str

このUserContextオブジェクトはすべての内部呼び出しと共に流れます。リクエストがエージェントに入るとコンテキストがスタンプされ、すべての下流関数が同じコンテキストを受け取るため、認可チェックはエントリーポイントだけでなくすべてのステップで実行されます。

認証の実装

認証とは、ユーザーの身元を証明することです。シンプルなJWTスタイルのトークンを使用します：ユーザーID、ロール、有効期限タイムスタンプを含むbase64エンコード文字列で、共有秘密鍵で署名されています。本番環境ではPyJWTのようなライブラリを使用しますが、ここでは仕組みを理解できるよう最小限に留めています。

python

import base64, json, hmac, hashlib, time

SECRET = "change-me-in-production"

def create_token(user_id: int, role: str) -> str:
    payload = {"uid": user_id, "role": role, "exp": int(time.time()) + 3600}
    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
    sig = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    return f"{body}.{sig}"

def verify_token(token: str) -> UserContext:
    body, sig = token.split(".")
    expected = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise PermissionError("Invalid token signature")
    payload = json.loads(base64.urlsafe_b64decode(body))
    if payload["exp"] < int(time.time()):
        raise PermissionError("Token expired")
    return UserContext(user_id=payload["uid"], role=payload["role"], token=token)

verify_token関数はゲートキーパーです。すべての保護されたエンドポイントが最初にこれを呼び出します。tokenが改ざんされている、期限切れ、または欠落している場合、データベースやAPIコードが実行される前リクエストは拒否されます。

データベースの安全な照会

認証が完了したので、次のステップはエージェントにユーザーレコードを読み取らせることです。ここでの重要なルールは、SQLを文字列連結で組み立てないことです。常にパラメータ化されたクエリを使用し、ユーザー入力を実行可能なSQLではなくデータとして扱います。

python

import sqlite3

def get_user_orders(ctx: UserContext, target_user_id: int) -> list:
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, product, total FROM orders WHERE user_id = ?", (target_user_id,))
    rows = cur.fetchall()
    conn.close()
    return rows

ユーザーIDが文字列に補間されるのではなくタプルパラメータとして渡されていることに注目してください。このたった一つの習慣が、SQLインジェクション（最も一般的なデータベース攻撃）を防ぎます。

外部APIの呼び出し

多くのエージェントタスクは、宅配便の配送状況や配送ルートの天候など、データベースにないデータを必要とします。エージェントは外部HTTP APIを呼び出し、外部サービスが呼び出し元を検証できるよう同じJWTを渡します。

python

import urllib.request

def get_shipping_status(ctx: UserContext, order_id: int) -> dict:
    req = urllib.request.Request(
        f"https://api.courier.example/orders/{order_id}",
        headers={"Authorization": f"Bearer {ctx.token}"}
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())

トークンは新規生成されず転送されます。これにより、宅配便サービスは誰が要求したかをログに記録し、ユーザーごとに独自のレート制限を適用できます。

ロールベースアクセス制御の実施

認証はユーザーの身元を伝えます。認可（RBAC）はユーザーに許可されている操作を伝えます。これを実施する最もクリーンな場所は、すべてのエントリーポイントで呼び出される小さなポリシー関数です。

python

def require_role(ctx: UserContext, allowed: list[str]):
    if ctx.role not in allowed:
        raise PermissionError(f"Role '{ctx.role}' not permitted")

python

def list_all_orders(ctx: UserContext) -> list:
    require_role(ctx, ["admin"])
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, user_id, product, total FROM orders")
    rows = cur.fetchall()
    conn.close()
    return rows

一般ユーザーがlist_all_ordersを呼び出すと拒否されます。管理者が呼び出すとテーブル全体を取得できます。ポリシーチェックは1行ですが、これがセキュアなシステムとデータ侵害の差を生みます。

すべてを統合する

メインエージェント関数は、認証、データベースアクセス、API呼び出し、RBACを1つのフローに統合します。

python

def handle_request(token: str, intent: str, params: dict):
    ctx = verify_token(token)
    if intent == "my_orders":
        return get_user_orders(ctx, params["user_id"])
    elif intent == "all_orders":
        return list_all_orders(ctx)
    elif intent == "shipping":
        return get_shipping_status(ctx, params["order_id"])
    else:
        raise ValueError("Unknown intent")

すべてのパスはverify_tokenから始まります。機密データに触れるすべてのパスはrequire_roleを呼び出します。すべてのデータベースクエリはパラメータ化されています。すべての外部API呼び出しはユーザーのトークンを伝送します。あらゆる場所で繰り返されるこれら4つの習慣が、エージェントを本番環境に対応させるものなのです。

ユーザーのリクエストがエージェントに入り、verify_tokenを通じて分岐し、require_roleチェックへルーティングされ、データベースクエリパスと外部API呼び出しパスに分岐し、両方のパスがユーザーに戻ることを示すフロー図

まとめ

これで動作するエンドツーエンドのエージェントが完成しました。認証が署名付きトークンで身元を証明します。パラメータ化されたSQLクエリがインジェクションリスクなくデータを読み書きします。外部API呼び出しが同じトークンを再利用するため、外部サービスはユーザーごとに監査とレート制限を行えます。ロールベースアクセス制御が正面玄関だけでなくすべてのエントリーポイントに配置されるため、権限昇格には複数のチェックを突破する必要があります。実際のエージェントを構築する際も、パターンは同じです：1回認証し、あらゆる場所で認可し、常にパラメータ化し、すべての呼び出しでコンテキストを伝播させることです。

レッスンの練習

レッスンのチェックポイント

学んだ内容を復習し、回答へのフィードバックを受けましょう。

0/1