Python

CGPT SDK Docs

ChainGPT Smart Contract Auditor Python SDK Documentation

The ChainGPT Smart Contract Auditor provides automated security vulnerability analysis for smart contracts. It offers comprehensive audit reports with support for both immediate and streaming responses, plus session-based history tracking.

Table of Contents

  1. Installation

  2. Quick Start

  3. Authentication & Client Setup

  4. AuditorService Methods

  5. Request Models & Parameters

  6. Response Models

  7. Error Handling

  8. Best Practices

  9. Complete Examples

Installation

pip install chaingpt

Requirements:

  • Python 3.8+

  • Valid ChainGPT API key

Quick Start

import asyncio
from chaingpt.client import ChainGPTClient
from chaingpt.models.auditor import SmartContractAuditRequestModel
from chaingpt.types import ChatHistoryMode

async def quick_audit():
    client = ChainGPTClient(api_key="your-api-key")
    
    # Simple contract audit
    request = SmartContractAuditRequestModel(
        question="Please audit this Solidity contract: contract Test { ... }"
    )
    
    response = await client.auditor.audit_contract(request)
    print(response.data.bot)  # The audit report
    
    await client.close()

asyncio.run(quick_audit())

Authentication & Client Setup

import os
from chaingpt.client import ChainGPTClient

# Recommended: Use environment variables
API_KEY = os.getenv("CHAINGPT_API_KEY")

# Initialize client
client = ChainGPTClient(
    api_key=API_KEY,
    base_url="https://api.chaingpt.org",  # Default
    timeout=30.0,  # Default request timeout
    stream_timeout=300.0  # Default streaming timeout
)

# Access the auditor service
auditor = client.auditor

Environment Setup:

# Create .env file
echo "CHAINGPT_API_KEY=your_actual_api_key_here" > .env

AuditorService Methods

1. audit_contract() - Complete Audit Response

Returns the full audit report in a single response.

async def audit_contract(
    self, 
    request_data: SmartContractAuditRequestModel
) -> LLMResponseModel

When to use:

  • You want the complete audit report at once

  • Working with shorter contracts

  • Building batch processing systems

Example:

request = SmartContractAuditRequestModel(
    question=f"Audit this contract for security issues:\n```solidity\n{contract_code}\n```",
    chatHistory=ChatHistoryMode.OFF
)

response = await client.auditor.audit_contract(request)
if response.status:
    print("Audit Report:")
    print(response.data.bot)

2. stream_audit() - Streaming Audit Response

Streams the audit report progressively as it's generated.

async def stream_audit(
    self, 
    request_data: SmartContractAuditRequestModel
) -> AsyncIterator[StreamChunk]

When to use:

  • Large contracts with lengthy audit reports

  • Real-time UI updates

  • Progressive result display

Example:

request = SmartContractAuditRequestModel(
    question="Audit this complex DeFi contract...",
    chatHistory=ChatHistoryMode.ON,
    sdkUniqueId="audit-session-123"
)

print("Streaming audit results:")
async for chunk in client.auditor.stream_audit(request):
    # Decode bytes to string
    text = chunk.decode('utf-8')
    print(text, end='', flush=True)

3. get_audit_history() - Session History

Retrieves past audit conversations for a session.

async def get_audit_history(
    self,
    limit: Optional[int] = 10,
    offset: Optional[int] = 0,
    sort_by: Optional[str] = "createdAt",
    sort_order: Optional[str] = "desc",
    sdk_unique_id: Optional[str] = None,
) -> GetChatHistoryResponseModel

Parameters:

  • limit: Max entries to return (default: 10)

  • offset: Pagination offset (default: 0)

  • sort_by: Sort field (default: "createdAt")

  • sort_order: "asc" or "desc" (default: "desc")

  • sdk_unique_id: Required for session-specific history

Example:

history = await client.auditor.get_audit_history(
    sdk_unique_id="audit-session-123",
    limit=5
)

print(f"Found {history.data.count} audit entries:")
for entry in history.data.rows:
    print(f"- {entry.createdAt}: {entry.question[:100]}...")

Request Models & Parameters

SmartContractAuditRequestModel

class SmartContractAuditRequestModel(BaseModel):
    model: Literal["smart_contract_auditor"] = "smart_contract_auditor"
    question: str  # Required, min_length=1
    chatHistory: Optional[ChatHistoryMode] = ChatHistoryMode.OFF
    sdkUniqueId: Optional[str] = None  # Required if chatHistory=ON

