Spaces:
Sleeping
Sleeping
File size: 5,657 Bytes
c17d149 60a0114 2d03947 86d64b7 c17d149 8feb09e c17d149 69d8bd9 8f01dc4 c17d149 86d64b7 c062c0e c17d149 b1b9f8f c17d149 b1b9f8f c17d149 b1b9f8f c17d149 212618b c17d149 7e272ca b1b9f8f c17d149 212618b 60a0114 c17d149 bbf9aed c17d149 b1b9f8f c17d149 b1b9f8f c17d149 b1b9f8f c17d149 bbf9aed c17d149 b1b9f8f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 |
from groq import Groq
import json
from solders.keypair import Keypair
from solana.rpc.async_api import AsyncClient
from .mongodbconnection import provideClient
client=provideClient()
db=client["agentpaymentdb"]
coll=db["agentpaymentcol"]
rpc=AsyncClient("https://mainnet.helius-rpc.com/?api-key=4e833ada-d32c-48c5-b020-c11b2253f25b")
async def create_keypair(agent):
"""Create a new keypair and return it as MongoDB-friendly format"""
try:
keypair = Keypair()
keypair_data = {
"agent_id": agent,
"public_key": str(keypair.pubkey()),
"secret_key": bytes(keypair).hex()
}
await coll.update_one(
{"owner": "system"},
{"$push": {"agents": keypair_data}}
)
except Exception as e:
print(e)
raise
def load_keypair(agent):
document=coll.find_one({"owner":"system"})
keypair=None
agents=document.get("agents")
print(agents)
for doc in agents:
if doc["agent_id"]==agent:
keypair=doc["secret_key"]
secret_key_bytes =bytes.fromhex(keypair)
keypair = Keypair.from_bytes(secret_key_bytes)
print(f"✅ Keypair loaded")
print(f"📍 Public Key: {keypair.pubkey()}")
publickey=str(keypair.pubkey())
return publickey
async def transfer_sol(agent_id, to_pubkey, amount_sol):
"""Transfer SOL from one account to another"""
try:
from_publickey=load_keypair(agent_id)
lamports = int(amount_sol * 1e9)
transfer_ix = transfer(
TransferParams(
from_pubkey=from_publickey,
to_pubkey=to_pubkey,
lamports=lamports
))
recent_blockhash =rpc.get_latest_blockhash()
txn = Transaction(recent_blockhash=recent_blockhash.value.blockhash)
txn.add(transfer_ix)
txn.sign(from_keypair)
result = await rpc.client.send_transaction(txn, from_keypair)
print(f"✅ SOL Transfer successful!")
print(f"📝 Signature: {result.value}")
return True
except Exception as e:
print(f"❌ Transfer failed: {e}")
return False
class VelocityPayableAgent():
def __init__(self):
self.tools = [transfer_sol]
def execute_tool_call(self,tool_call):
"""Parse and execute a single tool call"""
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
return function_to_call(**function_args)
async def Agentinit(self,agent):
await create_keypair(agent)
def Agent(self,user_input):
client = Groq(api_key="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")
available_functions= {
"transfer_sol":transfer_sol,
}
transfer_sol_tool = {
"type": "function",
"function": {
"name": "transfer_sol",
"description": "Transfer SOL from one account to other on Solana.",
"parameters": {
"type": "object",
"properties": {
"agent_id": {
"type": "string",
"description": "id of the agent"
},
"to_pubkey": {
"type": "string",
"description": "Destination public key."
},
"amount_sol": {
"type": "number",
"description": "Amount of SOL to send."
}
},
"required": ["from_keypair", "to_pubkey", "amount_sol"]
}
}
}
messages=[
{"role":"system","content":"""
You are a tool-only agent.
Rules:
1. You MUST call a tool for every query from the user.
2. You are NOT allowed to answer directly.
3. Always return tool call in the correct function-calling JSON format.
4. When unsure, still pick a tool instead of answering directly.
5. Do not apologize, do not explain, do not output reasoning—only valid tool calls.
"""},
{"role": "user", "content": user_input}]
response = client.chat.completions.create(
model="llama-3.1-8b-instant",
messages=messages,
tools=self.tools,
)
messages.append(response.choices[0].message)
if response.choices[0].message.tool_calls:
for tool_call in response.choices[0].message.tool_calls:
function_response = execute_tool_call(tool_call)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.function.name,
"content": str(function_response)
})
final = client.chat.completions.create(
model="llama-3.1-8b-instant",
messages=messages
)
print(final.choices[0].message.content)
return final.choices[0].message.content
|