Skip to main content

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')