export interface AbiItem {
type: string;
name?: string;
inputs?: { name: string; type: string; indexed?: boolean }[];
outputs?: { name: string; type: string }[];
stateMutability?: string;
anonymous?: boolean;
}
export interface GeneratedTool {
name: string;
type: "read" | "write" | "event";
description: string;
inputs: string[];
}
export function parseAbi(abiJson: string): AbiItem[] {
try {
return JSON.parse(abiJson);
} catch {
throw new Error("Invalid ABI JSON");
}
}
export function extractTools(abi: AbiItem[]): GeneratedTool[] {
const tools: GeneratedTool[] = [];
for (const item of abi) {
if (item.type === "function" && item.name) {
const isRead =
item.stateMutability === "view" ||
item.stateMutability === "pure";
tools.push({
name: item.name,
type: isRead ? "read" : "write",
description: isRead
? `Read ${item.name} from contract`
: `Execute ${item.name} transaction`,
inputs: item.inputs?.map(i => `${i.name}: ${i.type}`) || [],
});
} else if (item.type === "event" && item.name) {
tools.push({
name: `get_${item.name.toLowerCase()}_events`,
type: "event",
description: `Query ${item.name} events`,
inputs: item.inputs?.filter(i => i.indexed).map(i => `${i.name}: ${i.type}`) || [],
});
}
}
return tools;
}
function solidityTypeToJsonSchema(solType: string): string {
if (solType.startsWith("uint") || solType.startsWith("int")) {
return "string"; // Use string for big numbers
}
if (solType === "address") return "string";
if (solType === "bool") return "boolean";
if (solType === "string") return "string";
if (solType.startsWith("bytes")) return "string";
if (solType.endsWith("[]")) return "array";
return "string";
}
export function generateServerPy(
serverName: string,
contractAddress: string,
rpcUrl: string,
abi: AbiItem[]
): string {
const functions = abi.filter(item => item.type === "function");
const events = abi.filter(item => item.type === "event");
const toolFunctions = functions.map(fn => {
const isRead = fn.stateMutability === "view" || fn.stateMutability === "pure";
const args = fn.inputs?.map(i => i.name).join(", ") || "";
const argsWithTypes = fn.inputs?.map(i => `${i.name}: str`).join(", ") || "";
if (isRead) {
return `
@mcp.tool()
async def ${fn.name}(${argsWithTypes}) -> str:
"""Read ${fn.name} from the contract."""
try:
result = contract.functions.${fn.name}(${args}).call()
return json.dumps({"success": True, "result": str(result)})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
`;
} else {
return `
@mcp.tool()
async def ${fn.name}(${argsWithTypes}, simulate: bool = True) -> str:
"""
Execute ${fn.name} on the contract.
Set simulate=False to actually send the transaction.
"""
try:
func = contract.functions.${fn.name}(${args})
if simulate:
# Simulate the transaction
gas_estimate = func.estimate_gas({"from": "0x40252CFDF8B20Ed757D61ff157719F33Ec332402"})
return json.dumps({
"success": True,
"simulated": True,
"gas_estimate": gas_estimate,
"message": "Transaction simulated successfully. Set simulate=False to execute."
})
else:
return json.dumps({
"success": False,
"error": "Live transactions require a signer. Configure PRIVATE_KEY env var."
})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
`;
}
}).join("\n");
const eventFunctions = events.map(ev => {
return `
@mcp.tool()
async def get_${ev.name?.toLowerCase()}_events(from_block: str = "latest", to_block: str = "latest") -> str:
"""Query ${ev.name} events from the contract."""
try:
event_filter = contract.events.${ev.name}.create_filter(
from_block=int(from_block) if from_block != "latest" else "latest",
to_block=int(to_block) if to_block != "latest" else "latest"
)
events = event_filter.get_all_entries()
return json.dumps({
"success": True,
"count": len(events),
"events": [dict(e) for e in events[:100]] # Limit to 100
}, default=str)
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
`;
}).join("\n");
return `#!/usr/bin/env python3
"""
${serverName} MCP Server
Generated by Sperax MCP
https://github.com/Sperax/sperax-mcp-server
"""
import os
import json
from mcp.server.fastmcp import FastMCP
from web3 import Web3
# Configuration
RPC_URL = os.environ.get("RPC_URL", "${rpcUrl}")
CONTRACT_ADDRESS = "${contractAddress}"
# Initialize Web3
w3 = Web3(Web3.HTTPProvider(RPC_URL))
# Contract ABI
ABI = json.loads('''${JSON.stringify(abi)}''')
# Initialize contract
contract = w3.eth.contract(address=Web3.to_checksum_address(CONTRACT_ADDRESS), abi=ABI)
# Initialize MCP server
mcp = FastMCP("${serverName}")
@mcp.tool()
async def get_contract_info() -> str:
"""Get contract metadata and connection status."""
try:
connected = w3.is_connected()
block = w3.eth.block_number if connected else None
return json.dumps({
"success": True,
"contract_address": CONTRACT_ADDRESS,
"rpc_url": RPC_URL,
"connected": connected,
"current_block": block,
"chain_id": w3.eth.chain_id if connected else None
})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
@mcp.tool()
async def get_balance(address: str) -> str:
"""Get native token balance for an address."""
try:
balance = w3.eth.get_balance(Web3.to_checksum_address(address))
return json.dumps({
"success": True,
"address": address,
"balance_wei": str(balance),
"balance_eth": str(w3.from_wei(balance, "ether"))
})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
${toolFunctions}
${eventFunctions}
if __name__ == "__main__":
mcp.run()
`;
}
export function generateRequirementsTxt(): string {
return `# Sperax MCP Server Dependencies
mcp>=1.0.0
web3>=6.0.0
python-dotenv>=1.0.0
`;
}
export function generateReadme(serverName: string, rpcUrl: string): string {
return `# ${serverName} MCP Server
Generated by [Sperax MCP](https://github.com/Sperax/sperax-mcp-server)
## Quick Start
1. Install dependencies:
\`\`\`bash
pip install -r requirements.txt
\`\`\`
2. Add to Claude Desktop config (\`~/Library/Application Support/Claude/claude_desktop_config.json\`):
\`\`\`json
{
"mcpServers": {
"${serverName.toLowerCase().replace(/\s+/g, "-")}": {
"command": "python",
"args": ["${process.cwd()}/server.py"],
"env": {
"RPC_URL": "${rpcUrl}"
}
}
}
}
\`\`\`
3. Restart Claude Desktop
## Tools
This server includes tools for all contract functions and events.
All write operations simulate by default for safety.
## License
MIT
`;
}
