ChainAware MCP Server

Blockchain fraud detection, behaviour analytics, token audits, and agent trust scores — exposed as MCP tools over Server-Sent Events.

Live on port 5000

Available Tools — 14

Fraud predictive_fraud AI fraud probability score and AML check for a single wallet. Requires apiKey.

Fraud · Batch predictive_fraud_batch Schedule fraud analysis for up to 1 000 wallets. Returns job_id + signature immediately.

Behaviour predictive_behaviour Next-action predictions, risk profile, DeFi category segmentation, and personalised recommendations for a wallet.

Behaviour · Batch predictive_behaviour_batch Schedule behaviour analysis for up to 1 000 wallets. Returns job_id + signature immediately.

Rug Pull predictive_rug_pull Forecasts whether a liquidity pool or contract is likely to rug pull. Returns full risk indicators and liquidity events.

Credit credit_score AI-driven crypto trust score (1–9) combining on-chain inflows, outflows, fraud signals, and social graph analysis.

Token Rank token_rank_list Paginated, filterable list of tokens ranked by holder community strength. Supports sort, category filter, and name search.

Token Rank token_rank_single Community rank and top holders for a single token contract address and network.

Token Audit run_token_audit Get-or-create: returns a cached full audit immediately, or queues a new one returning a job_id.

Token Audit get_token_audit_result Poll for a completed token audit. Returns full risk report (ownership, liquidity, honeypot, reentrancy) when audit_status = "complete".

Agent Trust agents_trust_score_list Paginated list of ERC-8004 AI agents with their 0–1000 on-chain trust scores and tiers.

Agent Trust agents_trust_score_single Detailed trust report for a single agent by agent_id and chain_id.

Batch Jobs check_job_status Check progress of a batch job (completed / failed / pending counts). Requires job_id + signature.

Batch Jobs get_job_results Retrieve wallet address list from a completed or partial batch job. Call only when status is "completed" or "partial".

Endpoints

Method	Path	Description
GET	/	This page
GET	/sse	MCP SSE stream — external clients connect here
POST	/messages/	MCP message endpoint, keyed by `session_id`
POST	/chat	Single-turn GPT-4o agent — picks and calls a tool, returns final answer
POST	/generate-stream	Streaming Qwen3-32b agent response
GET	/.well-known/x402	x402 payment discovery — see chainaware.ai/x402-payment
GET	/.well-known/x402.json	x402 payment discovery (JSON)

Example Clients

OpenAI Function Calling + MCP (Interactive Chat)

import os, asyncio, json
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()
client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

async def init_and_process_query(sse_url, query):
    async with sse_client(sse_url) as streams:
        read_stream, write_stream = streams
        async with ClientSession(read_stream, write_stream) as sess:
            await sess.initialize()
            tools_resp = await sess.list_tools()
            functions = [
                {"name": t.name, "description": t.description, "parameters": t.inputSchema}
                for t in tools_resp.tools
            ]
            chat_resp = await client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": query}],
                functions=functions,
                function_call="auto",
            )
            msg = chat_resp.choices[0].message
            if msg.function_call:
                fn_name = msg.function_call.name
                fn_args = json.loads(msg.function_call.arguments)
                fn_args["apiKey"] = os.getenv("CA_MCP_API_KEY")  # injected server-side
                tool_resp = await sess.call_tool(fn_name, fn_args)
                output = tool_resp.content[0].text
                final = await client.chat.completions.create(
                    model="gpt-4o",
                    messages=[
                        {"role": "user", "content": query},
                        {"role": "assistant", "function_call": msg.function_call},
                        {"role": "function", "name": fn_name, "content": output},
                    ],
                )
                return final.choices[0].message.content
            return msg.content

asyncio.run(init_and_process_query("http://localhost:5000/sse", "Your query here"))

MCP Direct Client (No LLM)

import asyncio
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client

async def run_client(sse_url):
    async with sse_client(sse_url) as streams:
        read_stream, write_stream = streams
        async with ClientSession(read_stream, write_stream) as sess:
            await sess.initialize()
            tools = await sess.list_tools()
            print("Tools:", [t.name for t in tools.tools])

            fraud = await sess.call_tool("predictive_fraud", {
                "apiKey": "YOUR_API_KEY",
                "network": "ETH",
                "walletAddress": "vitalik.eth",
            })
            print("Fraud result:", fraud.content[0].text)

            audit = await sess.call_tool("run_token_audit", {
                "network": "eth",
                "contract_address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
            })
            print("Audit result:", audit.content[0].text)

asyncio.run(run_client("http://localhost:5000/sse"))

Enterprise account required. Tool calls that accept an apiKey require an active ChainAware Enterprise subscription. Visit chainaware.ai/pricing to get access.

x402 payments are supported at the backend API level. See chainaware.ai/x402-payment for integration details.