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