Python

CGPT SDK Docs Python

Smart Contract Generator SDK Documentation

The ChainGPT Smart Contract Generator allows you to generate Solidity smart contracts based on textual descriptions or prompts. It supports both immediate (buffered) responses and streaming for real-time code generation, along with comprehensive chat history management for maintaining conversation context across sessions.

Table of Contents

1. Installation

2. Quick Start

3. Initialization and Service Access

4. SmartContractService Methods

   • generate_contract()

   • stream_contract()

   • get_chat_history()

5. Request and Response Models

   • SmartContractGeneratorRequestModel

   • SmartContractGeneratorResponseModel

   • GetChatHistoryResponseModel

6. Enums and Types

7. Error Handling

8. Complete Examples

9. Best Practices

---

Installation

Install the ChainGPT SDK via pip:

pip install chaingpt

Or add to your requirements.txt:

chaingpt>=1.1.3

---

Quick Start

import asyncio
import os
from chaingpt.client import ChainGPTClient
from chaingpt.models.smart_contract import SmartContractGeneratorRequestModel
from chaingpt.types import ChatHistoryMode

async def quick_example():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    # Generate a simple ERC20 token
    request = SmartContractGeneratorRequestModel(
        question="Create an ERC20 token named 'MyToken' with symbol 'MTK'"
    )
    
    response = await client.smart_contract.generate_contract(request)
    print(f"Generated Contract:\n{response.data.bot}")
    
    await client.close()

asyncio.run(quick_example())

---

Initialization and Service Access

Basic Setup

import asyncio
import os
from chaingpt.client import ChainGPTClient
from chaingpt.models.smart_contract import (
    SmartContractGeneratorRequestModel
)
from chaingpt.models.chat_history import GetChatHistoryResponseModel
from chaingpt.types import ChatHistoryMode
from chaingpt.exceptions import ChainGPTError, APIError, ValidationError

# Load API key from environment variable (recommended)
API_KEY = os.getenv("CHAINGPT_API_KEY")

async def main():
    # Initialize the client
    client = ChainGPTClient(api_key=API_KEY)
    
    # Access the Smart Contract service
    smart_contract_service = client.smart_contract
    
    # Your smart contract operations here
    
    # Always close the client when done
    await client.close()

if __name__ == "__main__":
    if not API_KEY:
        print("Please set your CHAINGPT_API_KEY environment variable")
        exit(1)
    
    asyncio.run(main())

Environment Setup (Optional)

For development, you can use a .env file:

from dotenv import load_dotenv
load_dotenv()

API_KEY = os.getenv("CHAINGPT_API_KEY")

---

SmartContractService Methods

The SmartContractService provides three main methods for smart contract generation and history management.

generate_contract()

Generates a complete smart contract based on your prompt and returns the full response immediately.

Method Signature:

async def generate_contract(
    self, 
    request_data: SmartContractGeneratorRequestModel
) -> SmartContractGeneratorResponseModel

Parameters:

• request_data (SmartContractGeneratorRequestModel): The request configuration containing the prompt and settings

Returns:

• SmartContractGeneratorResponseModel: Complete response with generated contract code

Raises:

• TypeError: If request_data is not a SmartContractGeneratorRequestModel instance

• ValueError: If the model field is not 'smart_contract_generator'

• APIError: For API-related errors

• ValidationError: For request validation errors

Example:

async def generate_basic_contract():
    client = ChainGPTClient(api_key=API_KEY)
    
    try:
        request = SmartContractGeneratorRequestModel(
            question="Create a simple ERC20 token with name 'TestToken', symbol 'TST', and initial supply of 1000000",
            chatHistory=ChatHistoryMode.OFF
        )
        
        response = await client.smart_contract.generate_contract(request)
        
        if response.status:
            print(f"Status: {response.message}")
            print(f"User Prompt: {response.data.user}")
            print(f"Generated Contract:\n{response.data.bot}")
        else:
            print(f"Generation failed: {response.message}")
            
    except ValidationError as e:
        print(f"Validation error: {e.message}")
    except APIError as e:
        print(f"API error: {e.status_code} - {e.message}")
    finally:
        await client.close()

stream_contract()

