aptos-mcp

Instructions

Creates a new Aptos indexer project based on the example processor.

Args:
    project_name: Name of the indexer project
    processor_type: Type of processor (transaction, event)

Implementation Reference

• aptos_mcp_server.py:989-1128 (handler)

  The @mcp.tool()-decorated handler function that implements the core logic for creating a new Aptos indexer project. It clones a GitHub template, customizes package.json, generates processor TypeScript code based on transaction or event type, creates README and .env files, and sets up the project structure.

  @mcp.tool()
  async def create_aptos_indexer(project_name: str, processor_type: str = "transaction") -> str:
      """
      Creates a new Aptos indexer project based on the example processor.
      
      Args:
          project_name: Name of the indexer project
          processor_type: Type of processor (transaction, event)
      """
      project_name = project_name.strip().replace(" ", "-").lower()
      
      supported_types = ["transaction", "event"]
      if processor_type not in supported_types:
          return f"Unsupported processor type. Choose from: {', '.join(supported_types)}"
      
      try:
          # Create project directory
          project_dir = os.path.join(os.getcwd(), project_name)
          if os.path.exists(project_dir):
              return f"Directory {project_dir} already exists. Please choose a different name."
          
          os.makedirs(project_dir)
          
          # Clone the example repo to get the template
          temp_dir = os.path.join(tempfile.gettempdir(), "aptos-indexer-example")
          if os.path.exists(temp_dir):
              shutil.rmtree(temp_dir)
          
          clone_cmd = [
              "git", "clone", "https://github.com/aptos-labs/aptos-indexer-processor-example.git", 
              temp_dir, "--depth", "1"
          ]
          subprocess.run(clone_cmd, check=True, capture_output=True)
          
          # Copy relevant files from template
          shutil.copytree(os.path.join(temp_dir, "typescript"), project_dir, dirs_exist_ok=True)
          
          # Remove the git directory
          git_dir = os.path.join(project_dir, ".git")
          if os.path.exists(git_dir):
              shutil.rmtree(git_dir)
          
          # Customize the package.json
          package_json_path = os.path.join(project_dir, "package.json")
          with open(package_json_path, "r") as f:
              package_json = json.load(f)
          
          package_json["name"] = project_name
          package_json["description"] = f"Aptos indexer for {project_name}"
          
          with open(package_json_path, "w") as f:
              json.dump(package_json, f, indent=2)
          
          # Customize processor based on type
          processor_dir = os.path.join(project_dir, "src", "processors")
          os.makedirs(processor_dir, exist_ok=True)
          
          processor_code = ""
          if processor_type == "transaction":
              processor_code = generate_transaction_processor(project_name)
          elif processor_type == "event":
              processor_code = generate_event_processor(project_name)
          
          processor_file = os.path.join(processor_dir, f"{project_name}_processor.ts")
          with open(processor_file, "w") as f:
              f.write(processor_code)
          
          # Create README
          readme_content = f"""# {project_name} Aptos Indexer
  
  An Aptos indexer for processing {processor_type}s.
  
  ## Setup
  
  1. Install dependencies:
     ```
     npm install
     ```
  
  2. Configure connection:
     Edit the `.env` file to set your database and Aptos node URLs.
  
  3. Run the indexer:
     ```
     npm run start
     ```
  
  ## Architecture
  
  This indexer uses the Aptos Indexer Framework to process {processor_type}s from the Aptos blockchain.
  
  ## Development
  
  - `src/processors/{project_name}_processor.ts`: Contains the main processor logic
  - `src/models/`: Database models for storing indexed data
  """
  
          with open(os.path.join(project_dir, "README.md"), "w") as f:
              f.write(readme_content)
          
          # Create .env file
          env_content = """# Aptos Node URL
  APTOS_NODE_URL=https://fullnode.devnet.aptoslabs.com/v1
  
  # Database Configuration
  DB_HOST=localhost
  DB_PORT=5432
  DB_NAME=postgres
  DB_USER=postgres
  DB_PASS=postgres
  
  # Indexer Configuration
  STARTING_VERSION=0
  BATCH_SIZE=500
  """
          
          with open(os.path.join(project_dir, ".env"), "w") as f:
              f.write(env_content)
          
          return f"""
  Successfully created Aptos indexer project at {project_dir}!
  
  The project includes:
  - TypeScript boilerplate for an Aptos indexer
  - {processor_type.capitalize()} processor template
  - Database models and configurations
  - Environment setup
  
  Next steps:
  1. Navigate to the project directory: `cd {project_name}`
  2. Install dependencies: `npm install`
  3. Configure your database in .env
  4. Start developing your indexer!
  
  See the README.md file for more information.
  """
      
      except Exception as e:
          return f"Error creating indexer project: {str(e)}"

