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
Installation
Quick Start
Initialization and Service Access
SmartContractService Methods
generate_contract()
stream_contract()
get_chat_history()
Request and Response Models
SmartContractGeneratorRequestModel
SmartContractGeneratorResponseModel
GetChatHistoryResponseModel
Enums and Types
Error Handling
Complete Examples
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 instanceValueError
: If the model field is not 'smart_contract_generator'APIError
: For API-related errorsValidationError
: 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 indicatorstatusCode
(Optional[int]): HTTP status codemessage
(Optional[str]): Response messagedata
(SmartContractDataModel): Generated contract datauser
(Optional[str]): The processed user promptbot
(Optional[str]): The generated Solidity contract code
GetChatHistoryResponseModel
The response model for chat history retrieval.
Fields:
status
(Optional[bool]): Success indicatorstatusCode
(Optional[int]): HTTP status codemessage
(str): Response messagedata
(ChatHistoryDataModel): History datacount
(int): Total available entriesrows
(List[ChatHistoryEntryModel]): List of history entries
ChatHistoryEntryModel Fields:
id
(str): Entry identifierquestion
(str): User's promptanswer
(str): AI's response (contract code)createdAt
(str): Creation timestamptype
(str): Entry typechatBot
(ChatBotModel): Bot informationuser
(UserModel): User informationsdkUniqueId
(Optional[str]): Session identifiercontractAddress
(Optional[str]): Deployed contract addresschainId
(Optional[str]): Blockchain chain IDauditType
(Optional[str]): Audit type if applicableauditMethod
(Optional[str]): Audit method if applicablepromptId
(Optional[str]): Prompt identifierchartData
(Optional[Any]): Chart data if applicablechartType
(Optional[str]): Chart type if applicablechartInterval
(Optional[str]): Chart interval if applicabletrend
(Optional[str]): Trend information if applicablechatFeedbacks
(List[Any]): Feedback entries
Enums and Types
ChatHistoryMode
Controls chat history behavior:
ChatHistoryMode.ON
("on"): Enables session-based chat historyChatHistoryMode.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 errorsAPIError
: API-related errors with status codesValidationError
: Request validation failuresAuthenticationError
: 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 timeoutsStreamingError
: Streaming-specific errorsConfigurationError
: 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.
Last updated
Was this helpful?