Generates a smart contract and streams the response in real-time, allowing you to display the generated code as it's being created.

Method Signature:

async def stream_contract(
    self, 
    request_data: SmartContractGeneratorRequestModel
) -> AsyncIterator[StreamChunk]

Parameters:

• request_data (SmartContractGeneratorRequestModel): The request configuration

Returns:

• AsyncIterator[StreamChunk]: Async iterator yielding chunks of generated content

Yields:

• StreamChunk: Raw bytes that need to be decoded (typically UTF-8)

Raises:

• Same exceptions as generate_contract()

• StreamingError: For streaming-specific errors

Example:

async def stream_contract_example():
    client = ChainGPTClient(api_key=API_KEY)
    session_id = "550e8400-e29b-41d4-a716-446655440000"  # UUID format recommended
    
    try:
        request = SmartContractGeneratorRequestModel(
            question="Create an ERC721 NFT contract with minting functionality",
            chatHistory=ChatHistoryMode.ON,
            sdkUniqueId=session_id
        )
        
        print("Streaming contract generation:")
        full_contract = ""
        
        async for chunk in client.smart_contract.stream_contract(request):
            decoded_chunk = chunk.decode('utf-8')
            print(decoded_chunk, end="", flush=True)
            full_contract += decoded_chunk
            
        print("\n--- Streaming Complete ---")
        # full_contract now contains the complete generated code
        
    except Exception as e:
        print(f"Streaming error: {e}")
    finally:
        await client.close()

get_chat_history()

Retrieves the chat history for smart contract generation sessions, including past prompts and generated contracts.

Method Signature:

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

Parameters:

• limit (Optional[int]): Maximum number of entries to return (default: 10)

• offset (Optional[int]): Number of entries to skip for pagination (default: 0)

• sort_by (Optional[str]): Field to sort by (default: "createdAt")

• sort_order (Optional[str]): Sort order - "asc" or "desc" (default: "desc")

Returns:

• GetChatHistoryResponseModel: Response containing history entries and metadata

Example:

async def get_history_example():
    client = ChainGPTClient(api_key=API_KEY)
    
    try:
        history = await client.smart_contract.get_chat_history(
            limit=5,
            sort_order="desc"
        )
        
        if history.statusCode == 200:
            print(f"Total entries: {history.data.count}")
            
            for entry in history.data.rows:
                print(f"\nEntry ID: {entry.id}")
                print(f"Created: {entry.createdAt}")
                print(f"Session ID: {entry.sdkUniqueId or 'N/A'}")
                print(f"Bot: {entry.chatBot.name}")
                print(f"User: {entry.user.email}")
                print(f"Question: {entry.question[:100]}...")
                print(f"Answer Preview: {entry.answer[:150]}...")
                
                # Additional contract-specific fields
                if entry.contractAddress:
                    print(f"Contract Address: {entry.contractAddress}")
                if entry.chainId:
                    print(f"Chain ID: {entry.chainId}")
                    
        else:
            print(f"Failed to retrieve history: {history.message}")
            
    except APIError as e:
        print(f"API error: {e.status_code} - {e.message}")
    finally:
        await client.close()

---

Request and Response Models

SmartContractGeneratorRequestModel

The request model for smart contract generation.

Fields:

• model (Literal["smart_contract_generator"]): Required. Must be "smart_contract_generator"

• question (str): Required. Your prompt describing the contract (min length: 1)

• chatHistory (Optional[ChatHistoryMode]): Chat history mode (default: ChatHistoryMode.OFF)

• sdkUniqueId (Optional[str]): Session identifier for chat history (1-100 characters, required if chatHistory is ON)

Example:

request = SmartContractGeneratorRequestModel(
    model="smart_contract_generator",  # Auto-set, but can be explicit
    question="Create a multi-signature wallet contract with 3 owners requiring 2 signatures",
    chatHistory=ChatHistoryMode.ON,
    sdkUniqueId="my-session-123"
)

SmartContractGeneratorResponseModel

The response model for contract generation.

Fields:

• status (bool): Success indicator

• statusCode (Optional[int]): HTTP status code

• message (Optional[str]): Response message

• data (SmartContractDataModel): Generated contract data

  • user (Optional[str]): The processed user prompt

  • bot (Optional[str]): The generated Solidity contract code