Field Details:

Field
Type
Required
Description

model

Literal

No

Always "smart_contract_auditor" (auto-set)

question

str

Yes

Contract code + audit instructions

chatHistory

enum

No

ON or OFF (default: OFF)

sdkUniqueId

str

Conditional

Required if chatHistory=ON, 1-100 chars

Crafting Effective Questions:

# ❌ Too vague
question = "Check this contract"

# ✅ Specific and detailed
question = """
Please perform a comprehensive security audit on this Solidity contract:

```solidity
pragma solidity ^0.8.0;
contract MyToken {
    // contract code here
}

Focus on:

  1. Reentrancy vulnerabilities

  2. Integer overflow/underflow

  3. Access control issues

  4. Gas optimization opportunities

  5. Best practice compliance """

Chat History & Sessions

Without History (Stateless):

request = SmartContractAuditRequestModel(
    question="Audit this contract...",
    chatHistory=ChatHistoryMode.OFF
)

With History (Session-based):

session_id = "audit-session-" + str(uuid.uuid4())

# First audit
request1 = SmartContractAuditRequestModel(
    question="Audit this ERC20 contract...",
    chatHistory=ChatHistoryMode.ON,
    sdkUniqueId=session_id
)

# Follow-up question in same session
request2 = SmartContractAuditRequestModel(
    question="Can you explain the reentrancy issue in more detail?",
    chatHistory=ChatHistoryMode.ON,
    sdkUniqueId=session_id
)

Response Models

LLMResponseModel (for audit results)

class LLMResponseModel:
    status: bool              # Success indicator
    statusCode: Optional[int] # HTTP status
    message: str             # Response message
    data: LLMResponseDataModel

class LLMResponseDataModel:
    bot: str                 # The actual audit report

GetChatHistoryResponseModel (for history)

class GetChatHistoryResponseModel:
    status: Optional[bool]
    statusCode: Optional[int]
    message: str
    data: ChatHistoryDataModel

class ChatHistoryDataModel:
    count: int                           # Total entries found
    rows: List[ChatHistoryEntryModel]    # History entries

class ChatHistoryEntryModel:
    id: str
    question: str      # Your original audit request
    answer: str        # The AI's audit report
    createdAt: str     # Timestamp
    sdkUniqueId: str   # Session ID
    # ... other fields

Error Handling

Exception Hierarchy

from chaingpt.exceptions import (
    ChainGPTError,           # Base exception
    APIError,                # API-related errors
    AuthenticationError,     # Invalid API key
    ValidationError,         # Invalid request data
    RateLimitError,         # Rate limit exceeded
    TimeoutError,           # Request timeout
)

Comprehensive Error Handling

import logging
from chaingpt.exceptions import *

async def robust_audit(contract_code: str):
    try:
        request = SmartContractAuditRequestModel(
            question=f"Audit: ```solidity\n{contract_code}\n```"
        )
        
        response = await client.auditor.audit_contract(request)
        
        if not response.status:
            logging.warning(f"Audit warning: {response.message}")
        
        return response.data.bot
        
    except AuthenticationError:
        logging.error("Invalid API key - check your credentials")
        raise
    
    except ValidationError as e:
        logging.error(f"Invalid request: {e.message}")
        if e.field:
            logging.error(f"Problem with field: {e.field}")
        raise
    
    except RateLimitError as e:
        logging.error(f"Rate limited. Retry after: {e.retry_after} seconds")
        raise
    
    except TimeoutError:
        logging.error("Request timed out - try with smaller contract or increase timeout")
        raise
    
    except APIError as e:
        logging.error(f"API error {e.status_code}: {e.message}")
        raise
    
    except ChainGPTError as e:
        logging.error(f"SDK error: {e.message}")
        if e.details:
            logging.error(f"Details: {e.details}")
        raise
    
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        raise

Best Practices

1. Resource Management

# ✅ Always use context manager or explicit cleanup
async with ChainGPTClient(api_key=API_KEY) as client:
    # Your audit code here
    pass
# Client automatically closed

# Or explicit cleanup:
client = ChainGPTClient(api_key=API_KEY)
try:
    # Your audit code here
    pass
finally:
    await client.close()

2. Session Management

import uuid

def generate_session_id() -> str:
    """Generate unique session ID for audit history"""
    return f"audit-{uuid.uuid4()}"

