pure-agent-dev

[zyntromedia]

Code Guide: pure-agent-dev

โครงสร้างนี้เหมาะกับการทำ Agent + FastAPI + Provider Adapter + BytePlus ECS โดยแยก responsibility ชัดเจนระหว่าง API, AI logic, cloud infrastructure และ validation. จุดสำคัญคือ Agent ไม่ควรรู้ว่า BytePlus SDK ทำงานอย่างไร เพราะวันหนึ่ง provider เปลี่ยนแล้วเราไม่ควรต้องผ่าตัดสมอง Agent กันใหม่

This architecture is suitable for Agent + FastAPI + Provider Adapter + BytePlus ECS. The key principle is that the Agent should never depend directly on the BytePlus SDK.

pure-agent-dev/
│
├── app/
│ ├── init.py
│ │
│ ├── main.py
│ │
│ ├── api/
│ │ ├── init.py
│ │ └── routes/
│ │ ├── init.py
│ │ ├── health.py
│ │ ├── tasks.py
│ │ └── compute.py
│ │
│ ├── agents/
│ │ ├── init.py
│ │ ├── planner.py
│ │ └── executor.py
│ │
│ ├── providers/
│ │ ├── init.py
│ │ ├── base.py
│ │ │
│ │ └── byteplus/
│ │ ├── init.py
│ │ ├── client.py
│ │ └── ecs.py
│ │
│ ├── schemas/
│ │ ├── init.py
│ │ ├── agent.py
│ │ ├── compute.py
│ │ └── task.py
│ │
│ └── services/
│ ├── init.py
│ ├── task_service.py
│ └── compute_service.py
│
├── schemas/
│ └── agent-task.schema.json
│
├── tests/
│ ├── init.py
│ ├── test_health.py
│ ├── test_planner.py
│ ├── test_executor.py
│ ├── test_compute_service.py
│ └── providers/
│ └── test_byteplus_ecs.py
│
├── agent.yaml
│
├── requirements.txt
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env.example
│
├── .github/
│ └── workflows/
│ └── ci.yml
│
└── README.md

1. Layer Architecture

┌──────────────────┐
│ Client │
└────────┬─────────┘
│ HTTP
▼
┌──────────────────┐
│ FastAPI │
│ API Routes │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Services │
│ Business Logic │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Agent │
│ Planner/Executor │
└────────┬─────────┘
│
▼
┌─────────────────────────┐
│ Provider Interface │
│ ComputeProvider │
└────────────┬────────────┘
│
▼
┌──────────────────┐
│ BytePlus Adapter │
│ ECS API │
└──────────────────┘

The dependency direction should remain:

API
↓
Services
↓
Agents
↓
Provider Interface
↓
Provider Implementation
↓
External Cloud SDK

Avoid this:

Agent
↓
BytePlus SDK

That would tightly couple the AI layer to infrastructure.

---

2. app/main.py

เป็น entry point ของ FastAPI application.

from fastapi import FastAPI

from app.api.routes import compute, health, tasks

app = FastAPI(
title="Pure Agent API",
version="1.0.0",
)

app.include_router(health.router)
app.include_router(tasks.router, prefix="/v1")
app.include_router(compute.router, prefix="/v1")

Run:

uvicorn app.main:app --reload

Production:

uvicorn app.main:app --host 0.0.0.0 --port 8000

---

3. providers/base.py

นี่คือ abstraction สำคัญที่สุดของ architecture.

from abc import ABC, abstractmethod

class ComputeProvider(ABC):

@abstractmethod
async def list_instances(self):
    raise NotImplementedError

@abstractmethod
async def start_instance(self, instance_id: str):
    raise NotImplementedError

@abstractmethod
async def stop_instance(self, instance_id: str):
    raise NotImplementedError

@abstractmethod
async def reboot_instance(self, instance_id: str):
    raise NotImplementedError

The Agent only depends on ComputeProvider.

It does not care whether the implementation is:

BytePlus
AWS
Azure
GCP
Mock
Local Docker
Kubernetes

---

4. providers/byteplus/client.py

ไฟล์นี้รับผิดชอบเฉพาะการสร้าง BytePlus client และ credential configuration.

import os

