# Development Guide

Development Environment Setup

#### Prerequisites

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

<figure><img src="/files/9RF0EvjEdKD4izVLFjA3" alt=""><figcaption></figcaption></figure>

#### Local Development Setup

1. Clone the repository:

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

2. Install dependencies:

```
poetry install
```

3. Set up environment variables:

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

4. Configure your `.env` file:

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

5. Start the development server:

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

### Project Structure

```
deepcore/
├── 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

The agent system is the heart of DeepCore. It consists of several key components:

1. **Agent Core**

```
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:
        # Implement agent processing logic
        response = await self._think(input_data)
        return response
​
    async def _think(self, input_data: str) -> str:
        # Implement thinking logic
        pass
```

2. **Tool Integration**

```
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:
        # Implement tool logic
        result = await self.process_data(input_data)
        return {"result": result}
​
    async def process_data(self, data: dict) -> str:
        # Implement specific data processing
        pass
```

#### API Layer

The API layer is built using FastAPI:

```
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

Services handle business logic:

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

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

    async def _initialize_tools(self, agent: AgentModel):
        # Tool initialization logic
        pass
```

### Development Guidelines

#### Code Style

1. Follow PEP 8 guidelines
2. Use type hints
3. Document functions and classes
4. Write meaningful commit messages

Example:

```
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

1. Write unit tests for all components
2. Use pytest for testing
3. Maintain high test coverage

Example:

```
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 proper error handling:

```
from agents.exceptions import AgentError

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

### Database Management

#### Migrations

Use Alembic for database migrations:

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

# Run migrations
alembic upgrade head
```

#### Models

Define SQLAlchemy models:

```
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 Deployment

1. Build the image:

```
docker build -t deepcore .
```

2. Run the container:

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

#### Production Considerations

1. Use proper logging:

```
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
```

2. Implement monitoring:

```
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 documentation
* Add changelog entry

### Troubleshooting

#### Common Issues

1. Database Connection Issues:

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

2. Authentication Problems:

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

#### Debugging

Enable debug logging:

```
import logging

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

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

### Performance Optimization

1. Use connection pooling:

```
from databases import Database

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

2. Implement caching:

```
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

1. Input validation:

```
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
```

2. Rate limiting:

```
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)
```


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.deepcore.top/developer-guide/openapi.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.