# Use consistent session IDs for related audits
session_id = generate_session_id()

3. Contract Preparation

def prepare_audit_request(contract_code: str, focus_areas: List[str] = None) -> str:
    """Prepare a well-structured audit request"""
    base_prompt = f"""
Please perform a comprehensive security audit on this Solidity contract:

```solidity
{contract_code}

"""
    if focus_areas:
        base_prompt += f"\nPlease focus on: {', '.join(focus_areas)}"
    
    return base_prompt

4. Streaming Best Practices

async def stream_with_timeout(request: SmartContractAuditRequestModel, timeout: int = 300):
    """Stream with custom timeout handling"""
    import asyncio
    
    async def stream_audit():
        result = ""
        async for chunk in client.auditor.stream_audit(request):
            result += chunk.decode('utf-8')
            yield result
    
    try:
        async for partial_result in asyncio.wait_for(stream_audit(), timeout=timeout):
            # Process partial results
            pass
    except asyncio.TimeoutError:
        print(f"Streaming timed out after {timeout} seconds")

Complete Examples

Example 1: Basic Contract Audit

import asyncio
import os
from chaingpt.client import ChainGPTClient
from chaingpt.models.auditor import SmartContractAuditRequestModel
from chaingpt.exceptions import ChainGPTError

async def basic_audit_example():
    API_KEY = os.getenv("CHAINGPT_API_KEY")
    
    contract_code = """
    pragma solidity ^0.8.0;
    
    contract SimpleBank {
        mapping(address => uint256) public balances;
        
        function deposit() public payable {
            balances[msg.sender] += msg.value;
        }
        
        function withdraw(uint256 amount) public {
            require(balances[msg.sender] >= amount, "Insufficient balance");
            balances[msg.sender] -= amount;
            payable(msg.sender).transfer(amount);
        }
    }
    """
    
    async with ChainGPTClient(api_key=API_KEY) as client:
        try:
            request = SmartContractAuditRequestModel(
                question=f"Please audit this contract for security vulnerabilities:\n\n```solidity\n{contract_code}\n```"
            )
            
            response = await client.auditor.audit_contract(request)
            
            print("=== AUDIT REPORT ===")
            print(response.data.bot)
            
        except ChainGPTError as e:
            print(f"Audit failed: {e}")

if __name__ == "__main__":
    asyncio.run(basic_audit_example())

Example 2: Session-Based Multi-Contract Audit

import asyncio
import uuid
from chaingpt.client import ChainGPTClient
from chaingpt.models.auditor import SmartContractAuditRequestModel
from chaingpt.types import ChatHistoryMode

async def session_audit_example():
    API_KEY = os.getenv("CHAINGPT_API_KEY")
    session_id = f"audit-{uuid.uuid4()}"
    
    contracts = {
        "Token": "contract Token { ... }",
        "Vault": "contract Vault { ... }",
        "Factory": "contract Factory { ... }"
    }
    
    async with ChainGPTClient(api_key=API_KEY) as client:
        # Audit each contract in the same session
        for name, code in contracts.items():
            print(f"\n=== Auditing {name} ===")
            
            request = SmartContractAuditRequestModel(
                question=f"Audit the {name} contract:\n```solidity\n{code}\n```",
                chatHistory=ChatHistoryMode.ON,
                sdkUniqueId=session_id
            )
            
            # Stream the response
            async for chunk in client.auditor.stream_audit(request):
                print(chunk.decode('utf-8'), end='', flush=True)
            print("\n")
        
        # Ask follow-up question about all contracts
        followup = SmartContractAuditRequestModel(
            question="Are there any security issues when these contracts interact with each other?",
            chatHistory=ChatHistoryMode.ON,
            sdkUniqueId=session_id
        )
        
        response = await client.auditor.audit_contract(followup)
        print("\n=== INTERACTION ANALYSIS ===")
        print(response.data.bot)
        
        # Get session history
        history = await client.auditor.get_audit_history(
            sdk_unique_id=session_id,
            limit=10
        )
        
        print(f"\n=== SESSION SUMMARY ===")
        print(f"Total interactions: {history.data.count}")
        for entry in history.data.rows:
            print(f"- {entry.createdAt}: {entry.question[:50]}...")

if __name__ == "__main__":
    asyncio.run(session_audit_example())

Example 3: Production-Ready Audit Service