class BytePlusClient:
def init(self):
self.access_key = os.environ["BYTEPLUS_ACCESS_KEY"]
self.secret_key = os.environ["BYTEPLUS_SECRET_KEY"]
self.region = os.getenv(
"BYTEPLUS_REGION",
"ap-southeast-1",
)

def get_region(self) -> str:
    return self.region

ไม่ควรใส่ business logic ในไฟล์นี้.

client.py
↓
credentials
region
SDK initialization
connection configuration

---

5. providers/byteplus/ecs.py

ไฟล์นี้ implements ComputeProvider.

from app.providers.base import ComputeProvider
from app.providers.byteplus.client import BytePlusClient

class BytePlusECSProvider(ComputeProvider):

def __init__(self, client: BytePlusClient):
    self.client = client

async def list_instances(self):
    # Call BytePlus ECS DescribeInstances
    return []

async def start_instance(self, instance_id: str):
    # Call BytePlus ECS StartInstance
    return {
        "instance_id": instance_id,
        "status": "starting",
    }

async def stop_instance(self, instance_id: str):
    return {
        "instance_id": instance_id,
        "status": "stopping",
    }

async def reboot_instance(self, instance_id: str):
    return {
        "instance_id": instance_id,
        "status": "rebooting",
    }

ตัว implementation จริงค่อยเสียบ BytePlus SDK ตรงนี้.

The rest of the application remains unchanged.

---

6. schemas/compute.py

ใช้ Pydantic สำหรับ runtime validation.

from pydantic import BaseModel, Field

class InstanceRequest(BaseModel):
instance_id: str = Field(min_length=1)

class InstanceResponse(BaseModel):
instance_id: str
status: str

---

7. schemas/task.py

Agent task ควรเป็น structured data ไม่ใช่ free-form string อย่างเดียว.

from typing import Literal

from pydantic import BaseModel, Field

class AgentTask(BaseModel):
task_id: str
action: Literal[
"list_instances",
"start_instance",
"stop_instance",
"reboot_instance",
]
instance_id: str | None = Field(
default=None,
)

ตัวนี้ต้องสอดคล้องกับ:

schemas/agent-task.schema.json

ดังนั้นจะมีสองระดับ:

JSON Schema
↓
External contract
↓
Pydantic
↓
Runtime validation

---

8. agents/planner.py

Planner แปลง intent → structured task.

from app.schemas.task import AgentTask

class AgentPlanner:

def plan(
    self,
    action: str,
    instance_id: str | None = None,
) -> AgentTask:
    return AgentTask(
        task_id="generated-task",
        action=action,
        instance_id=instance_id,
    )

Planner ไม่ควรเรียก BytePlus.

User intent
↓
Planner
↓
AgentTask

ไม่ใช่:

User intent
↓
Planner
↓
BytePlus API

---

9. agents/executor.py

Executor เป็นคนเอา task ไป execute ผ่าน provider.

from app.providers.base import ComputeProvider
from app.schemas.task import AgentTask

class AgentExecutor:

def __init__(self, provider: ComputeProvider):
    self.provider = provider

async def execute(self, task: AgentTask):

    if task.action == "list_instances":
        return await self.provider.list_instances()

    if task.action == "start_instance":
        return await self.provider.start_instance(
            task.instance_id
        )

    if task.action == "stop_instance":
        return await self.provider.stop_instance(
            task.instance_id
        )

    if task.action == "reboot_instance":
        return await self.provider.reboot_instance(
            task.instance_id
        )

    raise ValueError(
        f"Unsupported action: {task.action}"
    )

Flow:

Planner
↓
AgentTask
↓
Executor
↓
ComputeProvider
↓
BytePlusECSProvider
↓
BytePlus ECS

---

10. services/compute_service.py

Service เป็น application/business layer.

from app.agents.executor import AgentExecutor
from app.agents.planner import AgentPlanner

class ComputeService:

def __init__(
    self,
    planner: AgentPlanner,
    executor: AgentExecutor,
):
    self.planner = planner
    self.executor = executor

async def execute(
    self,
    action: str,
    instance_id: str | None = None,
):
    task = self.planner.plan(
        action=action,
        instance_id=instance_id,
    )

    return await self.executor.execute(task)

---

11. api/routes/compute.py

FastAPI route ควรบางที่สุด.

from fastapi import APIRouter