GetChatHistoryResponseModel

The response model for chat history retrieval.

Fields:

• status (Optional[bool]): Success indicator

• statusCode (Optional[int]): HTTP status code

• message (str): Response message

• data (ChatHistoryDataModel): History data

  • count (int): Total available entries

  • rows (List[ChatHistoryEntryModel]): List of history entries

ChatHistoryEntryModel Fields:

• id (str): Entry identifier

• question (str): User's prompt

• answer (str): AI's response (contract code)

• createdAt (str): Creation timestamp

• type (str): Entry type

• chatBot (ChatBotModel): Bot information

• user (UserModel): User information

• sdkUniqueId (Optional[str]): Session identifier

• contractAddress (Optional[str]): Deployed contract address

• chainId (Optional[str]): Blockchain chain ID

• auditType (Optional[str]): Audit type if applicable

• auditMethod (Optional[str]): Audit method if applicable

• promptId (Optional[str]): Prompt identifier

• chartData (Optional[Any]): Chart data if applicable

• chartType (Optional[str]): Chart type if applicable

• chartInterval (Optional[str]): Chart interval if applicable

• trend (Optional[str]): Trend information if applicable

• chatFeedbacks (List[Any]): Feedback entries

---

Enums and Types

ChatHistoryMode

Controls chat history behavior:

• ChatHistoryMode.ON ("on"): Enables session-based chat history

• ChatHistoryMode.OFF ("off"): Disables chat history

StreamChunk

Type for streaming data:

• StreamChunk = Union[str, bytes]: Raw chunk data from streaming responses

Usage Example

from chaingpt.types import ChatHistoryMode, StreamChunk

# Enable chat history for a session
request = SmartContractGeneratorRequestModel(
    question="Create a token contract",
    chatHistory=ChatHistoryMode.ON,
    sdkUniqueId="session-uuid"
)

# Process streaming chunks
async for chunk in client.smart_contract.stream_contract(request):
    decoded: str = chunk.decode('utf-8') if isinstance(chunk, bytes) else chunk
    print(decoded, end="")

---

Error Handling

The SDK provides comprehensive error handling with specific exception types:

Exception Hierarchy

• ChainGPTError: Base exception for all SDK errors

  • APIError: API-related errors with status codes

  • ValidationError: Request validation failures

  • AuthenticationError: Invalid API key (401)

  • InsufficientCreditsError: Account credit issues (402/403)

  • RateLimitError: Rate limiting (429)

  • NotFoundError: Endpoint not found (404)

  • ServerError: Server errors (5xx)

  • TimeoutError: Request timeouts

  • StreamingError: Streaming-specific errors

  • ConfigurationError: SDK configuration issues

Comprehensive Error Handling Example

from chaingpt.exceptions import (
    ChainGPTError, APIError, ValidationError, 
    AuthenticationError, InsufficientCreditsError,
    RateLimitError, StreamingError
)

async def robust_contract_generation():
    client = ChainGPTClient(api_key=API_KEY)
    
    try:
        request = SmartContractGeneratorRequestModel(
            question="Create a DeFi staking contract with rewards"
        )
        
        response = await client.smart_contract.generate_contract(request)
        return response.data.bot
        
    except ValidationError as e:
        print(f"Request validation failed: {e.message}")
        if e.field:
            print(f"Problem with field: {e.field}")
            
    except AuthenticationError as e:
        print(f"Authentication failed: {e.message}")
        print("Please check your API key")
        
    except InsufficientCreditsError as e:
        print(f"Insufficient credits: {e.message}")
        print("Please top up your account")
        
    except RateLimitError as e:
        print(f"Rate limit exceeded: {e.message}")
        if e.retry_after:
            print(f"Retry after {e.retry_after} seconds")
            
    except StreamingError as e:
        print(f"Streaming error: {e.message}")
        
    except APIError as e:
        print(f"API error {e.status_code}: {e.message}")
        if e.details:
            print(f"Details: {e.details}")
            
    except ChainGPTError as e:
        print(f"SDK error: {e.message}")
        if e.details:
            print(f"Details: {e.details}")
            
    except Exception as e:
        print(f"Unexpected error: {e}")
        
    finally:
        await client.close()