• aptos_mcp_server.py:468-595 (helper)

  Helper function that generates TypeScript code for a transaction processor class used in the indexer project.

  def generate_transaction_processor(project_name: str) -> str:
      """Generate code for a transaction processor"""
      return f"""import {{
    InputModels,
    Models,
    OutputModels,
    ProcessingResult,
    Transaction,
    TransactionModel,
    TransactionProcessor,
    UserTransactionInput,
    parseTransaction,
  }} from "@aptos-labs/indexer-sdk";
  
  /**
   * {project_name.capitalize()} Transaction Processor
   * 
   * This processor handles transactions and extracts relevant information.
   */
  export class {project_name.capitalize()}TransactionProcessor extends TransactionProcessor {{
    constructor() {{
      super();
    }}
  
    /**
     * Process a batch of transactions
     */
    async process(
      transactionInputs: UserTransactionInput[],
    ): Promise<ProcessingResult> {{
      const processingResult = new ProcessingResult();
      console.log(`Processing ${{transactionInputs.length}} transactions`);
  
      for (const transactionInput of transactionInputs) {{
        const transaction = parseTransaction(transactionInput);
        
        // Process each transaction
        try {{
          await this.processTransaction(transaction, processingResult);
        }} catch (e) {{
          console.error(
            `Error processing transaction ${{transaction.version}}: ${{e}}`,
          );
        }}
      }}
  
      return processingResult;
    }}
  
    /**
     * Process a single transaction
     */
    async processTransaction(
      transaction: Transaction,
      processingResult: ProcessingResult,
    ): Promise<void> {{
      // Check if transaction is successful
      if (!transaction.success) {{
        return;
      }}
  
      // Extract basic transaction data
      const txModel = new TransactionModel();
      txModel.version = transaction.version;
      txModel.hash = transaction.hash;
      txModel.sender = transaction.sender;
      txModel.success = transaction.success;
      txModel.timestamp = new Date(Number(transaction.timestamp) / 1000);
      txModel.blockHeight = transaction.blockHeight;
      
      // Add to processing result
      processingResult.transactionModel = txModel;
  
      // Process specific entry functions
      if (transaction.payload?.type === "entry_function_payload") {{
        const entryFunctionFullStr = transaction.payload.function;
        
        // Example: Process a specific entry function
        if (entryFunctionFullStr === "0x1::coin::transfer") {{
          // Handle coin transfer function
          this.processCoinTransfer(transaction, processingResult);
        }}
        
        // TODO: Add more function handlers
      }}
    }}
  
    /**
     * Process a coin transfer transaction
     */
    private processCoinTransfer(
      transaction: Transaction,
      processingResult: ProcessingResult,
    ): void {{
      if (
        transaction.payload?.type !== "entry_function_payload" ||
        !transaction.payload.arguments
      ) {{
        return;
      }}
  
      try {{
        // Extract function arguments
        const [recipient, amount] = transaction.payload.arguments;
        
        // Create custom transaction model
        const transferModel = new Models.{project_name.capitalize()}TransferModel();
        transferModel.version = transaction.version;
        transferModel.sender = transaction.sender;
        transferModel.recipient = recipient as string;
        transferModel.amount = BigInt(amount as string);
        transferModel.timestamp = new Date(Number(transaction.timestamp) / 1000);
        
        // Add to processing result
        processingResult.models.push(transferModel);
        
        console.log(
          `Processed transfer: ${{transaction.sender}} -> ${{recipient}} (${{amount}})`,
        );
      }} catch (e) {{
        console.error(`Error processing coin transfer: ${{e}}`);
      }}
    }}
  }}
  
  // Register processor
  new {project_name.capitalize()}TransactionProcessor().start();
  """

