nanograph-sdk 0.2.3

Official Python SDK for Nanograph

Nano SDK for Python

This package provides the Python implementation of the Nanograph SDK, which allows you to create node servers that can communicate with frontend clients like Volted Studio.

Installation

The package is available on PyPI and can be installed using pip:

pip install nanograph-sdk

Using with NanoCore

NanoCore is the orchestrator for your node servers. It manages both JavaScript and Python servers. To use your Python server with NanoCore:

1. Create a nanoserver.json in your project root (see Configuration section for all options)
2. Register your server with NanoCore:

nanocore register /path/to/your/server

This will:

• Validate your nanoserver.json
• Create a Python virtual environment
• Install dependencies if you have a requirements.txt

3. Start all registered servers:

nanocore start

NanoCore will:

• Start an asset server for file management
• Assign a port to your server
• Set up required environment variables (NANOCORE_HTTP_ENDPOINT, NANOCORE_TOKEN)
• Start and monitor your server process
• Restart it if configuration changes

For additional commands and features, please refer to the NanoCore documentation.

Usage

Creating a Server

from nanograph_sdk import NanoSDK
import asyncio

# Initialize SDK (configuration is loaded from nanoserver.json)
sdk = NanoSDK()

# Start the server
async def main():
    await sdk.start()
    print('Python Server started')

# Handle shutdown
async def shutdown_handler():
    print('Python Server is shutting down')
    # Add any cleanup logic here

sdk.on_shutdown(shutdown_handler)

# Graceful shutdown
async def run():
    try:
        await main()
    except KeyboardInterrupt:
        print('Interrupted, stopping server...')
    finally:
        await sdk.stop()

if __name__ == '__main__':
    asyncio.run(run())

Configuration

The SDK requires a nanoserver.json file in your project root. Here's a complete example with all available options:

{
    "domain": "local-python.nanograph",     // Required: Domain to group servers
    "serverDisplayName": "My Python Server",// Required: Display name of your server
    "serverUid": "my-python-server",        // Required: Unique server identifier
    "serverPackageId": "my-python-server-package", // Optional: Package identifier for your server (e.g. npm package name)
    "language": "python",                   // Required: Must be 'python'
    "port": 3017,                          // Optional: HTTP port (default: 3017)
    "nodesPath": "nodes",                  // Optional: Path to nodes directory
    "autoWatch": true,                     // Optional: Auto-reload on changes
    "watchDebounceTime": 500               // Optional: Debounce time for reload
}

Key	Type	Default	Description
domain	str	—	Domain to group servers (required)
serverDisplayName	str	—	Display name of your server (required)
serverPackageId	str	—	Package identifier for your server (optional)
language	str	—	Must be 'python' for Python servers (required)
port	int	3017	HTTP port to listen on
nodesPath	str	'nodes'	Path to the directory containing node files
autoWatch	bool	True	If true, automatically reload nodes on file changes
watchDebounceTime	int	500	Debounce time in milliseconds for file watcher reloads

Note: The port can be overridden by setting the PORT environment variable.

Asset Handling

Assets are identified by their content-addressed SHA-256 hash:

sha256:<64-hex-chars>

For example: sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855

