Token Creations

Comprehensive token creation events across all supported protocols and platforms on Solana, including SPL tokens, Token-2022 standard, and NFT collections.

Overview

The token creations WebSocket stream provides real-time notifications for:

• SPL token deployments
• Token-2022 creations with extensions
• NFT collection launches
• Metadata updates and verification
• Authority configurations
• Token extension activations

Connection

WebSocket Endpoint:

wss://api.nolimitnodes.com/socket?apikey=YOUR_API_KEY

Subscription

Subscribe Request

{
    "method": "dataSubscribe",
    "params": {
        "referenceId": "TOKEN_CREATIONS",
        "streams": [
            {
                "stream": "tokenCreationsSubscribe",
                "params": {
                    "creation_type": "mint",
                    "program_id": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
                    "standard": "spl-token-2022"
                }
            }
        ]
    },
    "id": 1
}

Filter Parameters

Parameter	Type	Required	Description
mint_address	string	No	Filter by specific token mint address
creation_type	string	No	Filter by creation type (e.g., "spl_token_initialize_mint2_inner")
program_id	string	No	Filter by program ID (e.g., "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
decimals	number	No	Filter by token decimals

Response Messages

Subscription Confirmation

{
    "status": "Subscribed to tokenCreations",
    "subscription_id": "77ea428b-8572-4efb-8d3d-492454631d32",
    "stream": "tokenCreationsSubscribe",
    "reference_id": "TOKEN_CREATIONS"
}

Token Creation Event

{
    "method": "tokenCreationsSubscribe",
    "subscription_id": "77ea428b-8572-4efb-8d3d-492454631d32",
    "result": {
        "mint_address": "Dwzu6sNdC2VafKMN9cPHUZWGfT5tfX1mkqvCSBDZpump",
        "decimals": 6,
        "mint_authority": "TSLvdd1pWpHVjahSpsvCXUbgwsL3JAcvokwaKt1eokM",
        "freeze_authority": null,
        "creation_type": "spl_token_initialize_mint2_inner",
        "program_id": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
        "block_num": 366363984,
        "block_time": 1757694473,
        "transaction_signature": "2QdTKjKCLy8bTeXDNAAnFY7HWVeDYtagbZ16Fia1rh12ThtXL7ictXNiVjrfmJLzBhyjiL9oxk9J78asJNK5DHao"
    }
}

Field Descriptions

• mint_address: Token mint address
• decimals: Token decimal places
• mint_authority: Address that can mint tokens (null if renounced)
• freeze_authority: Address that can freeze accounts (null if renounced)
• creation_type: Type of creation event
• program_id: Token program ID
• block_num: Block number
• block_time: Unix timestamp
• transaction_signature: Solana transaction signature

Token Information

• mint_address: New token mint address
• decimals: Number of decimal places for the token
• creation_type: Type of token creation (e.g., "spl_token_initialize_mint2_inner")
• program_id: Program ID that created the token (typically TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA)

Authorities

• mint_authority: Address that can mint new tokens (null if renounced)
• freeze_authority: Address that can freeze accounts (null if renounced)

Transaction Details

• transaction_signature: Solana transaction signature for the creation
• block_num: Block number where token was created
• block_time: Unix timestamp of the block

Unsubscribe

To stop receiving token creation events, use the dataUnsubscribe method:

Unsubscribe Request

{
    "method": "dataUnsubscribe",
    "params": {
        "subscriptionIds": ["your_subscription_id_here"],
        "referenceId": "UNSUB_REF"
    }
}

Unsubscribe Response

{
    "status": "Unsubscribed from tokenCreations",
    "subscription_id": "your_subscription_id_here",
    "stream": "tokenCreationsSubscribe",
    "reference_id": "UNSUB_REF"
}

Code Examples

JavaScript/Node.js

const WebSocket = require('ws');

class TokenCreationMonitor {
    constructor(apiKey) {
        this.ws = new WebSocket(`wss://api.nolimitnodes.com/socket?apikey=${apiKey}`);
        this.setupHandlers();
    }

    setupHandlers() {
        this.ws.on('open', () => {
            this.subscribe();
        });

        this.ws.on('message', (data) => {
            const message = JSON.parse(data);
            if (message.method === 'tokenCreationsSubscribe') {
                this.processToken(message.result);
            }
        });
    }

    subscribe() {
        this.ws.send(JSON.stringify({
            method: 'dataSubscribe',
            params: {
                referenceId: 'TOKEN_CREATIONS',
                streams: [{
                    stream: 'tokenCreationsSubscribe',
                    params: {}
                }]
            }
        }));
    }

    processToken(token) {
        console.log('\nNew Token Created:');
        console.log(`  Mint: ${token.mint_address}`);
        console.log(`  Decimals: ${token.decimals}`);
        console.log(`  Mint Authority: ${token.mint_authority || 'Renounced'}`);
        console.log(`  Freeze Authority: ${token.freeze_authority || 'Renounced'}`);
        console.log(`  Block: ${token.block_num}`);
    }
}

Python

import json
import asyncio
import websockets
from enum import Enum
from typing import Dict, List

class TokenStandard(Enum):
    SPL = "spl-token"
    SPL2022 = "spl-token-2022"

class TokenCreationMonitor:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.uri = f"wss://api.nolimitnodes.com/pump-fun?apikey={api_key}"
        self.token_database = []
        
    async def connect(self):
        self.websocket = await websockets.connect(self.uri)
        print("Connected to token creations stream")
        
    async def subscribe(self, standard: str = None):
        params = {"creation_type": "mint"}
        if standard:
            params["standard"] = standard
            
        subscription = {
            "method": "dataSubscribe",
            "params": {
                "referenceId": "TOKENS#1",
                "streams": [{
                    "stream": "token-creationsSubscribe",
                    "params": params
                }]
            }
        }
        await self.websocket.send(json.dumps(subscription))
        
    async def listen(self):
        async for message in self.websocket:
            data = json.loads(message)
            if data.get('stream') == 'token-creations':
                await self.process_token(data['data'])
                
    async def process_token(self, token: Dict):
        # Store token data
        self.token_database.append(token)
        
        # Analyze token
        analysis = self.analyze_token(token)
        
        print(f"\n{'='*60}")
        print(f"🪙 NEW TOKEN: {token.get('metadata', {}).get('name', 'Unknown')}")
        print(f"{'='*60}")
        
        # Basic info
        print(f"Symbol: {token.get('metadata', {}).get('symbol', 'N/A')}")
        print(f"Mint: {token['mint'][:20]}...")
        print(f"Standard: {token['token']['standard']}")
        print(f"Decimals: {token['token']['decimals']}")
        
        # Supply info
        supply = token['token']['supply']
        print(f"\n📊 Supply:")
        print(f"  Type: {supply['type']}")
        print(f"  Initial: {self.format_amount(supply['initial'], token['token']['decimals'])}")
        if supply.get('max'):
            print(f"  Maximum: {self.format_amount(supply['max'], token['token']['decimals'])}")
        
        # Authorities
        auth = token['authorities']
        print(f"\n🔐 Authorities:")
        print(f"  Mint: {'🔓 Active' if auth.get('mintAuthority') else '🔒 Renounced'}")
        print(f"  Freeze: {'🔓 Active' if auth.get('freezeAuthority') else '🔒 Renounced'}")
        
        # Extensions
        if token.get('extensions'):
            print(f"\n🔧 Token-2022 Extensions ({len(token['extensions'])})")
            for ext in token['extensions']:
                print(f"  • {ext}")
                self.describe_extension(ext, token.get('extensionData', {}))
        
        # Risk assessment
        print(f"\n⚠️ Risk Assessment:")
        for risk in analysis['risks']:
            print(f"  • {risk}")
        
        if not analysis['risks']:
            print(f"  ✅ No significant risks detected")
        
        print(f"\nRisk Level: {'🔴' * analysis['risk_level']}{'⚪' * (5 - analysis['risk_level'])}")
        
    def analyze_token(self, token: Dict) -> Dict:
        risks = []
        risk_level = 0
        
        # Check authorities
        if token['authorities'].get('mintAuthority'):
            risks.append("Mint authority active - unlimited supply possible")
            risk_level += 2
            
        if token['authorities'].get('freezeAuthority'):
            risks.append("Freeze authority active - accounts can be frozen")
            risk_level += 1
        
        # Check extensions
        if 'permanent-delegate' in token.get('extensions', []):
            risks.append("Permanent delegate - irrevocable transfer authority")
            risk_level += 3
            
        if 'transfer-fees' in token.get('extensions', []):
            fee_data = token.get('extensionData', {}).get('transferFeeConfig', {})
            fee = fee_data.get('transferFeeBasisPoints', 0)
            if fee > 500:  # >5%
                risks.append(f"High transfer fee: {fee/100}%")
                risk_level += 2
        
        # Check metadata
        if not token.get('metadata', {}).get('uri'):
            risks.append("No metadata URI - possible scam token")
            risk_level += 1
        
        return {
            'risks': risks,
            'risk_level': min(risk_level, 5)
        }
    
    def describe_extension(self, extension: str, data: Dict):
        descriptions = {
            'transfer-fees': f"Transfer fee: {data.get('transferFeeConfig', {}).get('transferFeeBasisPoints', 0)/100}%",
            'interest-bearing': f"Interest rate: {data.get('interestRate', 0)}% APY",
            'permanent-delegate': f"Delegate: {data.get('permanentDelegate', 'Unknown')[:8]}...",
            'confidential-transfers': "Private balance transfers enabled",
            'default-account-state': "Accounts frozen by default",
            'memo-required': "Memo required for transfers"
        }
        
        if extension in descriptions:
            print(f"    → {descriptions[extension]}")
    
    def format_amount(self, amount: str, decimals: int) -> str:
        value = float(amount) / (10 ** decimals)
        return f"{value:,.2f}"

async def main():
    monitor = TokenCreationMonitor("YOUR_API_KEY")
    await monitor.connect()
    await monitor.subscribe()  # Subscribe to all token creations
    await monitor.listen()

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

Best Practices

1. Authority & Security Analysis

• Check Authority Status: Always verify mint and freeze authority settings (null = renounced = safer)
• Review Extensions: Avoid tokens with permanent delegates or high transfer fees (>5%)
• Validate Metadata: Fetch and verify token metadata from IPFS URI for authenticity

2. Creator Due Diligence

• Research Creator History: Track wallet history for previous token launches and success rates
• Identify Red Flags: Flag prolific creators (>10 tokens) or high-frequency launches (>5 per day)
• Check Social Presence: Verify social links and community engagement levels

3. Risk Management

• Monitor Supply Mechanics: Understand total supply, initial distribution, and inflation patterns
• Set Creation Alerts: Track new tokens by specific creators or with certain characteristics
• Filter by Standards: Focus on SPL Token-2022 for advanced features and better security

Related Documentation

• PumpFun Token Creation
• Liquidity Pools
• Smart Contract Interactions