Shinkai Python Library
How to install​
pip install shinkai_message_pyo-XXXX
How to Use​
Import​
import shinkai_message_pyo3
Generate encryption keys​
import nacl.public
import nacl.signing
import nacl.utils
def to_hex_string(byte_array):
return byte_array.hex()
def generate_encryption_keys(seed):
# Curve25519 key pair generation using X25519
priv_key = nacl.public.PrivateKey(seed)
pub_key = priv_key.public_key
return {
"my_encryption_sk_string": to_hex_string(priv_key._private_key),
"my_encryption_pk_string": to_hex_string(pub_key._public_key)
}
def generate_signature_keys():
# Ed25519 key pair generation
signing_key = nacl.signing.SigningKey.generate()
verify_key = signing_key.verify_key
return {
"my_identity_sk_string": to_hex_string(signing_key._seed),
"my_identity_pk_string": to_hex_string(verify_key._key)
}
# Example usage:
seed = nacl.utils.random(nacl.public.PrivateKey.SIZE)
encryption_keys = generate_encryption_keys(seed)
signature_keys = generate_signature_keys()
print(encryption_keys)
print(signature_keys)
Initiate registration​
initial_registration_msg = shinkai_message_pyo3.PyShinkaiMessageBuilder.initial_registration_with_no_code_for_device(
my_encryption_sk_string, # device
my_identity_sk_string, # device
my_encryption_sk_string, # profile
my_identity_sk_string, # profile
"main_device",
"@@node1.shinkai",
"",
"@@node1.shinkai",
)
msg_dict = json.loads(initial_registration_msg)
print("result initial code msg", msg_dict)
url = "http://localhost:9550/v1/use_registration_code"
# Make the POST request
response = requests.post(url, json=msg_dict)
# Print response
print(response.status_code)
print(response.text)
parsed_response = json.loads(response.text)
node_encryption_pk = parsed_response["encryption_public_key"]
node_signature_pk = parsed_response["identity_public_key"]
Job Creation​
job_scope = shinkai_message_pyo3.PyJobScope()
create_job_msg = shinkai_message_pyo3.PyShinkaiMessageBuilder.job_creation(
my_encryption_sk_string,
my_identity_sk_string,
node_encryption_pk,
job_scope,
"@@node1.shinkai",
"main",
"@@node1.shinkai",
"main/agent/my_gpt"
)
print("text", create_job_msg)
job_creation_dict = json.loads(create_job_msg)
print("result job creation msg", msg_dict)
url = "http://kao0112.duckdns.org:9550/v1/create_job"
url = "http://localhost:9550/v1/create_job"
# Make the POST request
response = requests.post(url, json=job_creation_dict)
# Print response
print(response.status_code)
print(response.text)
job_id = json.loads(response.text)["data"]
Message Job​
content = "What is a car?"
job_message_msg = shinkai_message_pyo3.PyShinkaiMessageBuilder.job_message(
job_id,
content,
"",
my_encryption_sk_string,
my_identity_sk_string,
node_encryption_pk,
"@@node1.shinkai",
"main",
"@@node1.shinkai",
"main/agent/my_gpt",
)
job_message_dict = json.loads(job_message_msg)
url = "http://kao0112.duckdns.org:9550/v1/job_message"
url = "http://localhost:9550/v1/job_message"
# Make the POST request
response = requests.post(url, json=job_message_dict)
# Print response
print(response.status_code)
print(response.text)
Get last messages from inbox​
last_messages = shinkai_message_pyo3.PyShinkaiMessageBuilder.get_last_messages_from_inbox(
my_encryption_sk_string,
my_identity_sk_string,
node_encryption_pk,
"job_inbox::" + job_id + "::false",
10,
"@@node1.shinkai",
"main",
"@@node1.shinkai",
"main/agent/my_gpt",
None,
)
last_messages_dict = json.loads(last_messages)
url = "http://kao0112.duckdns.org:9550/v1/last_messages_from_inbox"
url = "http://localhost:9550/v1/last_messages_from_inbox"
# Make the POST request
response = requests.post(url, json=last_messages_dict)
# Print response
print(response.status_code)
print(response.text)
File Uploading​
import os
import blake3
import json
import requests
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
class FileUploader:
def __init__(self, base_url, my_encryption_secret_key, my_signature_secret_key, receiver_public_key, job_inbox, sender, sender_subidentity, receiver):
self.base_url = base_url
self.my_encryption_secret_key = my_encryption_secret_key
self.my_signature_secret_key = my_signature_secret_key
self.receiver_public_key = receiver_public_key
self.job_inbox = job_inbox
self.job_id = FileUploader.extract_job_id(job_inbox)
self.sender = sender
self.sender_subidentity = sender_subidentity
self.receiver = receiver
self.symmetric_key = None
self.folder_id = None
def extract_job_id(job_inbox):
parts = job_inbox.split('::')
if len(parts) >= 2:
return parts[1]
else:
raise ValueError('Invalid job_inbox format')
def calculate_hash_from_symmetric_key(self):
if not self.symmetric_key:
raise ValueError('Symmetric key is not set')
raw_key_string = self.symmetric_key
hash = blake3.blake3(raw_key_string.encode()).hexdigest()
return hash
def create_folder(self):
key_data = os.urandom(32)
self.symmetric_key = key_data.hex()
message = shinkai_message_pyo3.PyShinkaiMessageBuilder.create_files_inbox_with_sym_key(
self.my_encryption_secret_key,
self.my_signature_secret_key,
self.receiver_public_key,
self.job_inbox,
self.symmetric_key,
self.sender_subidentity,
self.sender,
self.receiver
)
message_dict = json.loads(message)
response = requests.post(f'{self.base_url}/v1/create_files_inbox_with_symmetric_key', json=message_dict)
if response.status_code == 200:
print("Successfully created folder: ", self.calculate_hash_from_symmetric_key())
print("Response: ", response.text)
if response.status_code != 200:
print("Error:", response.status_code)
print("Response content:", response.text)
response.raise_for_status()
self.folder_id = self.calculate_hash_from_symmetric_key()
def upload_encrypted_file(self, file_path, filename=None):
with open(file_path, 'rb') as f:
file_data = f.read()
print("First 100 characters of the file: ", file_data[:100])
symmetric_key_bytes = bytes.fromhex(self.symmetric_key)
aesgcm = AESGCM(symmetric_key_bytes)
nonce = os.urandom(12)
encrypted_file_data = aesgcm.encrypt(nonce, file_data, None)
# Use os.path.basename to get the filename without the directory
filename = filename or os.path.basename(file_path)
form_data = {
'file': (filename, encrypted_file_data)
}
hash = self.folder_id
response = requests.post(f'{self.base_url}/v1/add_file_to_inbox_with_symmetric_key/{hash}/{nonce.hex()}', files=form_data)
if response.status_code == 200:
print("Successfully uploaded file: ", file_path)
print("Response: ", response.text)
if response.status_code != 200:
print("Error:", response.status_code)
print("Response content:", response.text)
response.raise_for_status()
def finalize_and_send(self, content):
message = shinkai_message_pyo3.PyShinkaiMessageBuilder.job_message(
self.job_id,
content,
self.folder_id,
self.my_encryption_secret_key,
self.my_signature_secret_key,
self.receiver_public_key,
self.sender,
self.sender_subidentity,
self.receiver,
self.sender_subidentity
)
message_dict = json.loads(message)
response = requests.post(f'{self.base_url}/v1/job_message', json=message_dict)
response.raise_for_status()
base_url = "http://kao0112.duckdns.org:9550"
base_url = "http://localhost:9550"
job_inbox = "job_inbox::" + job_id + "::false"
print("job inbox" , job_inbox)
sender = "@@node1.shinkai"
sender_subidentity = "main"
receiver = sender
uploader = FileUploader(base_url, my_encryption_sk_string, my_identity_sk_string, node_encryption_pk, job_inbox, sender, sender_subidentity, receiver)
uploader.create_folder()
uploader.upload_encrypted_file('content/Zeko_Mina_Rollup.pdf')
uploader.finalize_and_send('tell me about Zeko')