---

Complete Examples

Example 1: Session-Based Contract Development

import asyncio
import uuid
from chaingpt.client import ChainGPTClient
from chaingpt.models.smart_contract import SmartContractGeneratorRequestModel
from chaingpt.types import ChatHistoryMode

async def session_based_development():
    client = ChainGPTClient(api_key=API_KEY)
    session_id = str(uuid.uuid4())  # Generate unique session ID
    
    try:
        # Step 1: Create initial contract
        print("Step 1: Creating basic ERC20 token...")
        request1 = SmartContractGeneratorRequestModel(
            question="Create a basic ERC20 token named 'MyToken' with symbol 'MTK' and initial supply of 1,000,000 tokens",
            chatHistory=ChatHistoryMode.ON,
            sdkUniqueId=session_id
        )
        
        response1 = await client.smart_contract.generate_contract(request1)
        print("✓ Basic token created")
        
        # Step 2: Add functionality using chat history
        print("\nStep 2: Adding burn functionality...")
        request2 = SmartContractGeneratorRequestModel(
            question="Add a burn function to the previous token contract that allows token holders to burn their own tokens",
            chatHistory=ChatHistoryMode.ON,
            sdkUniqueId=session_id
        )
        
        response2 = await client.smart_contract.generate_contract(request2)
        print("✓ Burn functionality added")
        
        # Step 3: Add more features
        print("\nStep 3: Adding pause functionality...")
        request3 = SmartContractGeneratorRequestModel(
            question="Now add pause/unpause functionality that only the owner can control",
            chatHistory=ChatHistoryMode.ON,
            sdkUniqueId=session_id
        )
        
        print("Streaming the enhanced contract:")
        full_contract = ""
        async for chunk in client.smart_contract.stream_contract(request3):
            decoded_chunk = chunk.decode('utf-8')
            print(decoded_chunk, end="", flush=True)
            full_contract += decoded_chunk
        
        print("\n\n✓ Final contract with pause functionality completed")
        
        # Step 4: Review session history
        print("\nStep 4: Reviewing session history...")
        history = await client.smart_contract.get_chat_history(limit=10)
        
        session_entries = [
            entry for entry in history.data.rows 
            if entry.sdkUniqueId == session_id
        ]
        
        print(f"Found {len(session_entries)} entries for this session:")
        for i, entry in enumerate(session_entries, 1):
            print(f"  {i}. {entry.question[:80]}...")
            
    except Exception as e:
        print(f"Error during development: {e}")
    finally:
        await client.close()

Example 2: Batch Contract Generation

async def batch_contract_generation():
    client = ChainGPTClient(api_key=API_KEY)
    
    contract_requests = [
        "Create a simple ERC20 token with basic functionality",
        "Create an ERC721 NFT contract with metadata support",
        "Create a multi-signature wallet contract for 3 owners",
        "Create a decentralized voting contract",
        "Create a time-locked token vesting contract"
    ]
    
    generated_contracts = []
    
    try:
        for i, prompt in enumerate(contract_requests, 1):
            print(f"\nGenerating contract {i}/{len(contract_requests)}: {prompt[:50]}...")
            
            request = SmartContractGeneratorRequestModel(
                question=prompt,
                chatHistory=ChatHistoryMode.OFF  # Independent contracts
            )
            
            response = await client.smart_contract.generate_contract(request)
            
            if response.status and response.data.bot:
                generated_contracts.append({
                    'prompt': prompt,
                    'code': response.data.bot,
                    'user_processed': response.data.user
                })
                print(f"✓ Contract {i} generated successfully")
            else:
                print(f"✗ Contract {i} generation failed: {response.message}")
        
        print(f"\n🎉 Successfully generated {len(generated_contracts)} contracts")
        
        # Optionally save contracts to files
        for i, contract in enumerate(generated_contracts, 1):
            filename = f"contract_{i}.sol"
            with open(filename, 'w') as f:
                f.write(f"// Generated from prompt: {contract['prompt']}\n\n")
                f.write(contract['code'])
            print(f"Saved {filename}")
            
    except Exception as e:
        print(f"Batch generation error: {e}")
    finally:
        await client.close()