• aptos_mcp_server.py:597-739 (helper)

  Helper function that generates TypeScript code for an event processor class used in the indexer project.

  def generate_event_processor(project_name: str) -> str:
      """Generate code for an event processor"""
      return f"""import {{
    InputModels,
    Models,
    OutputModels,
    ProcessingResult,
    Event,
    EventProcessor,
    UserTransactionInput,
    parseEvent,
  }} from "@aptos-labs/indexer-sdk";
  
  /**
   * {project_name.capitalize()} Event Processor
   * 
   * This processor handles events and extracts relevant information.
   */
  export class {project_name.capitalize()}EventProcessor extends EventProcessor {{
    constructor() {{
      super();
    }}
  
    /**
     * Process a batch of events
     */
    async process(
      eventInputs: InputModels.Event[],
    ): Promise<ProcessingResult> {{
      const processingResult = new ProcessingResult();
      console.log(`Processing ${{eventInputs.length}} events`);
  
      for (const eventInput of eventInputs) {{
        const event = parseEvent(eventInput);
        
        // Process each event
        try {{
          await this.processEvent(event, processingResult);
        }} catch (e) {{
          console.error(
            `Error processing event ${{event.type}} (version: ${{event.version}}): ${{e}}`,
          );
        }}
      }}
  
      return processingResult;
    }}
  
    /**
     * Process a single event
     */
    async processEvent(
      event: Event,
      processingResult: ProcessingResult,
    ): Promise<void> {{
      // Create base event model
      const eventModel = new Models.{project_name.capitalize()}EventModel();
      eventModel.transactionVersion = event.version;
      eventModel.eventType = event.type;
      eventModel.data = event.data ? JSON.stringify(event.data) : null;
      eventModel.timestamp = new Date(Number(event.timestamp) / 1000);
      
      // Process specific event types
      if (event.type.includes("0x1::coin::DepositEvent")) {{
        await this.processDepositEvent(event, processingResult);
      }} else if (event.type.includes("0x1::coin::WithdrawEvent")) {{
        await this.processWithdrawEvent(event, processingResult);
      }}
      
      // Add base event to processing result 
      processingResult.models.push(eventModel);
    }}
  
    /**
     * Process a deposit event
     */
    private async processDepositEvent(
      event: Event,
      processingResult: ProcessingResult,
    ): Promise<void> {{
      if (!event.data) {{
        return;
      }}
  
      try {{
        // Extract event data
        const {{ amount }} = event.data;
        
        // Create deposit model
        const depositModel = new Models.{project_name.capitalize()}DepositModel();
        depositModel.transactionVersion = event.version;
        depositModel.address = event.accountAddress;
        depositModel.amount = BigInt(amount);
        depositModel.timestamp = new Date(Number(event.timestamp) / 1000);
        
        // Add to processing result
        processingResult.models.push(depositModel);
        
        console.log(
          `Processed deposit: ${{event.accountAddress}} (+${{amount}})`,
        );
      }} catch (e) {{
        console.error(`Error processing deposit event: ${{e}}`);
      }}
    }}
  
    /**
     * Process a withdraw event
     */
    private async processWithdrawEvent(
      event: Event,
      processingResult: ProcessingResult,
    ): Promise<void> {{
      if (!event.data) {{
        return;
      }}
  
      try {{
        // Extract event data
        const {{ amount }} = event.data;
        
        // Create withdraw model
        const withdrawModel = new Models.{project_name.capitalize()}WithdrawModel();
        withdrawModel.transactionVersion = event.version;
        withdrawModel.address = event.accountAddress;
        withdrawModel.amount = BigInt(amount);
        withdrawModel.timestamp = new Date(Number(event.timestamp) / 1000);
        
        // Add to processing result
        processingResult.models.push(withdrawModel);
        
        console.log(
          `Processed withdraw: ${{event.accountAddress}} (-${{amount}})`,
        );
      }} catch (e) {{
        console.error(`Error processing withdraw event: ${{e}}`);
      }}
    }}
  }}
  
  // Register processor
  new {project_name.capitalize()}EventProcessor().start();
  """

• aptos_mcp_server.py:989-989 (registration)

  The @mcp.tool() decorator registers the create_aptos_indexer function as an MCP tool.

  @mcp.tool()