This server exposes a set of tools via Server-Sent Events (SSE) for interacting with wallet behaviour prediction,fraud detection and rug pull prediction.
predictive_behaviour – AI-powered fraud detection, intention calculation & analytics for Ethereum, Polygon, BSC, TON and HAQQ.predictive_behaviour_batch – Batch AI-powered fraud detection, intention calculation & analytics for Ethereum, Polygon, BSC, TON and HAQQ.predictive_fraud – AI-powered fraud detection for Ethereum, Polygon, BSC, TON and HAQQ .predictive_fraud_batch – Batch AI-powered fraud detection for Ethereum, Polygon, BSC, TON and HAQQ .credit_score –AI-driven blockchain analytics evaluate the crypto trust score for each account by reviewing inflows and outflows from Ethereum accounts alongside other blockchain data.predictive_rug_pull – AI-powered rug pull detection for Ethereum, BSC and HAQQ .token_rank_list – TokenRank analyzes the community of token holders and ranks every token by the strength of its holders.token_rank_single – Similar to TokenRank List,Token Rank analyzes the community of token holders and ranks every token by the strength of its holders.Except the token rank and token details the token rank single tool fetches the best holders their details and its globalRank alongside others in same network.run_token_audit – Requests a Token Audit for a given token contract or returns already calculated audit data for requested token. This tool is "get-or-create": it first checks if a completed audit already exists for this contract, and if so returns the FULL risk report immediately. If no audit exists yet, it queues a new one and returns a job_id + "queued" status instead.get_token_audit_result – Fetches the current status or final results of a previously triggered Token Audit job for a given contract address and chain.agents_trust_score_list – The ChainAware Agent Trust Score is a 0-1000 score that measures how safe it is to interact with any ERC-8004 registered AI agent.agents_trust_score_single – Similar to Agent Trust List, Agent Trust Score Single is a 0-1000 score that measures how safe it is to interact with any ERC-8004 registered AI agent.check_job_status – Check the progress of a scheduled batch calculation job.Returns counts only (completed, failed, pending) — no wallet data. Call this when the user asks whether a job is done or how it is progressing. If status is 'processing' or 'pending', inform the user and do not call get_job_results. Only suggest fetching results when status is 'completed' or 'partial'.Both job_id and signature from schedule_calculation are required to call this tool.Never call this without both values present in context.get_job_results – Retrieve the results of a completed or partially completed batch job.Only call this when check_job_status shows status is 'completed' or 'partial'.Returns a list of completed wallet addresses and the shared chain/network — use these to query the main backend for actual wallet analysis data.This does NOT return wallet data directly, only the address list needed to fetch it.Both job_id and signature from schedule_calculation are required to call this tool.Never call this without both values present in context.import os, asyncio, json
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
import openai
from dotenv import load_dotenv
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
async def chat_loop(sse_url):
while True:
q = input("You: ").strip()
if not q or q.lower() == "quit":
break
resp = await init_and_process_query(sse_url, q)
print("Assistant:", resp)
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 = openai.chat.completions.create(
model="gpt-4-0613",
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")
tool_resp = await sess.call_tool(fn_name, fn_args)
output = tool_resp.content[0].text
final_resp = openai.chat.completions.create(
model="gpt-4-0613",
messages=[
{"role":"user","content":query},
{"role":"assistant","function_call":msg.function_call},
{"role":"function","name":fn_name,"content":output}
]
)
return final_resp.choices[0].message.content
return msg.content
# Run it:
asyncio.run(chat_loop("{mcpServerUrl}/sse"))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("check_fraud", {"network": "ETH", "walletAddress": "vitalik.eth"})
print("Fraud result:", fraud.content[0].text)
audit = await sess.call_tool("check_audit", {"network": "ETH", "walletAddress": "vitalik.eth"})
print("Audit result:", audit.content[0].text)
rug = await sess.call_tool("rug_pull_check", {"network": "BNB", "walletAddress": "0x89c5..."})
print("Rug Pull result:", rug.content[0].text)
# Run it:
asyncio.run(run_client("{mcpServerUrl}/sse"))SSE Endpoint: /sse
Post messages via: /messages/
**Note that to be able to use the tools you need to have an active Enterprise Account from ChainAware.ai
For detailed usage, tool definitions, and input/output schemas, check out the GitHub Documentation.