import asyncio
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass
from chaingpt.client import ChainGPTClient
from chaingpt.models.auditor import SmartContractAuditRequestModel
from chaingpt.types import ChatHistoryMode
from chaingpt.exceptions import *

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class AuditResult:
    success: bool
    report: Optional[str] = None
    error: Optional[str] = None
    session_id: Optional[str] = None

class SmartContractAuditor:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client: Optional[ChainGPTClient] = None
    
    async def __aenter__(self):
        self.client = ChainGPTClient(api_key=self.api_key)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.client:
            await self.client.close()
    
    async def audit_contract(
        self, 
        contract_code: str,
        focus_areas: Optional[list] = None,
        use_session: bool = False,
        session_id: Optional[str] = None
    ) -> AuditResult:
        """
        Audit a smart contract with comprehensive error handling
        """
        if not self.client:
            return AuditResult(False, error="Client not initialized")
        
        try:
            # Prepare audit question
            question = self._prepare_question(contract_code, focus_areas)
            
            # Configure request
            chat_mode = ChatHistoryMode.ON if use_session else ChatHistoryMode.OFF
            
            request = SmartContractAuditRequestModel(
                question=question,
                chatHistory=chat_mode,
                sdkUniqueId=session_id if use_session else None
            )
            
            # Execute audit
            response = await self.client.auditor.audit_contract(request)
            
            if response.status:
                logger.info("Audit completed successfully")
                return AuditResult(
                    success=True,
                    report=response.data.bot,
                    session_id=session_id
                )
            else:
                logger.warning(f"Audit completed with warnings: {response.message}")
                return AuditResult(
                    success=False,
                    error=response.message
                )
        
        except AuthenticationError:
            error_msg = "Authentication failed - check API key"
            logger.error(error_msg)
            return AuditResult(False, error=error_msg)
        
        except ValidationError as e:
            error_msg = f"Invalid request: {e.message}"
            logger.error(error_msg)
            return AuditResult(False, error=error_msg)
        
        except RateLimitError as e:
            error_msg = f"Rate limit exceeded. Retry after {e.retry_after}s"
            logger.error(error_msg)
            return AuditResult(False, error=error_msg)
        
        except Exception as e:
            error_msg = f"Unexpected error: {str(e)}"
            logger.error(error_msg)
            return AuditResult(False, error=error_msg)
    
    def _prepare_question(self, contract_code: str, focus_areas: Optional[list]) -> str:
        """Prepare a comprehensive audit question"""
        question = f"""
Please perform a comprehensive security audit on this Solidity contract:

```solidity
{contract_code}

Please analyze for:

- Security vulnerabilities (reentrancy, overflow, access control)
- Gas optimization opportunities
- Code quality and best practices
- Compliance with latest Solidity standards
"""

        if focus_areas:
            question += f"\nPay special attention to: {', '.join(focus_areas)}"
        
        return question

    async def get_session_summary(self, session_id: str) -> Dict[str, Any]:
        """Get summary of audit session"""
        if not self.client:
            return {"error": "Client not initialized"}

        try:
            history = await self.client.auditor.get_audit_history(
                sdk_unique_id=session_id,
                limit=50
            )
            
            return {
                "session_id": session_id,
                "total_audits": history.data.count,
                "entries": [
                    {
                        "timestamp": entry.createdAt,
                        "question_preview": entry.question[:100] + "...",
                        "report_preview": entry.answer[:200] + "..."
                    }
                    for entry in history.data.rows
                ]
            }
        
        except Exception as e:
            return {"error": str(e)}

# Usage example

async def main():
    API_KEY = os.getenv("CHAINGPT_API_KEY")

    contract = """
    pragma solidity ^0.8.0;
    contract MyContract {
        // Your contract code here
    }
    """
    
    async with SmartContractAuditor(API_KEY) as auditor:
        result = await auditor.audit_contract(
            contract_code=contract,
            focus_areas=["reentrancy", "access control"],
            use_session=True,
            session_id="production-audit-001"
        )
        
        if result.success:
            print("Audit Report:")
            print(result.report)
        else:
            print(f"Audit failed: {result.error}")

if __name__ == "__main__":
    asyncio.run(main())

This documentation provides a comprehensive guide for developers to effectively use the ChainGPT Smart Contract Auditor SDK with practical examples, best practices, and robust error handling patterns.

Last updated

Was this helpful?