DigitalKin Python SDK

The DigitalKin Python SDK for building and managing modules within the DigitalKin agentic mesh. Create custom Tools and Archetypes that communicate over gRPC, register with a service mesh, and scale independently.

Features

Async-native gRPC module system — every module is a gRPC server built on grpcio with full async support
Typed module contracts — Pydantic models for Input, Output, Setup, and Secret schemas with protocol-based trigger dispatch
Module-to-module communication — tools and archetypes discover each other via the registry and exchange requests over gRPC
Tool resolution — archetypes dynamically resolve and invoke tool modules at runtime
Admission queue & backpressure — built-in request admission with configurable concurrency limits
Healthcheck protocols — automatic ping, services, and status healthcheck triggers registered on every module
Profiling — optional [profiling] extra with asyncio-inspector, pyinstrument, viztracer, and yappi
Batched history writes — efficient storage writes for conversation history
Redis Streams — durable message passing, crash recovery, and reconnection

Installation

# With uv (recommended)
uv add digitalkin

# With pip
pip install digitalkin

Optional extras:

# Async profiling tools
uv add "digitalkin[profiling]"

# uvloop for faster event loop
uv add "digitalkin[performance]"

Quick Start

1. Define your models

from pydantic import BaseModel
from digitalkin.models.module.base_types import DataModel, DataTrigger


class MessageInput(DataTrigger):
    protocol: str = "message"
    content: str


class InputModel(DataModel[MessageInput]):
    root: MessageInput


class MessageOutput(DataTrigger):
    protocol: str = "message"
    reply: str


class OutputModel(DataModel[MessageOutput]):
    root: MessageOutput

2. Create a module and trigger

from digitalkin import ArchetypeModule, ModuleContext, TriggerHandler
from digitalkin.models.module.setup_types import SetupModel


class MyArchetype(ArchetypeModule[InputModel, OutputModel, SetupModel, BaseModel]):
    async def initialize(self, context: ModuleContext, setup_data: SetupModel) -> None:
        pass

    async def cleanup(self, context: ModuleContext) -> None:
        pass


@MyArchetype.register
class MessageTrigger(TriggerHandler[InputModel, SetupModel, OutputModel]):
    protocol = "message"
    input_format = InputModel
    output_format = OutputModel

    def __init__(self, context: ModuleContext) -> None:
        super().__init__(context)

    async def handle(
        self,
        input_data: InputModel,
        setup_data: SetupModel,
        context: ModuleContext,
    ) -> None:
        output = OutputModel(root=MessageOutput(reply=f"Echo: {input_data.root.content}"))
        await self.send_message(context, output)

3. Run the server

import asyncio
from digitalkin.grpc_servers.module_server import ModuleServer

async def main() -> None:
    server = ModuleServer(MyArchetype)
    await server.start_async()
    await server.await_termination()

asyncio.run(main())

Redis Gateway

The embedded gateway enables real-time bidirectional communication between modules via Redis Streams, with crash recovery and horizontal scaling.

Durable Streaming: Output persisted to Redis Streams — reconnection via from_seq.
Zero-Copy Proto: Binary proto serialization to Redis — no JSON intermediary.
Horizontal Scaling: Each module instance embeds its own gateway. Scale by adding replicas behind a load balancer.

Development

Prerequisites

Python 3.10+
uv — modern Python package management
Task — task runner

Setup

git clone --recurse-submodules https://github.com/DigitalKin-ai/digitalkin.git
cd digitalkin

task setup-dev
source .venv/bin/activate

Common Tasks

task linter               # Format + lint (ruff) + type check (mypy)
task check                # Linter + mypy + tests
task run-tests            # Run pytest via Docker
task build-package        # Build distribution
task bump-version -- patch|minor|major

task docs-serve           # Serve docs locally (mkdocs)
task docs-build           # Build docs

task generate-certificates  # Generate mTLS certs for gRPC
task clean                # Remove build artifacts + __pycache__
task clean-all            # Above + remove .venv

Publishing Process

Update code and commit changes (following conventional branch/commit standard).
Use task bump-version -- major|minor|patch to commit the new version.
Use GitHub "Create Release" workflow to publish the new version.
Workflow automatically publishes to Test PyPI and PyPI.

License

This project is licensed under the terms specified in the LICENSE file.

For more information, visit our Documentation or report issues on our Issues page.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

digitalkin-1.0.0.tar.gz (247.5 kB view details)

Uploaded Jul 21, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

digitalkin-1.0.0-py3-none-any.whl (335.5 kB view details)

Uploaded Jul 21, 2026 Python 3

File details

Details for the file digitalkin-1.0.0.tar.gz.

File metadata

Download URL: digitalkin-1.0.0.tar.gz
Upload date: Jul 21, 2026
Size: 247.5 kB
Tags: Source
Uploaded using Trusted Publishing? Yes
Uploaded via: twine/6.1.0 CPython/3.13.14

File hashes

Hashes for digitalkin-1.0.0.tar.gz
Algorithm	Hash digest
SHA256	`6a50b15d19b6c7379ef679c82e0f5144d12c9df9cd966490a14c777c71076915`
MD5	`0292178bb2b64f54ff08bb68ec62c184`
BLAKE2b-256	`b9d3603f19aa82310da66972445ec5bc467c5f05d607686ee49d1fbc6639e56d`

See more details on using hashes here.

Provenance

The following attestation bundles were made for digitalkin-1.0.0.tar.gz:

Publisher: release.yml on DigitalKin-ai/digitalkin

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Statement:
- Statement type: https://in-toto.io/Statement/v1
- Predicate type: https://docs.pypi.org/attestations/publish/v1
- Subject name: digitalkin-1.0.0.tar.gz
- Subject digest: 6a50b15d19b6c7379ef679c82e0f5144d12c9df9cd966490a14c777c71076915
- Sigstore transparency entry: 2211485612
- Sigstore integration time: Jul 21, 2026
Source repository:
- Permalink: DigitalKin-ai/digitalkin@fb47cd3e107ba696f689be8b8c15be9aee44e985
- Branch / Tag: refs/tags/v1.0.0
- Owner: https://github.com/DigitalKin-ai
- Access: public
Publication detail:
- Token Issuer: https://token.actions.githubusercontent.com
- Runner Environment: github-hosted
- Publication workflow: release.yml@fb47cd3e107ba696f689be8b8c15be9aee44e985
- Trigger Event: release