ダイナミックルーティング
ダイナミックルーティングは、Nekro Agentプラグインシステムによって提供される強力な機能で、プラグインがカスタムWeb APIエンドポイントを作成できるようにします。FastAPIフレームワークに基づいており、ダイナミックルーティングは完全なRESTful API設計、リクエスト検証、レスポンス処理、自動ドキュメント生成など、現代のWeb API開発のすべての機能をサポートします。
機能概要
ダイナミックルーティングを使用すると、プラグインは以下のことができます：
- RESTful APIの作成: GET、POST、PUT、DELETEなどのHTTPメソッドをサポート
- リクエスト検証: Pydanticモデルを使用したリクエストボディとクエリパラメータの検証
- レスポンス処理: 標準化されたレスポンス形式とエラーハンドリング
- APIドキュメント: OpenAPI/Swaggerドキュメントの自動生成
- パスパラメータ: 動的なルートパラメータとクエリパラメータのサポート
- ミドルウェアサポート: リクエスト前処理とレスポンス後処理
基本的な使用方法
ルートの登録
@plugin.mount_router()デコレータを使用してルート作成関数を登録します：
python
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from nekro_agent.services.plugin.base import NekroPlugin
# プラグインインスタンスを作成
plugin = NekroPlugin(
name="API Example Plugin",
module_name="api_example",
description="ダイナミックルーティング機能を実演",
version="1.0.0",
author="your_name"
)
@plugin.mount_router()
def create_router() -> APIRouter:
"""プラグインルートを作成して設定"""
router = APIRouter()
@router.get("/")
async def index():
return {"message": "プラグインAPIへようこそ"}
return routerルートアクセスパス
プラグインルートは/plugins/{author}.{module_name}/{path}の形式でアクセスされます。
例えば、上記の例のルートは以下でアクセスできます：
GET /plugins/your_name.api_example/- プラグインホームページ
データモデル定義
Pydanticモデルの使用
python
from pydantic import BaseModel, Field
from typing import Optional
class UserModel(BaseModel):
id: int
name: str = Field(..., min_length=1, max_length=50, description="ユーザー名")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="メールアドレス")
age: Optional[int] = Field(None, ge=0, le=150, description="年齢")
class CreateUserRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=50)
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: Optional[int] = Field(None, ge=0, le=150)
class UpdateUserRequest(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=50)
email: Optional[str] = Field(None, regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: Optional[int] = Field(None, ge=0, le=150)完全なCRUD例
python
from typing import Dict, List
from fastapi import APIRouter, HTTPException, Query, Path
# シミュレートされたデータストレージ
users_db: Dict[int, UserModel] = {}
next_id = 1
@plugin.mount_router()
def create_router() -> APIRouter:
router = APIRouter()
@router.get("/", summary="APIホームページ")
async def api_home():
"""基本的なAPI情報を返す"""
return {
"name": plugin.name,
"version": plugin.version,
"endpoints": [
"GET / - APIホームページ",
"GET /users - ユーザーリスト取得",
"POST /users - ユーザー作成",
"GET /users/{user_id} - ユーザー詳細取得",
"PUT /users/{user_id} - ユーザー更新",
"DELETE /users/{user_id} - ユーザー削除"
]
}
# ユーザーリスト取得
@router.get("/users", response_model=List[UserModel], summary="ユーザーリスト取得")
async def get_users(
limit: int = Query(10, ge=1, le=100, description="結果数の制限"),
offset: int = Query(0, ge=0, description="オフセット"),
name_filter: Optional[str] = Query(None, description="名前でフィルタリング")
):
"""ページネーションとフィルタリングをサポートするユーザーリストを取得"""
users = list(users_db.values())
# 名前でフィルタリング
if name_filter:
users = [u for u in users if name_filter.lower() in u.name.lower()]
# ページネーション
return users[offset:offset + limit]
# ユーザー作成
@router.post("/users", response_model=UserModel, summary="ユーザー作成", status_code=201)
async def create_user(user_data: CreateUserRequest):
"""新しいユーザーを作成"""
global next_id
# メールアドレスが既に存在するかチェック
for user in users_db.values():
if user.email == user_data.email:
raise HTTPException(status_code=400, detail="メールアドレスは既に存在します")
new_user = UserModel(
id=next_id,
name=user_data.name,
email=user_data.email,
age=user_data.age
)
users_db[next_id] = new_user
next_id += 1
return new_user
# ユーザー詳細取得
@router.get("/users/{user_id}", response_model=UserModel, summary="ユーザー詳細取得")
async def get_user(user_id: int = Path(..., ge=1, description="ユーザーID")):
"""IDでユーザー詳細を取得"""
if user_id not in users_db:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
return users_db[user_id]
# ユーザー更新
@router.put("/users/{user_id}", response_model=UserModel, summary="ユーザー更新")
async def update_user(
user_id: int = Path(..., ge=1, description="ユーザーID"),
user_data: UpdateUserRequest = ...
):
"""ユーザー情報を更新"""
if user_id not in users_db:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
user = users_db[user_id]
# メールアドレスの競合をチェック
if user_data.email and user_data.email != user.email:
for other_user in users_db.values():
if other_user.email == user_data.email and other_user.id != user_id:
raise HTTPException(status_code=400, detail="メールアドレスは既に他のユーザーによって使用されています")
# フィールドを更新
if user_data.name is not None:
user.name = user_data.name
if user_data.email is not None:
user.email = user_data.email
if user_data.age is not None:
user.age = user_data.age
return user
# ユーザー削除
@router.delete("/users/{user_id}", summary="ユーザー削除")
async def delete_user(user_id: int = Path(..., ge=1, description="ユーザーID")):
"""ユーザーを削除"""
if user_id not in users_db:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
deleted_user = users_db.pop(user_id)
return {"message": f"ユーザー '{deleted_user.name}' が削除されました", "deleted_user": deleted_user}
return router高度な機能
ミドルウェアサポート
python
from fastapi import Request, Response
import time
@plugin.mount_router()
def create_router() -> APIRouter:
router = APIRouter()
@router.middleware("http")
async def add_process_time_header(request: Request, call_next):
"""リクエスト処理時間ヘッダーを追加"""
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
# その他のルート定義...
return router依存性注入
python
from fastapi import Depends
def get_current_user(user_id: int = Query(...)) -> UserModel:
"""現在のユーザーを取得（依存性の例）"""
if user_id not in users_db:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
return users_db[user_id]
@router.get("/profile", response_model=UserModel)
async def get_profile(current_user: UserModel = Depends(get_current_user)):
"""現在のユーザープロファイルを取得"""
return current_userファイルアップロードサポート
python
from fastapi import File, UploadFile
@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
"""ファイルアップロードの例"""
if not file.filename:
raise HTTPException(status_code=400, detail="ファイルが選択されていません")
# ファイルを保存
content = await file.read()
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(content)
}レスポンスモデルとステータスコード
python
from fastapi import status
class ErrorResponse(BaseModel):
error: str
detail: str
@router.post(
"/users",
response_model=UserModel,
status_code=status.HTTP_201_CREATED,
responses={
400: {"model": ErrorResponse, "description": "リクエストパラメータエラー"},
409: {"model": ErrorResponse, "description": "ユーザーは既に存在します"}
}
)
async def create_user_with_responses(user_data: CreateUserRequest):
"""詳細なレスポンス定義を持つユーザー作成エンドポイント"""
# 実装ロジック...
passプラグイン機能との統合
チャットチャネルへのメッセージ送信
python
from nekro_agent.api import message
from nekro_agent.api.schemas import AgentCtx
@router.post("/notify/{chat_key}")
async def send_notification(
chat_key: str,
notification: Dict[str, str]
):
"""指定されたチャットチャネルに通知を送信"""
try:
# コンテキストオブジェクトを作成
ctx = await AgentCtx.create_by_chat_key(chat_key)
await message.send_text(
chat_key=chat_key,
text=notification.get("message", ""),
ctx=ctx
)
return {"status": "success", "message": "通知が送信されました"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"送信に失敗しました: {str(e)}")ダイナミックルーティングは、より完全で標準的なWeb API開発体験を提供し、プラグインの外部インターフェースを構築するための推奨される方法です。