Skip to content
light theme Algorand Developer Portal dark theme Algorand Developer Portal Algorand Developer Portal

Multisig Transaction Signing with KMD

← Back to KMD Client

Description

Section titled “Description”

This example demonstrates how to sign multisig transactions using the KMD sign_multisig_transaction() method. It shows:

  • Creating a multisig account with 2-of-3 threshold
  • Funding the multisig account via the dispenser
  • Creating a payment transaction from the multisig account
  • Signing with the first participant (partial signature)
  • Signing with the second participant (completing the multisig)
  • Submitting the fully signed transaction to the network Prerequisites: Covered operations:
  • sign_multisig_transaction() - Sign a multisig transaction with a participant key

Prerequisites

Section titled “Prerequisites”
  • LocalNet running (via algokit localnet start)

Run This Example

Section titled “Run This Example”

From the repository’s examples directory:

Terminal window
cd examples
uv run python kmd_client/11_multisig_signing.py

Code

Section titled “Code”

View source on GitHub

11_multisig_signing.py
# ruff: noqa: N999, C901, PLR0912, PLR0915
"""
Example: Multisig Transaction Signing with KMD


This example demonstrates how to sign multisig transactions using the KMD
sign_multisig_transaction() method. It shows:
  - Creating a multisig account with 2-of-3 threshold
  - Funding the multisig account via the dispenser
  - Creating a payment transaction from the multisig account
  - Signing with the first participant (partial signature)
  - Signing with the second participant (completing the multisig)
  - Submitting the fully signed transaction to the network


Prerequisites:
- LocalNet running (via `algokit localnet start`)


Covered operations:
- sign_multisig_transaction() - Sign a multisig transaction with a participant key
"""


import sys


import msgpack
from shared import (
    cleanup_test_wallet,
    create_algod_client,
    create_algorand_client,
    create_kmd_client,
    create_test_wallet,
    print_error,
    print_header,
    print_info,
    print_step,
    print_success,
    shorten_address,
    wait_for_confirmation,
)


from algokit_common import address_from_public_key, public_key_from_address
from algokit_kmd_client.models import (
    GenerateKeyRequest,
    ImportMultisigRequest,
    MultisigSig,
    MultisigSubsig,
    SignMultisigTxnRequest,
)
from algokit_transact import (
    PaymentTransactionFields,
    Transaction,
    TransactionType,
    assign_fee,
    encode_signed_transaction,
    encode_transaction_raw,
)
from algokit_transact.models.signed_transaction import SignedTransaction
from algokit_transact.signing.types import MultisigSignature, MultisigSubsignature
from algokit_utils import AlgoAmount
from algokit_utils.transactions.types import PaymentParams




def format_micro_algo(micro_algo: int) -> str:
    """Format microAlgos to a human-readable string."""
    algo_value = micro_algo / 1_000_000
    return f"{micro_algo:,} microALGO ({algo_value:.6f} ALGO)"




def decode_kmd_multisig_response(multisig_bytes: bytes) -> dict:
    """
    Decode the KMD multisig response bytes into a MultisigSig structure.


    The KMD API returns msgpack-encoded MultisigSig with wire keys:
    - 'subsig' -> subsignatures array
    - 'thr' -> threshold
    - 'v' -> version
    Each subsig has:
    - 'pk' -> publicKey
    - 's' -> signature (optional)
    """
    decoded = msgpack.unpackb(multisig_bytes, strict_map_key=False)


    subsig_array = decoded.get("subsig", [])
    threshold = decoded.get("thr", 0)
    version = decoded.get("v", 0)


    subsignatures = []
    for subsig in subsig_array:
        public_key = subsig.get("pk")
        signature = subsig.get("s")
        subsignatures.append({"public_key": public_key, "signature": signature})


    return {"subsignatures": subsignatures, "threshold": threshold, "version": version}




def kmd_multisig_to_transact_multisig(kmd_msig: dict) -> MultisigSignature:
    """Convert a KMD MultisigSig to the transact MultisigSignature format."""
    return MultisigSignature(
        version=kmd_msig["version"],
        threshold=kmd_msig["threshold"],
        subsigs=[
            MultisigSubsignature(public_key=subsig["public_key"], sig=subsig["signature"])
            for subsig in kmd_msig["subsignatures"]
        ],
    )




def count_signatures(msig: dict) -> int:
    """Count the number of signatures in a KMD MultisigSig."""
    return sum(1 for subsig in msig["subsignatures"] if subsig["signature"] is not None)