from app.agents.executor import AgentExecutor
from app.agents.planner import AgentPlanner
from app.providers.byteplus.client import BytePlusClient
from app.providers.byteplus.ecs import BytePlusECSProvider
from app.services.compute_service import ComputeService

router = APIRouter(prefix="/compute", tags=["compute"])

def get_service() -> ComputeService:
client = BytePlusClient()
provider = BytePlusECSProvider(client)

executor = AgentExecutor(provider)
planner = AgentPlanner()

return ComputeService(
    planner=planner,
    executor=executor,
)

@router.post("/instances/{instance_id}/start")
async def start_instance(instance_id: str):

service = get_service()

return await service.execute(
    action="start_instance",
    instance_id=instance_id,
)

ใน production ควรเปลี่ยน get_service() เป็น FastAPI Dependency Injection และจัด lifecycle ของ SDK client ให้เหมาะสม.

---

12. agent.yaml

ใช้เป็น declarative configuration.

name: pure-agent
version: "1.0"

runtime:
language: python
framework: fastapi

agent:
planner: app.agents.planner.AgentPlanner
executor: app.agents.executor.AgentExecutor

providers:
compute:
default: byteplus

byteplus:
  type: ecs
  region: ${BYTEPLUS_REGION}

tasks:
allowed_actions:
- list_instances
- start_instance
- stop_instance
- reboot_instance

แนวคิดคือ:

agent.yaml
↓
Configuration
↓
Application

ไม่ควร hard-code provider selection ไว้ใน Agent.

---

13. schemas/agent-task.schema.json

ควรเป็น contract ระหว่าง Agent components.

{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AgentTask",
"type": "object",
"required": [
"task_id",
"action"
],
"properties": {
"task_id": {
"type": "string",
"minLength": 1
},
"action": {
"type": "string",
"enum": [
"list_instances",
"start_instance",
"stop_instance",
"reboot_instance"
]
},
"instance_id": {
"type": [
"string",
"null"
]
}
},
"additionalProperties": false
}

JSON Schema เป็น external contract ส่วน Pydantic เป็น runtime model.

---

14. Environment

.env.example

BYTEPLUS_ACCESS_KEY=
BYTEPLUS_SECRET_KEY=
BYTEPLUS_REGION=ap-southeast-1

APP_ENV=development
LOG_LEVEL=INFO

Secret จริงไม่ควร commit:

.env
*.secret

GitHub Actions ใช้:

BYTEPLUS_ACCESS_KEY
BYTEPLUS_SECRET_KEY
BYTEPLUS_REGION

จาก GitHub Secrets/Environment.

---

15. Dockerfile

FROM python:3.12-slim

WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app
COPY schemas ./schemas
COPY agent.yaml .

EXPOSE 8000

CMD [
"uvicorn",
"app.main:app",
"--host",
"0.0.0.0",
"--port",
"8000"
]

---

16. docker-compose.yml

services:
api:
build: .
ports:
- "8000:8000"
env_file:
- .env
restart: unless-stopped

Architecture ตอน local development:

Docker Compose
│
▼
FastAPI
│
▼
Agent
│
▼
BytePlus ECS

---

17. Testing Strategy

ไม่ควรให้ unit tests ยิง BytePlus จริงทุกครั้ง เพราะ CI จะกลายเป็นเครื่องสล็อตแมชชีนที่มี cloud bill เป็นรางวัล

ใช้ mock provider:

class MockComputeProvider:

async def list_instances(self):
    return []

async def start_instance(self, instance_id):
    return {
        "instance_id": instance_id,
        "status": "starting",
    }

async def stop_instance(self, instance_id):
    return {
        "instance_id": instance_id,
        "status": "stopping",
    }

async def reboot_instance(self, instance_id):
    return {
        "instance_id": instance_id,
        "status": "rebooting",
    }

Test:

Unit Test
│
├── Planner
├── Executor
├── Service
└── Provider interface
│
▼
Mock Provider

Integration Test
│
▼
BytePlus sandbox/test account

---

18. GitHub Actions

.github/workflows/ci.yml

name: CI

on:
push:
branches:
- main
- develop

pull_request:

permissions:
contents: read

jobs:
test:
runs-on: ubuntu-latest

