aptos-mcp

Instructions

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()