> For the complete documentation index, see [llms.txt](https://nuvex-ai.gitbook.io/nuvex-ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://nuvex-ai.gitbook.io/nuvex-ai/developer-guide/development-guide.md).

# Development Guide

### **Development Environment Setup**

#### **Prerequisites**

* Python 3.11+
* Poetry for dependency management
* Docker (optional)
* Git

***

### **Local Development Setup**

#### Clone the repository:

```bash
git clone https://github.com/0xdevpro/nuvex.git
cd nuvex
```

#### Install dependencies:

```bash
poetry install
```

#### Set up environment variables:

```bash
cp .env.example .env
```

#### Configure your `.env` file:

```env
HOST=localhost
PORT=8000
DEBUG=true
JWT_SECRET=your_jwt_secret
DATABASE_URL=postgresql://user:password@localhost:5432/nuvex
```

#### Start the development server:

```bash
poetry run python api.py
```

***

### **Project Structure**

```
nuvex/
├── agents/
│   ├── agent/          # Core agent implementation
│   ├── api/            # API endpoints
│   ├── common/         # Shared utilities
│   ├── middleware/     # Middleware components
│   ├── models/         # Data models
│   ├── protocol/       # Protocol definitions
│   ├── services/       # Business logic
│   └── utils/          # Utility functions
├── sql/                # Database migrations
├── tests/              # Test suite
├── api.py              # Main application entry
├── pyproject.toml      # Project dependencies
└── README.md           # Project documentation
```

***

### **Core Components**

#### **Agent System**

```python
from agents.agent.base import BaseAgent

class CustomAgent(BaseAgent):
    def __init__(self, config: dict):
        super().__init__(config)
        self.tools = []
        self.memory = []

    async def process(self, input_data: str) -> str:
        response = await self._think(input_data)
        return response

    async def _think(self, input_data: str) -> str:
        pass
```

***

#### **Tool Integration**

```python
from agents.agent.tool import BaseTool

class CustomTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="custom_tool",
            description="Tool description"
        )

    async def _run(self, input_data: dict) -> dict:
        result = await self.process_data(input_data)
        return {"result": result}

    async def process_data(self, data: dict) -> str:
        pass
```

***

#### **API Layer (FastAPI)**

```python
from fastapi import APIRouter, Depends
from agents.services.agent_service import AgentService

router = APIRouter()

@router.post("/agent/create")
async def create_agent(data: AgentCreate, service: AgentService = Depends()):
    return await service.create_agent(data)

@router.get("/agent/{agent_id}")
async def get_agent(agent_id: str, service: AgentService = Depends()):
    return await service.get_agent(agent_id)
```

***

#### **Service Layer**

```python
from agents.models.db import AgentModel
from agents.protocol.schemas import AgentCreate

class AgentService:
    async def create_agent(self, data: AgentCreate) -> AgentModel:
        self._validate_agent_data(data)
        agent = await AgentModel.create(**data.dict())
        await self._initialize_tools(agent)
        return agent

    async def _initialize_tools(self, agent: AgentModel):
        pass
```

### **Development Guidelines**

#### **Code Style**

* Follow **PEP 8**
* Use **type hints**
* Document all classes and methods
* Write clear, descriptive commit messages

**Example:**

```python
from typing import Optional, List

class DataProcessor:
    """
    Processes data for agent consumption.

    Attributes:
        batch_size: Size of data batches to process
        max_items: Maximum number of items to process
    """

    def __init__(self, batch_size: int = 100, max_items: Optional[int] = None):
        self.batch_size = batch_size
        self.max_items = max_items

    def process_batch(self, items: List[dict]) -> List[dict]:
        """
        Process a batch of items.

        Args:
            items: List of items to process

        Returns:
            List of processed items
        """
        return [self._process_item(item) for item in items]
```

***

### **Testing**

* Write unit tests for all components
* Use **pytest** for running tests
* Maintain high test coverage

**Example:**

```python
import pytest
from agents.services.agent_service import AgentService

@pytest.fixture
def agent_service():
    return AgentService()

@pytest.mark.asyncio
async def test_agent_creation(agent_service):
    data = AgentCreate(
        name="test_agent",
        description="Test agent",
        mode="ReAct"
    )

    agent = await agent_service.create_agent(data)
    assert agent.name == "test_agent"
    assert agent.mode == "ReAct"
```

***

### **Error Handling**

Implement robust error handling across all services.

```python
from agents.exceptions import AgentError

class AgentService:
    async def create_agent(self, data: AgentCreate) -> AgentModel:
        try:
            agent = await AgentModel.create(**data.dict())
            return agent
        except Exception as e:
            raise AgentError(f"Failed to create agent: {str(e)}")
```

***

### **Database Management**

#### **Migrations (Alembic)**

```bash
# Create a new migration
alembic revision --autogenerate -m "Add agent table"

# Run migrations
alembic upgrade head
```

#### **Models**

```python
from sqlalchemy import Column, String, Integer
from agents.models.base import Base

class Agent(Base):
    __tablename__ = "agents"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    description = Column(String)
    mode = Column(String, nullable=False)
```

***

### **Deployment**

#### **Docker**

**Build the image:**

```bash
docker build -t nuvex .
```

**Run the container:**

```bash
docker run -p 8000:8000 \
    -e DATABASE_URL=postgresql://user:password@host:5432/nuvex \
    -e JWT_SECRET=your_secret \
    nuvex
```

***

### **Production Considerations**

#### **Logging**

```python
import logging

logger = logging.getLogger(__name__)

class AgentService:
    async def create_agent(self, data: AgentCreate):
        logger.info(f"Creating agent: {data.name}")
        try:
            agent = await AgentModel.create(**data.dict())
            logger.info(f"Agent created successfully: {agent.id}")
            return agent
        except Exception as e:
            logger.error(f"Failed to create agent: {str(e)}")
            raise
```

#### **Monitoring**

```python
from prometheus_client import Counter, Histogram

request_count = Counter('agent_requests_total', 'Total agent requests')

request_latency = Histogram('agent_request_latency_seconds', 'Agent request latency')
```

***

### **Contributing**

1. Fork the repository
2. Create a feature branch
3. Write tests
4. Submit a pull request

#### **Pull Request Guidelines**

* Follow the code style guide
* Include tests
* Update docs if needed
* Add a changelog entry if relevant

***

### **Troubleshooting**

#### **Database Connection Issues**

```python
try:
    await database.connect()
except Exception as e:
    logger.error(f"Database connection failed: {str(e)}")
    sys.exit(1)
```

#### **Authentication Problems**

```python
if token_error:
    logger.warning(f"Authentication failed for user: {user_id}")
    raise AuthenticationError("Invalid token")
```

***

### **Debugging**

Enable detailed debug logs:

```python
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

logger.debug("Detailed debugging information")
```

***

### **Performance Optimization**

#### **Connection Pooling**

```python
from databases import Database

database = Database(
    DATABASE_URL,
    min_size=5,
    max_size=20
)
```

#### **Caching**

```python
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend

@router.get("/agent/{agent_id}")
@cache(expire=60)
async def get_agent(agent_id: str):
    return await agent_service.get_agent(agent_id)
```

***

### **Security Best Practices**

#### **Input Validation**

```python
from pydantic import BaseModel, validator

class UserInput(BaseModel):
    name: str
    email: str

    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email')
        return v
```

#### **Rate Limiting**

```python
from fastapi import Depends
from agents.middleware.rate_limit import rate_limit

@router.post("/agent/create")
@rate_limit(max_requests=10, window_seconds=60)
async def create_agent(data: AgentCreate):
    return await agent_service.create_agent(data)
```