def main() -> None:
    print_header("KMD Multisig Transaction Signing Example")


    kmd = create_kmd_client()
    algod = create_algod_client()
    algorand = create_algorand_client()
    wallet_handle_token = ""
    wallet_password = "test-password"


    try:
        # =========================================================================
        # Step 1: Create a Test Wallet
        # =========================================================================
        print_step(1, "Creating a test wallet for multisig signing")


        test_wallet = create_test_wallet(kmd, wallet_password)
        wallet_handle_token = test_wallet["wallet_handle_token"]


        print_success(f"Test wallet created: {test_wallet['wallet_name']}")
        print_info(f"Wallet ID: {test_wallet['wallet_id']}")


        # =========================================================================
        # Step 2: Generate 3 Keys for Multisig Participants
        # =========================================================================
        print_step(2, "Generating 3 keys to use as multisig participants")


        participant_addresses: list[str] = []
        public_keys: list[bytes] = []
        num_participants = 3


        for i in range(1, num_participants + 1):
            generate_key_response = kmd.generate_key(
                GenerateKeyRequest(
                    wallet_handle_token=wallet_handle_token,
                )
            )
            address = generate_key_response.address
            participant_addresses.append(address)
            pk = public_key_from_address(address)
            public_keys.append(pk)
            print_info(f"Participant {i}: {shorten_address(address)}")


        print_success(f"Generated {num_participants} participant keys")


        # =========================================================================
        # Step 3: Create a 2-of-3 Multisig Account
        # =========================================================================
        print_step(3, "Creating a 2-of-3 multisig account")


        threshold = 2  # Minimum signatures required
        multisig_version = 1  # Multisig format version


        import_multisig_response = kmd.import_multisig(
            ImportMultisigRequest(
                multisig_version=multisig_version,
                public_keys=public_keys,
                threshold=threshold,
                wallet_handle_token=wallet_handle_token,
            )
        )
        multisig_address = import_multisig_response.address


        print_success("Multisig account created!")
        print_info(f"Multisig Address: {multisig_address}")
        print_info(f"Threshold: {threshold}-of-{num_participants}")


        # =========================================================================
        # Step 4: Fund the Multisig Account Using the Dispenser
        # =========================================================================
        print_step(4, "Funding the multisig account using the dispenser")


        dispenser = algorand.account.localnet_dispenser()
        print_info(f"Dispenser address: {shorten_address(dispenser.addr)}")


        # Fund the multisig account with 1 ALGO
        fund_amount = AlgoAmount.from_algo(1)
        algorand.send.payment(
            PaymentParams(
                sender=dispenser.addr,
                receiver=multisig_address,
                amount=fund_amount,
            )
        )


        # Verify funding
        account_info = algod.account_information(multisig_address)
        print_success(f"Multisig funded: {format_micro_algo(account_info.amount)}")


        # =========================================================================
        # Step 5: Create a Payment Transaction from the Multisig Account
        # =========================================================================
        print_step(5, "Creating a payment transaction from the multisig account")


        suggested_params = algod.suggested_params()


        print_info("Suggested Parameters:")
        print_info(f"  First Valid Round: {suggested_params.first_valid:,}")
        print_info(f"  Last Valid Round:  {suggested_params.last_valid:,}")
        print_info(f"  Genesis ID:        {suggested_params.genesis_id}")
        print_info("")


        # Create a payment back to the dispenser
        receiver_address = dispenser.addr
        payment_amount = 100_000  # 0.1 ALGO


        transaction_without_fee = Transaction(
            transaction_type=TransactionType.Payment,
            sender=multisig_address,
            first_valid=suggested_params.first_valid,
            last_valid=suggested_params.last_valid,
            genesis_hash=suggested_params.genesis_hash,
            genesis_id=suggested_params.genesis_id,
            payment=PaymentTransactionFields(
                receiver=receiver_address,
                amount=payment_amount,
            ),
        )


        # Assign the fee
        transaction = assign_fee(
            transaction_without_fee,
            fee_per_byte=suggested_params.fee,
            min_fee=suggested_params.min_fee,
        )


        tx_id = transaction.tx_id()


        print_success("Transaction created!")
        print_info(f"Transaction ID:  {tx_id}")
        print_info(f"Sender:          {shorten_address(multisig_address)} (multisig)")
        print_info(f"Receiver:        {shorten_address(receiver_address)}")
        print_info(f"Amount:          {format_micro_algo(payment_amount)}")
        print_info(f"Fee:             {format_micro_algo(transaction.fee or 0)}")


        # =========================================================================
        # Step 6: Sign with the First Participant (Partial Signature)
        # =========================================================================
        print_step(6, "Signing with the first participant (partial signature)")


        print_info(f"First signer: {shorten_address(participant_addresses[0])}")
        print_info("")


        tx_bytes = encode_transaction_raw(transaction)


        first_result = kmd.sign_multisig_transaction(
            SignMultisigTxnRequest(
                public_key=public_keys[0],
                transaction=tx_bytes,
                wallet_handle_token=wallet_handle_token,
                wallet_password=wallet_password,
            )
        )
        first_sign_result = first_result.multisig


        print_success("First signature obtained!")
        print_info("")
        print_info("sign_multisig_transaction() response fields:")
        print_info(f"  multisig: bytes ({len(first_sign_result)} bytes)")
        print_info("")


        # Decode and display the partial multisig signature
        partial_kmd_multisig = decode_kmd_multisig_response(first_sign_result)
        sig_count_1 = count_signatures(partial_kmd_multisig)


        print_info("Partial Multisig Signature:")
        print_info(f"  version:   {partial_kmd_multisig['version']}")
        print_info(f"  threshold: {partial_kmd_multisig['threshold']}")
        print_info(f"  subsigs:   {len(partial_kmd_multisig['subsignatures'])} participants")
        print_info(f"  Signatures collected: {sig_count_1} of {threshold} required")
        print_info("")
        print_info("Subsignature details:")
        for i, subsig in enumerate(partial_kmd_multisig["subsignatures"]):
            has_sig = subsig["signature"] is not None
            status = "SIGNED" if has_sig else "pending"
            addr = address_from_public_key(subsig["public_key"])
            print_info(f"  {i + 1}. {shorten_address(addr)} - {status}")


        print_info("")
        print_info("Note: With only 1 signature, the transaction cannot yet be submitted.")
        print_info(f"      We need {threshold} signatures ({threshold - sig_count_1} more required).")


        # =========================================================================
        # Step 7: Sign with the Second Participant (Complete the Multisig)
        # =========================================================================
        print_step(7, "Signing with the second participant (completing the signature)")


        print_info(f"Second signer: {shorten_address(participant_addresses[1])}")
        print_info("")
        print_info("Passing the partial multisig from Step 6 to collect the second signature...")
        print_info("")


        # Convert decoded dict to MultisigSig for passing back to KMD
        partial_msig_obj = MultisigSig(
            version=partial_kmd_multisig["version"],
            threshold=partial_kmd_multisig["threshold"],
            subsignatures=[
                MultisigSubsig(public_key=s["public_key"], signature=s["signature"])
                for s in partial_kmd_multisig["subsignatures"]
            ],
        )


        second_result = kmd.sign_multisig_transaction(
            SignMultisigTxnRequest(
                public_key=public_keys[1],
                transaction=tx_bytes,
                wallet_handle_token=wallet_handle_token,
                wallet_password=wallet_password,
                partial_multisig=partial_msig_obj,
            )
        )
        second_sign_result = second_result.multisig


        print_success("Second signature obtained!")
        print_info("")


        # Decode and display the completed multisig signature
        completed_kmd_multisig = decode_kmd_multisig_response(second_sign_result)
        sig_count_2 = count_signatures(completed_kmd_multisig)


        print_info("Completed Multisig Signature:")
        print_info(f"  version:   {completed_kmd_multisig['version']}")
        print_info(f"  threshold: {completed_kmd_multisig['threshold']}")
        print_info(f"  Signatures collected: {sig_count_2} of {threshold} required")
        print_info("")
        print_info("Subsignature details:")
        for i, subsig in enumerate(completed_kmd_multisig["subsignatures"]):
            has_sig = subsig["signature"] is not None
            status = "SIGNED" if has_sig else "pending"
            addr = address_from_public_key(subsig["public_key"])
            print_info(f"  {i + 1}. {shorten_address(addr)} - {status}")


        print_info("")
        print_success(f"Threshold met! {sig_count_2} >= {threshold} signatures collected.")
        print_info("The transaction is now fully authorized and ready for submission.")


        # =========================================================================
        # Step 8: Construct and Submit the Signed Transaction
        # =========================================================================
        print_step(8, "Constructing and submitting the multisig-signed transaction")


        # Convert KMD MultisigSig to transact's MultisigSignature for the SignedTransaction
        completed_multisig = kmd_multisig_to_transact_multisig(completed_kmd_multisig)


        # Build the signed transaction with the multisig signature
        signed_txn = SignedTransaction(
            txn=transaction,
            msig=completed_multisig,
        )


        # Encode and submit
        encoded_signed_txn = encode_signed_transaction(signed_txn)
        print_info(f"Encoded signed transaction: {len(encoded_signed_txn)} bytes")
        print_info("")


        submit_response = algod.send_raw_transaction(encoded_signed_txn)


        print_success("Transaction submitted!")
        print_info(f"Transaction ID: {submit_response.tx_id}")


        # =========================================================================
        # Step 9: Wait for Confirmation
        # =========================================================================
        print_step(9, "Waiting for confirmation")


        pending_info = wait_for_confirmation(algod, tx_id)


        confirmed_round = pending_info.confirmed_round or 0
        if confirmed_round > 0:
            print_success(f"Transaction confirmed in round {confirmed_round:,}")
        else:
            print_error("Transaction not confirmed within expected rounds")


        # =========================================================================
        # Step 10: Verify the Transaction
        # =========================================================================
        print_step(10, "Verifying the transaction was successful")


        # Check multisig account balance
        multisig_info = algod.account_information(multisig_address)
        expected_deduction = payment_amount + (transaction.fee or suggested_params.min_fee)
        expected_balance = fund_amount.micro_algo - expected_deduction


        print_info(f"Multisig balance before: {format_micro_algo(fund_amount.micro_algo)}")
        print_info(f"Multisig balance after:  {format_micro_algo(multisig_info.amount)}")
        print_info(f"Expected balance:        ~{format_micro_algo(expected_balance)}")
        print_info("")


        if multisig_info.amount <= fund_amount.micro_algo - payment_amount:
            print_success("Transaction verified! Balance reduced as expected.")


        # =========================================================================
        # Cleanup
        # =========================================================================
        print_step(11, "Cleaning up test wallet")


        cleanup_test_wallet(kmd, wallet_handle_token)
        wallet_handle_token = ""  # Mark as cleaned up


        print_success("Test wallet handle released")


        # =========================================================================
        # Summary
        # =========================================================================
        print_header("Summary")
        print_info("This example demonstrated multisig transaction signing with KMD:")
        print_info("")
        print_info("  sign_multisig_transaction() - Sign a multisig transaction")
        print_info("     Parameters:")
        print_info("       - wallet_handle_token: The wallet session token")
        print_info("       - wallet_password:     The wallet password")
        print_info("       - transaction:         The Transaction object to sign")
        print_info("       - public_key:          The public key of the signer (must be in wallet)")
        print_info("       - partial_multisig:    (optional) Existing partial signature to add to")
        print_info("     Returns:")
        print_info("       - multisig: bytes of the multisig signature (msgpack encoded)")
        print_info("")
        print_info("Multisig signing workflow:")
        print_info("  1. Create a multisig account with import_multisig()")
        print_info("  2. Fund the multisig account")
        print_info("  3. Create a transaction with the multisig address as sender")
        print_info("  4. Sign with first participant (returns partial multisig signature)")
        print_info("  5. Sign with additional participants, passing the partial signature")
        print_info("  6. Once threshold is met, construct SignedTransaction with msig field")
        print_info("  7. Encode and submit the signed transaction")
        print_info("")
        print_info("Key points:")
        print_info("  - Each signer adds their signature to the multisig structure")
        print_info("  - The partial_multisig parameter chains signatures together")
        print_info("  - The response contains a msgpack-encoded MultisigSignature")
        print_info("  - Transaction is valid once threshold signatures are collected")
        print_info("  - Any 2 of the 3 participants could have signed (2-of-3 threshold)")
        print_info("")
        print_info("Note: The test wallet remains in KMD (wallets cannot be deleted via API).")
    except Exception as e:
        print_error(f"Error: {e}")
        print_info("")
        print_info("Troubleshooting:")
        print_info("  - Ensure LocalNet is running: algokit localnet start")
        print_info("  - If LocalNet issues occur: algokit localnet reset")
        print_info("  - Check that KMD is accessible on port 4002")
        print_info("  - Check that Algod is accessible on port 4001")


        # Cleanup on error
        if wallet_handle_token:
            cleanup_test_wallet(kmd, wallet_handle_token)


        sys.exit(1)




if __name__ == "__main__":
    main()

Other examples in KMD Client

Section titled “Other examples in KMD Client”