steps:
  - name: Checkout
    uses: actions/checkout@v4

  - name: Setup Python
    uses: actions/setup-python@v5
    with:
      python-version: "3.12"
      cache: pip

  - name: Install dependencies
    run: |
      python -m pip install --upgrade pip
      pip install -r requirements.txt
      pip install pytest ruff

  - name: Lint
    run: |
      ruff check .

  - name: Test
    run: |
      pytest -q

CI ไม่จำเป็นต้องมี BytePlus credentials สำหรับ unit tests.

---

19. Dependency Graph

สุดท้าย dependency graph ควรเป็นแบบนี้:

┌──────────────┐
│ FastAPI │
└──────┬───────┘
│
▼
┌──────────────┐
│ Services │
└──────┬───────┘
│
▼
┌──────────────┐
│ Agents │
└──────┬───────┘
│
▼
┌────────────────────┐
│ ComputeProvider │
│ Interface │
└─────────┬──────────┘
│
┌─────────────┴─────────────┐
▼ ▼
┌──────────────────┐ ┌──────────────────┐
│ BytePlus ECS │ │ Mock Provider │
│ Provider │ │ │
└────────┬─────────┘ └──────────────────┘
│
▼
┌──────────────────┐
│ BytePlus SDK/API │
└──────────────────┘

Core rule

FastAPI = HTTP/API
Services = Business orchestration
Planner = Intent → Task
Executor = Task → Provider
Provider = Infrastructure abstraction
BytePlus ECS = Infrastructure implementation
Pydantic = Runtime validation
JSON Schema = Contract
agent.yaml = Configuration
Docker = Runtime packaging
GitHub CI = Verification

ไทย: โครงสร้างนี้ทำให้ pure-agent-dev สามารถเริ่มจาก BytePlus ECS ได้ แต่ยังคงเปลี่ยน provider, เพิ่ม Agent, เพิ่ม task หรือย้าย deployment โดยไม่ต้องรื้อ architecture ทั้งระบบ

English: This structure lets pure-agent-dev start with BytePlus ECS while remaining provider-agnostic, testable, containerized, and suitable for adding additional agents or cloud providers later.

ผมสามารถต่อจาก guide นี้เป็น ชุดไฟล์ production-ready ทั้ง repository พร้อม requirements.txt, Pydantic v2, dependency injection, BytePlus SDK integration, validators, planner/executor tests และ CI ได้ครับ.

[zyntromedia]

Based on the GitHub issue #63 ("pure-agent-dev") from the fastapi-python-boilerplate project, the directory structure is designed to support a clean, modular architecture that separates concerns clearly between API, services, agents, providers, and external SDKs.

Overview of the Directory Structure

Here is a typical directory layout inspired by the issue discussion:

 
/app
/api
└── (FastAPI route handlers, request/response models)
/services
└── (Business logic, orchestrating calls between API and agents)
/agents
└── (Core AI agent logic, decision making, task planning)
/providers
├── /interfaces
│ └── provider_interface.py # Abstract base classes or protocols
└── /implementations
└── byteplus_provider.py # Concrete provider implementations (e.g., BytePlus)
└── cloudx_provider.py # Example new provider implementation
/config
└── config.yaml # Configuration files (e.g., cloud credentials)
/tests
└── (Unit and integration tests, including mocks for providers)
 

Key Points About Each Folder

api/
Contains FastAPI route handlers and Pydantic models for request validation and response serialization. This layer is responsible for handling HTTP requests and responses.
services/
Implements business logic that coordinates between the API layer and the AI agents. Services call agents to perform tasks and process results before returning to the API.
agents/
Contains the core AI agent logic, including task planning, execution strategies, and interaction with providers via the provider interface.
providers/interfaces/
Defines abstract interfaces or protocols that all provider implementations must follow. This ensures that agents can interact with any provider in a uniform way.
providers/implementations/
Contains concrete implementations of providers, such as BytePlus or any new cloud provider. Each implementation interacts with its respective external SDK or API.
config/
Stores configuration files such as YAML files for cloud credentials, API keys, endpoints, and other environment-specific settings.
tests/
Includes unit and integration tests. This folder also contains mock implementations of providers to facilitate testing without real cloud calls.

Benefits of This Structure

Separation of Concerns: Each layer has a clear responsibility, making the codebase easier to maintain and extend.
Modularity: Providers are isolated behind interfaces, allowing easy swapping or mocking.
Testability: Mock providers and isolated layers simplify writing tests.
Scalability: The architecture supports adding new providers, agents, or services without major refactoring.