---

Best Practices

1. Session Management

• Use meaningful, unique session IDs (UUID format recommended)

• Enable chat history for iterative development

• Keep sessions focused on related functionality

import uuid

# Good: Meaningful session ID
session_id = f"token-dev-{uuid.uuid4()}"

# Better: Include timestamp for debugging
import datetime
session_id = f"contract-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:8]}"

2. Prompt Engineering

• Be specific about contract requirements

• Include token standards (ERC20, ERC721, etc.)

• Specify security features needed

• Mention deployment considerations

# Good prompt
good_prompt = """
Create an ERC20 token contract with the following specifications:
- Name: 'MyToken'
- Symbol: 'MTK'
- Initial supply: 1,000,000 tokens
- 18 decimals
- Include burn functionality
- Add pause/unpause capability for owner
- Include basic access control
- Follow OpenZeppelin standards for security
"""

# Basic prompt (less effective)
basic_prompt = "Create a token contract"

3. Error Handling Strategy

• Always wrap SDK calls in try-except blocks

• Handle specific exception types appropriately

• Log errors for debugging

• Implement retry logic for transient errors

import asyncio
from chaingpt.exceptions import RateLimitError, APIError

async def resilient_generation(request, max_retries=3):
    client = ChainGPTClient(api_key=API_KEY)
    
    for attempt in range(max_retries):
        try:
            return await client.smart_contract.generate_contract(request)
            
        except RateLimitError as e:
            if attempt < max_retries - 1 and e.retry_after:
                print(f"Rate limited, waiting {e.retry_after} seconds...")
                await asyncio.sleep(e.retry_after)
                continue
            raise
            
        except APIError as e:
            if attempt < max_retries - 1 and e.status_code >= 500:
                print(f"Server error, retrying in {2 ** attempt} seconds...")
                await asyncio.sleep(2 ** attempt)
                continue
            raise
            
    raise Exception("Max retries exceeded")

4. Resource Management

• Always close the client properly

• Use context managers for automatic cleanup

• Monitor API usage and credits

# Context manager approach (recommended)
async def safe_generation():
    async with ChainGPTClient(api_key=API_KEY) as client:
        request = SmartContractGeneratorRequestModel(
            question="Create a simple token contract"
        )
        response = await client.smart_contract.generate_contract(request)
        return response

# Manual cleanup (ensure in finally block)
async def manual_cleanup():
    client = ChainGPTClient(api_key=API_KEY)
    try:
        # Your operations here
        pass
    finally:
        await client.close()

5. Code Organization

• Separate contract generation logic into dedicated functions

• Use configuration objects for repeated settings

• Implement proper logging

import logging
from dataclasses import dataclass
from typing import Optional

@dataclass
class ContractConfig:
    """Configuration for contract generation"""
    enable_history: bool = True
    session_id: Optional[str] = None
    streaming: bool = False
    
    def to_request_params(self):
        return {
            'chatHistory': ChatHistoryMode.ON if self.enable_history else ChatHistoryMode.OFF,
            'sdkUniqueId': self.session_id
        }

class SmartContractGenerator:
    def __init__(self, api_key: str):
        self.client = ChainGPTClient(api_key=api_key)
        self.logger = logging.getLogger(__name__)
    
    async def generate(self, prompt: str, config: ContractConfig = None) -> str:
        config = config or ContractConfig()
        
        request = SmartContractGeneratorRequestModel(
            question=prompt,
            **config.to_request_params()
        )
        
        self.logger.info(f"Generating contract with prompt: {prompt[:100]}...")
        
        if config.streaming:
            return await self._stream_generate(request)
        else:
            return await self._buffer_generate(request)
    
    async def _buffer_generate(self, request):
        response = await self.client.smart_contract.generate_contract(request)
        return response.data.bot
    
    async def _stream_generate(self, request):
        full_contract = ""
        async for chunk in self.client.smart_contract.stream_contract(request):
            full_contract += chunk.decode('utf-8')
        return full_contract
    
    async def close(self):
        await self.client.close()

This comprehensive documentation should provide developers with everything they need to effectively use the ChainGPT Smart Contract Generator SDK, from basic usage to advanced patterns and best practices.