1
+ from typing import Dict , Any , List , Optional
2
+ from solana_agent import AutoTool , ToolRegistry
3
+ import httpx
4
+
5
+ class SnsLookupTool (AutoTool ):
6
+ def __init__ (self , registry : Optional [ToolRegistry ] = None ):
7
+ super ().__init__ (
8
+ name = "sns_lookup" ,
9
+ description = "Lookup SNS domain with a QuickNode RPC." ,
10
+ registry = registry ,
11
+ )
12
+ self .quicknode_url = None
13
+
14
+ def get_schema (self ) -> Dict [str , Any ]:
15
+ return {
16
+ "type" : "object" ,
17
+ "properties" : {
18
+ "domain" : {
19
+ "type" : "string" ,
20
+ "description" : "SNS domain to resolve, e.g. domain.sol" ,
21
+ }
22
+ },
23
+ "required" : ["domain" ],
24
+ "additionalProperties" : False ,
25
+ }
26
+
27
+ def configure (self , config : Dict [str , Any ]) -> None :
28
+ super ().configure (config )
29
+ tool_cfg = config .get ("tools" , {}).get ("sns_lookup" , {})
30
+ self .quicknode_url = tool_cfg .get ("quicknode_url" )
31
+
32
+ async def execute (self , domain : str ) -> Dict [str , Any ]:
33
+ if not self .quicknode_url :
34
+ return {"status" : "error" , "message" : "No quicknode_url configured for SNS lookup." }
35
+ payload = {
36
+ "method" : "sns_resolveDomain" ,
37
+ "params" : [domain ],
38
+ "id" : 1 ,
39
+ "jsonrpc" : "2.0" ,
40
+ }
41
+ headers = {"Content-Type" : "application/json" }
42
+ async with httpx .AsyncClient () as client :
43
+ resp = await client .post (self .quicknode_url , json = payload , headers = headers )
44
+ resp .raise_for_status ()
45
+ data = resp .json ()
46
+ return {"status" : "success" , "result" : data }
47
+
48
+ class SnsLookupPlugin :
49
+ def __init__ (self ):
50
+ self .name = "sns_lookup"
51
+ self .config = None
52
+ self .tool_registry = None
53
+ self ._tool = None
54
+
55
+ @property
56
+ def description (self ):
57
+ return "Plugin for resolving SNS domains using a QuickNode endpoint."
58
+
59
+ def initialize (self , tool_registry : ToolRegistry ) -> None :
60
+ self .tool_registry = tool_registry
61
+ self ._tool = SnsLookupTool (registry = tool_registry )
62
+
63
+ def configure (self , config : Dict [str , Any ]) -> None :
64
+ self .config = config
65
+ if self ._tool :
66
+ self ._tool .configure (self .config )
67
+
68
+ def get_tools (self ) -> List [AutoTool ]:
69
+ return [self ._tool ] if self ._tool else []
70
+
71
+ def get_plugin ():
72
+ return SnsLookupPlugin ()
0 commit comments