When running under NanoCore, the SDK automatically points asset requests at the NANOCORE_HTTP_ENDPOINT it receives via environment variables (e.g., https://host:3001).

The SDK provides built-in support for handling assets through the following methods:

# Instance methods
await sdk.resolve_asset(hash, options)  # Resolve an asset hash to data
sdk.get_asset_download_url(hash)        # Get direct download URL
await sdk.get_asset_presigned_url(hash) # Get a presigned URL
await sdk.upload_asset(file, options)   # Upload an asset (returns result with result.hash)

# Static methods (can be used without SDK instance)
await NanoSDK.resolve_asset_static(hash, options)
NanoSDK.get_asset_download_url_static(hash)
await NanoSDK.get_asset_presigned_url_static(hash)
await NanoSDK.upload_asset_static(file, options)

To use asset handling capabilities, the following environment variables must be set:

• NANOCORE_HTTP_ENDPOINT: The endpoint URL for the Nanocore asset server
• NANOCORE_TOKEN: Authentication token for accessing the asset server

Node Initialization

Nodes can have an optional async initialization function that will be called when the node is loaded:

from nanograph_sdk import NanoSDK, NodeDefinition

# Define the node
definition = {
    'uid': 'my-node',
    'name': 'My Node',
    # ... other definition fields ...
}

# Create node instance
node = NanoSDK.register_node(definition)

# Optional async initialization function
async def init(node_instance):
    # Perform any async initialization here
    # This will be called when the node is loaded
    pass

# Export both the node and init function
export = node

Creating Nodes

from nanograph_sdk import NanoSDK, NodeDefinition, NodeInstance, ExecutionContext

# Define the node
definition = {
    'uid': 'my-unique-python-node-id',
    'name': 'My Python Node',
    'category': 'Processing',
    'version': '1.0.0',
    'description': 'Description of my python node',
    'inputs': [
        {'name': 'input1', 'type': 'string', 'description': 'First input'}
    ],
    'outputs': [
        {'name': 'output1', 'type': 'string', 'description': 'First output'}
    ],
    'parameters': [
        {
            'name': 'param1',
            'type': 'boolean',
            'value': True,
            'default': True,
            'label': 'Parameter 1',
            'description': 'Description of parameter 1'
        }
    ]
}

# Register the node
my_node = NanoSDK.register_node(definition)

# Implement the execution logic
async def execute_node(ctx: ExecutionContext):
    # Get input values
    input1 = ctx.inputs.get('input1', '')
    
    # Send status update
    await ctx.context['send_status']({'type': 'running', 'message': 'Processing...'})
    
    # Check for abort
    if ctx.context['is_aborted']():
        raise Exception('Execution aborted')
    
    # Process the inputs
    output1 = f'Processed by Python: {input1}'
    
    # Return the outputs
    return {'output1': output1}

my_node['execute'] = execute_node

# To export the node if it's in its own file:
# export = my_node 

Nodes are defined in `node.py` files. You can organize your nodes by placing each `node.py`
file (along with any helper modules it might need) into its own subdirectory within the
main `nodes` directory (or the path specified in `nodes_path` in the SDK configuration).
The SDK will scan these directories for `node.py` files to load the definitions.

---

## ExecutionContext Reference

When you implement a node's `execute` function, it receives a single argument: `ctx` (the execution context). This object provides everything your node needs to process inputs, parameters, and interact with the workflow engine.

**The `ExecutionContext` object has the following structure:**

| Field         | Type                | Description                                                                 |
|---------------|---------------------|-----------------------------------------------------------------------------|
| `inputs`      | `dict`              | Input values for this node, keyed by input name.                            |
| `parameters`  | `list`              | List of parameter dicts for this node (see your node definition).           |
| `context`     | `dict`              | Runtime context utilities and metadata (see below).                         |

### `ctx.context` fields

| Key            | Type        | Description                                                                 |
|----------------|-------------|-----------------------------------------------------------------------------|
| `send_status`  | `callable`  | `await ctx.context['send_status']({...})` to send a status/progress update. |
| `is_aborted`   | `callable`  | `ctx.context['is_aborted']()` returns `True` if execution was aborted.      |
| `graph_node`   | `dict`      | The full graph node definition (with position, etc).                        |
| `instance_id`  | `str`       | The workflow instance ID for this execution.                                |

**Example usage in a node:**

```python
async def execute_node(ctx):
    # Access input
    value = ctx.inputs.get('input1')
    # Access parameter
    param = next((p for p in ctx.parameters if p['name'] == 'param1'), None)
    # Send a running status
    await ctx.context['send_status']({'type': 'running', 'message': 'Working...'})
    # Check for abort
    if ctx.context['is_aborted']():
        raise Exception('Aborted!')
    # ...

---

NodeStatus Reference

The NodeStatus object is used to communicate the current status, progress, or result of a node execution back to the orchestrator. You send it using await ctx.context['send_status'](status) from within your node's execute function.

NodeStatus fields:

Field	Type	Description
type	str	One of: 'idle', 'running', 'complete', 'error', 'missing'
message	str (optional)	Human-readable status or error message
progress	dict (optional)	Progress info, e.g. { 'step': 2, 'total': 5 }
outputs	dict (optional)	Output values (only for 'complete' status)

Example: Sending progress updates from a node

async def execute_node(ctx):
    total_steps = 5
    for step in range(1, total_steps + 1):
        # Abort fast if needed
        if ctx.context['is_aborted']():
            raise Exception('Aborted!')
        # Simulate work
        await asyncio.sleep(1)
        # Send progress update
        await ctx.context['send_status']({
            'type': 'running',
            'message': f'Processing step {step}/{total_steps}',
            'progress': {'step': step, 'total': total_steps}
        })
    # Just return the outputs; the SDK will send the 'complete' status automatically
    return {'result': 'done'}

Note: You do not need to manually send a 'complete' status at the end. The SDK will automatically send a 'complete' status with the outputs you return from your execute function.

---

Folder Structure

Recommended project structure for a Python NanoServer:

my-python-nodeserver/
├── main.py           # Entry point
├── nanoserver.json   # Server configuration (required)
├── nodes/            # Nodes directory (scans for node.py files in subdirectories)
│   ├── processing/   # Category directory (optional organization)
│   │   ├── simple_text_node/   # Directory for a single node
│   │   │   └── node.py          # Node definition for simple_text_node
│   │   └── complex_math_node/ # Directory for a more complex node
│   │       ├── __init__.py    # Optional, makes 'complex_math_node' a Python package
│   │       ├── node.py        # Main node definition for complex_math_node
│   │       └── math_utils.py  # Helper functions specific to this node
│   └── another_category/      # Another category directory
│       └── another_node/      # Directory for another_node
│           └── node.py        # Node definition for another_node
├── pyproject.toml    # Dependencies and package info
└── README.md

License

MIT