If you want, I can help you with a more detailed example of files inside each folder or how to set up the configuration and testing based on this structure.

[zyntromedia]

To implement a new provider in the fastapi-python-boilerplate project based on the "pure-agent-dev" architecture described in issue #63, you should follow these steps:

1. Define the Provider Interface

Ensure you have a clear interface (abstract base class or protocol) that specifies the methods your provider must implement.
This interface defines the contract for all providers, including your new one.

2. Create the Provider Implementation

Implement the provider interface in a new class specific to your cloud provider or service.
This class will contain the actual logic to interact with the external cloud SDK or API.
For example, if your new provider is "CloudX," create a class like  CloudXProvider  that implements all required methods.

3. Integrate with Dependency Injection

Register your new provider implementation in the dependency injection system used by the project.
This allows the AI Agent and services to use your provider seamlessly without changing their code.

4. Configuration

Add any necessary configuration parameters (e.g., API keys, endpoints) for your new provider in the configuration files (e.g., YAML).
Ensure these configurations are loaded and passed to your provider implementation correctly.

5. Testing

Create a mock or stub implementation of your provider interface for unit testing.
This mock should simulate interactions without making real cloud calls.
Write tests to verify your provider implementation works as expected.

6. Update Directory Structure

Follow the project's directory conventions for placing your provider implementation and related files.
Typically, this would be under a  providers  or similar folder.

Summary of Implementation Flow

API calls Services
Services call Agents
Agents call Provider Interface
Provider Interface calls your Provider Implementation
Provider Implementation interacts with the external cloud SDK/API

By following this pattern, your new provider will be modular, testable, and easily swappable with other providers.

If you want, I can help you draft a sample code template for a new provider implementation based on this architecture. Would you like that?

[zyntromedia]

Writing tests in the fastapi-python-boilerplate project, especially following the "pure-agent-dev" architecture from issue #63, involves focusing on modularity and mocking external dependencies like cloud providers. Here's a detailed explanation of how to write a test, particularly for a provider or agent component:

 

How to Write a Test

1. Choose What to Test

Unit tests: Test individual components in isolation, e.g., a provider implementation or an agent method.
Integration tests: Test interactions between components, e.g., service calling an agent with a provider.

2. Use Mock Providers

To avoid real cloud calls, use mock implementations of the provider interface.
This ensures tests are fast, reliable, and cost-free.

3. Use a Testing Framework

The project likely uses  pytest  (common in Python FastAPI projects).
Use  pytest  fixtures for setup and teardown.

 

Example: Unit Test for a Provider Implementation Using a Mock

 
import pytest
from app.providers.interfaces.provider_interface import ProviderInterface
from app.providers.implementations.mock_provider import MockProvider

@pytest.fixture
def mock_provider() -> ProviderInterface:
return MockProvider()

def test_execute_task(mock_provider):
task_data = {"task": "test_task"}
result = mock_provider.execute_task(task_data)
assert result["status"] == "mocked"
assert "details" in result

def test_get_status(mock_provider):
instance_id = "12345"
status = mock_provider.get_status(instance_id)
assert status["instance_id"] == instance_id
assert status["status"] == "mocked"
 

 

Example: Testing an Agent Using Dependency Injection

 
from fastapi import Depends
from app.agents.some_agent import SomeAgent
from app.providers.interfaces.provider_interface import ProviderInterface
from app.providers.implementations.mock_provider import MockProvider

def get_mock_provider() -> ProviderInterface:
return MockProvider()

def test_agent_execute_task():
agent = SomeAgent(provider=get_mock_provider())
task_data = {"task": "example"}
result = agent.perform_task(task_data)
assert result["status"] == "mocked"
 

 

Key Points for Writing Tests

Isolate external dependencies: Use mocks for providers.
Test expected outputs: Assert on returned data structures and values.
Use fixtures: For reusable setup like mock providers or test data.
Keep tests small and focused: Each test should verify one behavior.

 

Folder for Tests

Place your test files under the  /tests  directory, mirroring the app structure:

 
/tests
/providers
└── test_mock_provider.py
/agents
└── test_some_agent.py
/services
└── test_some_service.py
 

 

If you want, I can help you generate a full test file template or assist with setting up test configuration for pytest in this project.