Dynamic Routing

Dynamic routing is a powerful feature provided by the Nekro Agent plugin system that allows plugins to create custom Web API endpoints. Based on the FastAPI framework, dynamic routing supports all the features of modern Web API development, including complete RESTful API design, request validation, response handling, and automatic documentation generation.

Feature Overview

With dynamic routing, plugins can:

• Create RESTful APIs: Support HTTP methods such as GET, POST, PUT, DELETE
• Request Validation: Use Pydantic models for request body and query parameter validation
• Response Handling: Standardized response formats and error handling
• API Documentation: Automatically generate OpenAPI/Swagger documentation
• Path Parameters: Support for dynamic route parameters and query parameters
• Middleware Support: Request preprocessing and response post-processing

Basic Usage

Registering Routes

Use the @plugin.mount_router() decorator to register a route creation function:

from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from nekro_agent.services.plugin.base import NekroPlugin

# Create plugin instance
plugin = NekroPlugin(
    name="API Example Plugin",
    module_name="api_example",
    description="Demonstrates dynamic routing functionality",
    version="1.0.0",
    author="your_name"
)

@plugin.mount_router()
def create_router() -> APIRouter:
    """Create and configure plugin routes"""
    router = APIRouter()

    @router.get("/")
    async def index():
        return {"message": "Welcome to the Plugin API"}

    return router

Route Access Paths

Plugin routes are accessed in the format: /plugins/{author}.{module_name}/{path}

For example, the routes in the above example can be accessed at:

• GET /plugins/your_name.api_example/ - Plugin homepage

Data Model Definition

Using Pydantic Models

from pydantic import BaseModel, Field
from typing import Optional

class UserModel(BaseModel):
    id: int
    name: str = Field(..., min_length=1, max_length=50, description="User name")
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="Email address")
    age: Optional[int] = Field(None, ge=0, le=150, description="Age")

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)

Complete CRUD Example

from typing import Dict, List
from fastapi import APIRouter, HTTPException, Query, Path

# Simulated data storage
users_db: Dict[int, UserModel] = {}
next_id = 1

@plugin.mount_router()
def create_router() -> APIRouter:
    router = APIRouter()

    @router.get("/", summary="API Homepage")
    async def api_home():
        """Return basic API information"""
        return {
            "name": plugin.name,
            "version": plugin.version,
            "endpoints": [
                "GET / - API Homepage",
                "GET /users - Get user list",
                "POST /users - Create user",
                "GET /users/{user_id} - Get user details",
                "PUT /users/{user_id} - Update user",
                "DELETE /users/{user_id} - Delete user"
            ]
        }

    # Get user list
    @router.get("/users", response_model=List[UserModel], summary="Get User List")
    async def get_users(
        limit: int = Query(10, ge=1, le=100, description="Result count limit"),
        offset: int = Query(0, ge=0, description="Offset"),
        name_filter: Optional[str] = Query(None, description="Filter by name")
    ):
        """Get user list with pagination and filtering support"""
        users = list(users_db.values())

        # Filter by name
        if name_filter:
            users = [u for u in users if name_filter.lower() in u.name.lower()]

        # Pagination
        return users[offset:offset + limit]

    # Create user
    @router.post("/users", response_model=UserModel, summary="Create User", status_code=201)
    async def create_user(user_data: CreateUserRequest):
        """Create a new user"""
        global next_id

        # Check if email already exists
        for user in users_db.values():
            if user.email == user_data.email:
                raise HTTPException(status_code=400, detail="Email already exists")

        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

    # Get user details
    @router.get("/users/{user_id}", response_model=UserModel, summary="Get User Details")
    async def get_user(user_id: int = Path(..., ge=1, description="User ID")):
        """Get user details by ID"""
        if user_id not in users_db:
            raise HTTPException(status_code=404, detail="User not found")

        return users_db[user_id]

    # Update user
    @router.put("/users/{user_id}", response_model=UserModel, summary="Update User")
    async def update_user(
        user_id: int = Path(..., ge=1, description="User ID"),
        user_data: UpdateUserRequest = ...
    ):
        """Update user information"""
        if user_id not in users_db:
            raise HTTPException(status_code=404, detail="User not found")

        user = users_db[user_id]

        # Check for email conflicts
        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="Email is already used by another user")

        # Update fields
        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

    # Delete user
    @router.delete("/users/{user_id}", summary="Delete User")
    async def delete_user(user_id: int = Path(..., ge=1, description="User ID")):
        """Delete user"""
        if user_id not in users_db:
            raise HTTPException(status_code=404, detail="User not found")

        deleted_user = users_db.pop(user_id)
        return {"message": f"User '{deleted_user.name}' has been deleted", "deleted_user": deleted_user}

    return router

Advanced Features

Middleware Support

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):
        """Add request processing time header"""
        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

    # Other route definitions...

    return router

Dependency Injection

from fastapi import Depends

def get_current_user(user_id: int = Query(...)) -> UserModel:
    """Get current user (example dependency)"""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

@router.get("/profile", response_model=UserModel)
async def get_profile(current_user: UserModel = Depends(get_current_user)):
    """Get current user profile"""
    return current_user

File Upload Support

from fastapi import File, UploadFile

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """File upload example"""
    if not file.filename:
        raise HTTPException(status_code=400, detail="No file selected")

    # Save file
    content = await file.read()

    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(content)
    }

Response Models and Status Codes

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": "Request parameter error"},
        409: {"model": ErrorResponse, "description": "User already exists"}
    }
)
async def create_user_with_responses(user_data: CreateUserRequest):
    """Create user endpoint with detailed response definitions"""
    # Implementation logic...
    pass

Integration with Plugin Features

Sending Messages to Chat Channels

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]
):
    """Send notification to specified chat channel"""
    try:
        # Create context object
        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": "Notification sent"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to send: {str(e)}")

Dynamic routing provides a more complete and standard Web API development experience and is the recommended way to build external interfaces for plugins.