### Complete Example: Calling a Calculation Tool

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

A full example demonstrating how to connect to an MCP server, retrieve available tools, and call a specific calculation tool with its arguments. Includes asynchronous execution setup.

```python
import asyncio
from pathlib import Path
from dp.agent.client import MCPClient

async def run_calculation():
    # Connect to calculation MCP server
    client = MCPClient("http://localhost:8000/sse")
    session = await client.get_session()
    
    # Get available tools
    tools = await session.list_tools()
    print(f"Available tools: {[t.name for t in tools.tools]}")
    
    # Call a calculation tool
    result = await session.call_tool(
        name="run_simulation",
        arguments={
            "input_data": "local:///path/to/data.npy",
            "num_steps": 1000,
            "temperature": 300.0
        }
    )
    
    print(f"Calculation result: {result}")

if __name__ == "__main__":
    asyncio.run(run_calculation())
```

--------------------------------

### Python Example: Getting MQTT Instance and Sending Control

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Shows how to obtain the MQTTCloud client instance and use it to send a device control command.

```python
from dp.agent.cloud import get_mqtt_cloud_instance

mqtt = get_mqtt_cloud_instance()
request_id = mqtt.send_device_control(...)
```

--------------------------------

### Complete Python Example: Cloud Device Control

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

A comprehensive example demonstrating device definition, MCP server setup, and client-side interaction including sending control commands and waiting for results.

```python
import asyncio
from dp.agent.cloud import mcp, get_mqtt_cloud_instance
from dp.agent.device.device import Device, action, register_mcp_tools, BaseParams, SuccessResult

# Define device
class TescanDevice(Device):
    device_name = "tescan_microscope"
    
    @action("take_picture")
    def take_picture(self, params: BaseParams) -> SuccessResult:
        hw = params.get("horizontal_width", "default")
        return SuccessResult(
            message=f"Picture taken with width {hw}",
            data={"image_id": "pic_123"}
        )

# MCP server setup
def main():
    # Create device
    device = TescanDevice()
    
    # Register actions as MCP tools
    register_mcp_tools(mcp, device)
    
    # Start MCP server
    print("Starting MCP server...")
    mcp.run(transport="sse")

# Client code example
async def client_example():
    mqtt = get_mqtt_cloud_instance()
    
    # Send control
    request_id = mqtt.send_device_control(
        device_name="tescan_microscope",
        device_action="take_picture",
        device_params={"horizontal_width": "2048"}
    )
    
    # Wait for result
    result = mqtt.wait_for_status_update(request_id, timeout=30.0)
    if result:
        print(f"Success: {result['result']}")
    else:
        print("Timeout")
    
    mqtt.stop()

if __name__ == "__main__":
    main()
```

--------------------------------

### DeviceTwin Run Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Starts the DeviceTwin to listen for control commands. This method blocks execution.

```python
device = MyDevice()
twin = DeviceTwin(device)
twin.run()  # Blocks and listens for commands
```

--------------------------------

### DeviceTwin Initialization Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Initializes the DeviceTwin with a device instance and an optional environment path.

```python
from dp.agent.device import DeviceTwin

device = MyDevice()
twin = DeviceTwin(device, env_path=".env")
twin.run()
```

--------------------------------

### Device Initialization Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Example of creating a custom device class that inherits from Device and initializing it. Actions are registered during this initialization.

```python
class MyDevice(Device):
    device_name = "my_device"
    
    @action("do_something")
    def do_something(self, params: MyParams) -> SuccessResult:
        return SuccessResult("Done")

device = MyDevice()  # Actions are registered here
```

--------------------------------

### Get Available Actions Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Example of using the get_available_actions class method to list all available actions for a device class.

```python
actions = MyDevice.get_available_actions()
# Returns: {"do_something", "another_action", ...}
```

--------------------------------

### Register MCP Tools and Run Server

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Example of registering device tools with the global MCP instance and starting the MCP server. The server can be run with different transports, such as 'sse'.

```python
from dp.agent.cloud import mcp
from dp.agent.device.device import register_mcp_tools

device = MyDevice()
register_mcp_tools(mcp, device)

# Start the MCP server
mcp.run(transport="sse")
```

--------------------------------

### Complete Device Implementation with DeviceTwin

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

A full example demonstrating a custom Device implementation with actions and running it via DeviceTwin.

```python
from typing import TypedDict
from dp.agent.device.device import Device, action, BaseParams, SuccessResult
from dp.agent.device import DeviceTwin

# Define parameter types
class CalibrateParams(BaseParams):
    temperature: float
    pressure: float

# Define result type
class CalibrationResult(SuccessResult):
    pass

# Implement device
class AnalysisDevice(Device):
    device_name = "analysis_system"
    
    @action("calibrate")
    def calibrate(self, params: CalibrateParams) -> CalibrationResult:
        temp = params.get("temperature", 20.0)
        press = params.get("pressure", 1.0)
        
        # Perform calibration
        print(f"Calibrating at {temp}°C and {press} atm")
        
        return CalibrationResult(
            message="Calibration successful",
            data=None
        )
    
    @action("reset")
    def reset(self, params: BaseParams) -> SuccessResult:
        print("Resetting device")
        return SuccessResult("Device reset complete")

# Run as device twin
if __name__ == "__main__":
    device = AnalysisDevice()
    twin = DeviceTwin(device)
    twin.run()
```

--------------------------------

### Register MCP Tools and Run

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Example of registering MCP tools with a device and running the MCP transport.

```python
from dp.agent.cloud import mcp, get_mqtt_cloud_instance
from dp.agent.device.device import register_mcp_tools

device = MyDevice()
register_mcp_tools(mcp, device)
mcp.run(transport="sse")
```

--------------------------------

### Complete Workflow Example: Create Project and Run Tool

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Demonstrates the end-to-end workflow for setting up a new calculation project, implementing a tool, and running the MCP server.

```bash
# 1. Create new project
mkdir my_science_project
cd my_science_project

# 2. Generate calculation scaffolding
dp-agent fetch scaffolding --type calculation

# 3. Download configuration template
dp-agent fetch config

# 4. Edit .env with your credentials
nano .env

# 5. Implement your calculation tool
nano calculation/simple.py

# 6. Run calculation MCP server
dp-agent run tool calculation
# Server now listening for tool calls

# In another terminal, call the server:
# python -c "
# import asyncio
# from dp.agent.client import MCPClient
# 
# async def test():
#     client = MCPClient('server.py')
#     session = await client.get_session()
#     result = await session.call_tool('my_tool', {{}})
#     print(result)
# 
# asyncio.run(test())
# "
```

--------------------------------

### DeviceTwin.run

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Starts the device twin, connecting to the MQTT broker, subscribing to control topics, and dispatching commands.

```APIDOC
## DeviceTwin.run

### Description
Start the device twin and listen for control commands. This method connects to the MQTT broker, subscribes to the device control topic, dispatches incoming control messages, and publishes execution results.

### Method
run

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
None

### Returns
None

### Behavior:
- Connects to MQTT broker using configured credentials
- Subscribes to device control topic
- Dispatches incoming control messages to the device
- Publishes execution results to device status topic
- Blocks until interrupted (Ctrl+C)

### Request Example
```python
device = MyDevice()
twin = DeviceTwin(device)
twin.run()  # Blocks and listens for commands
```

### Response
#### Success Response (None)
None

#### Response Example
None
```

--------------------------------

### Starting the MCP Server with run()

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Starts the MCP server, listening for tool invocation requests and handling artifact processing and job execution. Supports 'sse' or 'stdio' transport.

```python
def run(self, transport: str = "sse", ...)
```

--------------------------------

### Minimal Local Development .env Configuration

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Example .env file for minimal local testing, including MQTT and Redis configurations.

```bash
# .env - Minimal for local testing
MQTT_INSTANCE_ID=local-test
MQTT_ENDPOINT=localhost
MQTT_DEVICE_ID=test_device
MQTT_GROUP_ID=test_group
MQTT_AK=test_ak
MQTT_SK=test_sk

REDIS_HOST=localhost
REDIS_PORT=6379
```

--------------------------------

### Complete Example: Calling a Calculation Tool

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Demonstrates a complete workflow of connecting to an MCP server, obtaining a session, listing available tools, and then calling a specific tool named 'run_simulation' with provided arguments.

```APIDOC
## Complete Example: Calling a Calculation Tool

### Description
Demonstrates a complete workflow of connecting to an MCP server, obtaining a session, listing available tools, and then calling a specific tool named 'run_simulation' with provided arguments.

### Example:
```python
import asyncio
from pathlib import Path
from dp.agent.client import MCPClient

async def run_calculation():
    # Connect to calculation MCP server
    client = MCPClient("http://localhost:8000/sse")
    session = await client.get_session()
    
    # Get available tools
    tools = await session.list_tools()
    print(f"Available tools: {[t.name for t in tools.tools]}")
    
    # Call a calculation tool
    result = await session.call_tool(
        name="run_simulation",
        arguments={
            "input_data": "local:///path/to/data.npy",
            "num_steps": 1000,
            "temperature": 300.0
        }
    )
    
    print(f"Calculation result: {result}")

if __name__ == "__main__":
    asyncio.run(run_calculation())
```
```

--------------------------------

### Dispatch Device Action Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Example of calling the dispatch_device_actions method to execute a registered action on a specific device.

```python
device = MyDevice()
result = device.dispatch_device_actions(
    device_name="my_device",
    device_action="do_something",
    device_params={"param1": "value"}
)
print(result.status)  # "success" or "error"
```

--------------------------------

### Bash Example: Fetching Configuration for .env File

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Command to fetch a configuration template, typically used to create a local .env file for environment variable settings.

```bash
# Copy template
dp-agent fetch config
```

--------------------------------

### Complete Example: Deep Potential Model Training

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

This snippet demonstrates how to set up and run a Deep Potential model training process using the CalculationMCPServer. It includes a tool definition with default preprocessing and simulates the training process.

```python
from pathlib import Path
from typing import Optional, Literal, TypedDict
import time
import os
from dp.agent.server import CalculationMCPServer

mcp = CalculationMCPServer("DPTrainer")

def set_defaults(executor, storage, kwargs):
    """Set default container image for Bohrium execution."""
    if executor and executor.get("type") == "dispatcher":
        machine = executor.get("machine", {})
        remote_profile = machine.get("remote_profile", {})
        input_data = remote_profile.get("input_data", {})
        input_data["image_name"] = input_data.get(
            "image_name",
            "registry.dp.tech/dptech/ubuntu:22.04-py3.10"
        )
    return executor, storage, kwargs

@mcp.tool(preprocess_func=set_defaults)
def train_deep_potential(
    training_data: Path,
    validation_data: Optional[Path] = None,
    model_type: Literal["se_e2_a", "dpa2", "dpa3"] = "dpa3",
    rcut: float = 9.0,
    rcut_smth: float = 8.0,
    sel: int = 120,
    numb_steps: int = 1000000,
    decay_steps: int = 5000,
    start_lr: float = 0.001,
) -> TypedDict("results", {
    "model": Path,
    "log": Path,
    "lcurve": Path
}):
    """
    Train a Deep Potential model on training data.
    
    Args:
        training_data: Training data in DeePMD npy format
        validation_data: Optional validation data
        model_type: Model architecture type
        rcut: Cutoff radius in Angstroms
        rcut_smth: Smooth cutoff radius
        sel: Max neighbors in cutoff
        numb_steps: Total training steps
        decay_steps: Learning rate decay period
        start_lr: Initial learning rate
    
    Returns:
        Dictionary with:
        - model: Trained potential model file
        - log: Training log directory
        - lcurve: Learning curve output
    """
    print(f"Training {model_type} model with {numb_steps} steps")
    
    # Simulate training
    time.sleep(2)
    
    # Create output files
    with open("model.pt", "w") as f:
        f.write("Trained model data")
    os.makedirs("logs", exist_ok=True)
    with open("logs/training.log", "w") as f:
        f.write("Training log")
    with open("lcurve.out", "w") as f:
        f.write("Learning curve")
    
    return {
        "model": Path("model.pt"),
        "log": Path("logs"),
        "lcurve": Path("lcurve.out")
    }

if __name__ == "__main__":
    mcp.run(transport="sse")

```

--------------------------------

### Production with Bohrium Code Setup

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Python code to set up a production environment with CalculationMCPServer, specifying executor and storage types. Includes a sample tool registration.

```python
from dp.agent.server import CalculationMCPServer
from pathlib import Path

mcp = CalculationMCPServer(
    "ProductionCalculator",
    executor={"type": "dispatcher", "machine": {...}},
    storage={"type": "bohrium"}
)

@mcp.tool()
def production_tool(data: Path) -> Path:
    return Path("output.tgz")

if __name__ == "__main__":
    mcp.run(transport="sse")
```

--------------------------------

### Python Example: Setting a Status Update Callback

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Demonstrates how to send a device control command and register an asynchronous callback to process its status update.

```python
mqtt = get_mqtt_cloud_instance()

async def on_status_update(response):
    print(f"Device action completed: {response['result']}")

request_id = mqtt.send_device_control(...)
mqtt.set_callback(request_id, on_status_update)
```

--------------------------------

### Minimal Local Development Code Setup

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Python code to set up a local development environment using DeviceTwin, which loads configuration from the .env file.

```python
from dp.agent.device import DeviceTwin, Device
from dp.agent.device.device import action

class TestDevice(Device):
    device_name = "test_device"
    
    @action("test")
    def test_action(self, params):
        return SuccessResult("Test passed")

device = TestDevice()
twin = DeviceTwin(device)  # Uses .env configuration
twin.run()
```

--------------------------------

### run() Method

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Starts the Calculation MCP server, listening for tool invocation requests and handling artifact processing and job execution.

```APIDOC
## run() Method

### Description
Starts the MCP server to listen for tool invocation requests and handle artifact processing and job execution.

### Method
Instance Method

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Method Signature
```python
run(self, transport: str = "sse", ...)
```

### Parameters
- **transport** (str) - Default: "sse" - Transport protocol: "sse" or "stdio"

### Behavior
- Starts FastMCP server with specified transport
- Listens for tool invocation requests
- Handles artifact processing and job execution
```

--------------------------------

### Scaffold and Run Calculation Project Workflow

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/index.md

Guide to setting up a calculation project, generating scaffolding, implementing calculations, running the MCP server, and calling it from a client.

```bash
# 1. Create directory
mkdir my_calc && cd my_calc

# 2. Generate scaffolding
dp-agent fetch scaffolding --type calculation

# 3. Fetch config (optional)
dp-agent fetch config

# 4. Implement your calculation
nano calculation/simple.py

# 5. Run MCP server
dp-agent run tool calculation

# 6. In another terminal, call it
python -c "
import asyncio
from dp.agent.client import MCPClient

async def test():
    client = MCPClient('server.py')
    session = await client.get_session()
    result = await session.call_tool('my_tool', {'input': 'data'})
    print(result)

asyncio.run(test())
"
```

--------------------------------

### Python Example: Sending Control and Waiting for Status

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Demonstrates sending a device control command and then waiting for its status update with a specified timeout.

```python
mqtt = get_mqtt_cloud_instance()

request_id = mqtt.send_device_control(
    device_name="microscope_001",
    device_action="take_picture",
    device_params={}
)

response = mqtt.wait_for_status_update(request_id, timeout=30.0)
if response:
    print(f"Result: {response['result']}")
else:
    print("Timeout waiting for response")
```

--------------------------------

### DispatcherExecutor Machine Configuration Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Example of machine configuration for DispatcherExecutor, specifying batch and context types for Bohrium.

```python
{
    "batch_type": "Bohrium",
    "context_type": "Bohrium",
    "remote_profile": {
        "input_data": {
            "image_name": "registry.dp.tech/dptech/ubuntu:22.04-py3.10",
            "job_type": "container",
            "platform": "ali",
            "scass_type": "c2_m4_cpu"
        }
    }
}
```

--------------------------------

### Run Agent with Configuration

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Starts the complete agent with its web UI, including configuration options. This command initiates the React-based web application, backend services, and loads agent configuration.

```bash
dp-agent run agent --config
# Opens web UI at http://localhost:3000
```

--------------------------------

### Environment Configuration for Bohr Agent

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Example `.env` file content for configuring MQTT connection and computing resources.

```bash
# MQTT connection configuration
MQTT_INSTANCE_ID=your_instance_id
MQTT_ENDPOINT=your_endpoint
MQTT_DEVICE_ID=your_device_id
MQTT_GROUP_ID=your_group_id
MQTT_AK=your_access_key
MQTT_SK=your_secret_key

# Computing resource configuration
BOHRIUM_USERNAME=your_username
BOHRIUM_PASSWORD=your_password
```

--------------------------------

### Example Preprocessor Usage

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Demonstrates how to define and use a custom preprocessor function with a tool to set a default Docker image for the executor if it's of type 'dispatcher'.

```python
def set_default_image(executor, storage, kwargs):
    if executor is None:
        executor = {}
    if executor.get("type") == "dispatcher":
        machine = executor.setdefault("machine", {})
        remote_profile = machine.setdefault("remote_profile", {})
        input_data = remote_profile.setdefault("input_data", {})
        input_data.setdefault("image_name", 
                             "registry.dp.tech/dptech/pytorch:latest")
    return executor, storage, kwargs

mcp = CalculationMCPServer("MyServer", executor={...})

@mcp.tool(preprocess_func=set_default_image)
def train_model(data: Path) -> Path:
    return Path("model.pt")
```

--------------------------------

### Create device control project scaffolding

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Example of creating a new project for device control tasks. This involves creating a directory and then running the scaffolding command.

```bash
# Create device control project
mkdir my_device_project
cd my_device_project
dp-agent fetch scaffolding --type device
```

--------------------------------

### Create calculation project scaffolding

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Example of creating a new project for calculation tasks. This involves creating a directory and then running the scaffolding command.

```bash
# Create calculation project
mkdir my_calc_project
cd my_calc_project
dp-agent fetch scaffolding --type calculation
```

--------------------------------

### Production with Bohrium .env Configuration

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Example .env file for a production environment using Bohrium, detailing MQTT, Redis, Bohrium, and storage configurations.

```bash
# .env - Production configuration
# MQTT (Cloud)
MQTT_INSTANCE_ID=prod-instance-123
MQTT_ENDPOINT=prod-mqtt.aliyuncs.com
MQTT_DEVICE_ID=prod_microscope_001
MQTT_GROUP_ID=prod_group
MQTT_AK=prod_ak_value
MQTT_SK=prod_sk_value

# Redis
REDIS_HOST=prod-redis.example.com
REDIS_PORT=6379
REDIS_PASSWORD=secure_password

# Bohrium
BOHRIUM_USERNAME=user
BOHRIUM_PASSWORD=password
BOHRIUM_PROJECT_ID=12345
BOHRIUM_ACCESS_KEY=ak_prod_value
BOHRIUM_APP_KEY=myapp

# Storage
BOHRIUM_BOHRIUM_URL=https://bohrium.dp.tech
```

--------------------------------

### Install Bohrium Science Agent SDK

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Install the bohr-agent-sdk package using pip. It is recommended to upgrade to the latest version.

```bash
pip install bohr-agent-sdk -i https://pypi.org/simple --upgrade
```

--------------------------------

### ClientSession Interface Examples

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Demonstrates common operations available through the ClientSession interface, including discovering tools, calling tools with arguments, retrieving input schemas, listing resources, and reading resource content.

```python
# Discover tools
tools = await session.list_tools()  # Get available tools

# Call a tool
result = await session.call_tool(name="tool_name", arguments={...})

# Get tool input schema
schema = await session.get_input_schema("tool_name")

# Discover resources
resources = await session.list_resources()

# Read a resource
content = await session.read_resource(uri="resource://path")
```

--------------------------------

### Python Example: Stopping the MQTT Client

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Illustrates the usage of the stop() method to gracefully disconnect the MQTT client and clean up resources.

```python
mqtt = get_mqtt_cloud_instance()
# ... use mqtt ...
mqtt.stop()  # Cleanup
```

--------------------------------

### Calculation Tool Server

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/README.md

Set up a server to run calculations using MCP. This example defines a calculation tool that processes an input file and returns an output file path.

```python
from dp.agent.server import CalculationMCPServer
from pathlib import Path

mcp = CalculationMCPServer("MyCalculator")

@mcp.tool()
def run_calculation(input_file: Path) -> Path:
    # Process input_file
    return Path("output.txt")

if __name__ == "__main__":
    mcp.run(transport="sse")
```

--------------------------------

### Cloud Mode Agent Development with MCP

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Example demonstrating MCP protocol-based cloud device control, including signal handling and device tool registration.

```python
"""
MCP protocol-based cloud device control example
"""
import signal
import sys
from dp.agent.cloud import mcp, get_mqtt_cloud_instance
from dp.agent.device.device import TescanDevice, register_mcp_tools

def signal_handler(sig, frame):
    """Graceful shutdown handling"""
    print("Shutting down...")
    get_mqtt_cloud_instance().stop()
    sys.exit(0)

def main():
    """Start cloud services"""
    print("Starting Tescan Device Twin Cloud Services...")

    # Register signal handler
    signal.signal(signal.SIGINT, signal_handler)

    # Create device instance
    device = TescanDevice(mcp, device)

    # Automatically register device tools to MCP server
    # register_mcp_tools implements automatic registration through Python introspection
    register_mcp_tools(device)

    # Start MCP server
    print("Starting MCP server...")
    mcp.run(transport="sse")

if __name__ == "__main__":
    main()
```

--------------------------------

### run tool cloud

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Runs the agent in cloud environment for device control, starting an MCP server and connecting to MQTT.

```APIDOC
## `run tool cloud`

Run the agent in cloud environment (cloud device control).

### Usage

```bash
dp-agent run tool cloud
```

### Behavior

- Executes `main.py cloud` subprocess
- Starts MCP server with cloud transport
- Connects to MQTT for device control
- Requires configured `.env` file

### Environment

Cloud device twin with MQTT/MCP integration

### Example

```bash
dp-agent run tool cloud
# Starts cloud MCP server listening for tool calls
```
```

--------------------------------

### .env File Format

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Example of a .env file format, including comments, configuration variables, and handling of empty values. Variable interpolation is not supported.

```bash
# Comments start with #

# MQTT Configuration
MQTT_INSTANCE_ID=value
MQTT_ENDPOINT=value

# Variable interpolation not supported
# All values are strings

# Empty values
OPTIONAL_VAR=
```

--------------------------------

### action Decorator Example

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Demonstrates how to use the @action decorator to register a method as a device action and an MCP tool, including custom parameters and return types.

```python
from dp.agent.device.device import Device, action, BaseParams, SuccessResult

class TakePictureParams(BaseParams):
    horizontal_width: str

class PictureData(TypedDict):
    image_id: str

class PictureResult(SuccessResult):
    data: PictureData

class MyDevice(Device):
    device_name = "my_microscope"
    
    @action("take_picture")
    def take_picture(self, params: TakePictureParams) -> PictureResult:
        """
        Take a picture with the specified horizontal width.
        
        This action is automatically registered as an MCP tool.
        """
        hw = params.get("horizontal_width", "default")
        return PictureResult(
            message=f"Picture taken with width {hw}",
            data={"image_id": "image_123"}
        )
```

--------------------------------

### Run Agent with Custom Python Path

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Execute the agent using a specific Python environment. Ensure the SDK is installed in that environment first.

```bash
# Ensure SDK is installed
pip install bohr-agent-sdk

# Run with specific Python environment
/path/to/python -m dp.agent.cli.cli run tool cloud
```

--------------------------------

### Custom Job Execution Workflow

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/executor-storage-api.md

A complete example demonstrating the workflow of submitting a custom calculation job, monitoring its status, and retrieving results using LocalExecutor and LocalStorage.

```python
from pathlib import Path
from dp.agent.server.executor import LocalExecutor
from dp.agent.server.storage import LocalStorage
import time

def my_calculation(input_file: Path, num_steps: int) -> str:
    """Simulate calculation."""
    print(f"Processing {input_file} for {num_steps} steps")
    return "result_data"

# Set up executor and storage
executor = LocalExecutor(env={"COMPUTE": "yes"})
storage = LocalStorage()

# Submit job
result = executor.submit(my_calculation, {
    "input_file": Path("input.txt"),
    "num_steps": 100
})
job_id = result["job_id"]
print(f"Job ID: {job_id}")

# Query status
while True:
    status = executor.query_status(job_id)
    print(f"Status: {status}")
    if status != "Running":
        break
    time.sleep(1)

# Get results
if status == "Succeeded":
    results = executor.get_results(job_id)
    print(f"Results: {results}")
else:
    print("Job failed")
```

--------------------------------

### Run agent in local device control mode

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Executes the agent in a lab environment, starting a local device twin that listens for MQTT control messages and executes actions locally. Requires a configured .env file.

```bash
dp-agent run tool device
# Starts device twin listening for control commands
```

--------------------------------

### Run agent in cloud device control mode

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Executes the agent in a cloud environment, starting an MCP server with cloud transport and connecting via MQTT for device control. Requires a configured .env file.

```bash
dp-agent run tool cloud
# Starts cloud MCP server listening for tool calls
```

--------------------------------

### Run Deep Potential Training with Artifact Handling

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Call a remote Deep Potential model training tool and handle artifact paths. This example demonstrates verifying tool availability, submitting training arguments, and parsing JSON results to extract output model and log paths.

```python
import asyncio
import json
from pathlib import Path
from dp.agent.client import MCPClient

async def run_deep_potential_training():
    """
    Call a Deep Potential model training tool running on remote MCP server.
    """
    client = MCPClient("https://compute.example.com/mcp")
    session = await client.get_session()
    
    # List tools to verify availability
    tools = await session.list_tools()
    tool_names = [t.name for t in tools.tools]
    
    if "train_deep_potential" not in tool_names:
        print(f"Available tools: {tool_names}")
        return
    
    # Call training tool with artifact paths
    result = await session.call_tool(
        name="train_deep_potential",
        arguments={
            "training_data": "bohrium:///project/training.npy",
            "model_type": "dpa3",
            "rcut": 9.0,
            "numb_steps": 1000000,
            "start_lr": 0.001
        }
    )
    
    # Parse result
    result_text = result.content[0].text
    print(f"Training result: {result_text}")
    
    # Extract artifact paths from result
    try:
        result_data = json.loads(result_text)
        print(f"Output model: {result_data.get('model')}")
        print(f"Output log: {result_data.get('log')}")
    except:
        pass
    
    await client.exit_stack.aclose()

if __name__ == "__main__":
    asyncio.run(run_deep_potential_training())
```

--------------------------------

### Configure OSS Storage (Aliyun)

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Sets up CalculationMCPServer to use Aliyun OSS for storage. Requires specifying bucket name and access credentials.

```python
mcp = CalculationMCPServer(
    "MyServer",
    storage={
        "type": "oss",
        "bucket": "my-bucket",
        "access_key_id": "...",
        "access_key_secret": "..."
    }
)
```

--------------------------------

### Example Parameter Type Definition

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/types.md

An example of extending the BaseParams to define specific parameters for an action, such as 'horizontal_width' and 'vertical_height'.

```python
class TakePictureParams(BaseParams):
    horizontal_width: str
    vertical_height: str
```

--------------------------------

### Scaffold and Run Device Project Workflow

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/index.md

Steps to create a new device project, generate scaffolding, configure credentials, implement device logic, and run it in lab or cloud mode.

```bash
# 1. Create directory
mkdir my_device && cd my_device

# 2. Generate scaffolding
dp-agent fetch scaffolding --type device

# 3. Get configuration template
dp-agent fetch config

# 4. Edit .env with MQTT credentials
nano .env

# 5. Implement your device
nano device/tescan_device.py

# 6. Run as device twin (lab mode)
dp-agent run tool device

# 7. OR run as cloud service
dp-agent run tool cloud
```

--------------------------------

### Cloud Services Exports

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/module-index.md

Imports components for cloud services, including the global FastMCP instance, a function to get the MCP instance, the MQTT client class, and a function to get the MQTT client.

```python
from dp.agent.cloud import (
    mcp,                       # Global FastMCP instance
    get_mcp_instance,         # Get global MCP instance
    MQTTCloud,                # MQTT client class
    get_mqtt_cloud_instance   # Get global MQTT client
)
```

--------------------------------

### Executor Get Results

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/executor-storage-api.md

Retrieves the results of a completed job using its job ID.

```APIDOC
## Executor Get Results

### Description
Retrieves the results of a completed job.

### Method
```python
executor.get_results(job_id: str)
```

### Parameters
#### Path Parameters
- **job_id** (str) - Required - The ID of the job whose results are to be retrieved.

### Returns
- **results** (any) - The results of the job.
```

--------------------------------

### Get Session

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Retrieves an existing MCP session or creates a new one if none is available.

```APIDOC
## `get_session()`

### Description
Gets the current MCP session or creates a new one if it does not exist.

### Method
GET

### Endpoint
`/session`

### Parameters
None

### Response
#### Success Response (200)
* **session** (object) - The MCP session object.

### Response Example
```json
{
  "session": "<mcp_session_instance>"
}
```
```

--------------------------------

### Get Logger Instance

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/module-index.md

Initializes and retrieves a logger instance for use within the application. Configurable with different logging levels.

```python
from dp.agent.server.utils import get_logger

logger = get_logger(__name__, level="INFO")
logger.info("Message")
```

--------------------------------

### Fetch Configuration File

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Use the dp-agent command to fetch the necessary configuration file for your project.

```bash
dp-agent fetch config
```

--------------------------------

### run tool calculation

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Runs the agent in calculation mode, starting an MCP server and registering calculation tools for job submission.

```APIDOC
## `run tool calculation`

Run the agent in calculation mode.

### Usage

```bash
dp-agent run tool calculation
```

### Behavior

- Executes `main.py calculation` subprocess
- Starts calculation MCP server
- Registers calculation tools
- Submits jobs to executor

### Environment

Calculation MCP server with executor backend

### Example

```bash
dp-agent run tool calculation
# Starts calculation MCP server
```
```

--------------------------------

### Get MCP Instance Function

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Retrieve the global MCP instance using the get_mcp_instance() function. This function returns an instance of FastMCP.

```python
from dp.agent.cloud import get_mcp_instance

mcp = get_mcp_instance()
```

--------------------------------

### fetch config

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Downloads and sets up the .env configuration file by prompting for necessary credentials and environment variables.

```APIDOC
## `fetch config`

Download and set up .env configuration file.

### Usage

```bash
dp-agent fetch config
```

### Parameters

None

### Behavior

- Downloads remote .env template
- Creates local `.env` file
- Prompts for required values:
  - MQTT_INSTANCE_ID
  - MQTT_ENDPOINT
  - MQTT_DEVICE_ID
  - MQTT_GROUP_ID
  - MQTT_AK
  - MQTT_SK
  - MQTT_BROKER (optional)
  - MQTT_PORT (optional)
  - TESCAN_API_BASE (optional)

### Environment Variables

Required after running this command:

```bash
# MQTT Configuration
MQTT_INSTANCE_ID=your_instance_id
MQTT_ENDPOINT=your_endpoint
MQTT_DEVICE_ID=your_device_id
MQTT_GROUP_ID=your_group_id
MQTT_AK=your_access_key
MQTT_SK=your_secret_key

# Computing Resources
BOHRIUM_USERNAME=your_username
BOHRIUM_PASSWORD=your_password
```

**Note:** This command only works in internal network environments and performs dynamic variable replacement.

### Example

```bash
dp-agent fetch config
# Edit the generated .env file with your credentials
nano .env
```
```

--------------------------------

### Display CLI Help

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/README.md

Run this command to view all available CLI commands and options. Refer to cli-api.md for detailed explanations.

```bash
dp-agent --help
```

--------------------------------

### Lab Mode Agent Development

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Example of developing a custom device class for lab mode, including action registration and parameter handling.

```python
from typing import Dict, TypedDict
from dp.agent.device.device import Device, action, BaseParams, SuccessResult

class TakePictureParams(BaseParams):
    """Picture taking parameters"""
    horizontal_width: str  # Image horizontal width

class PictureData(TypedDict):
    """Picture data structure"""
    image_id: str

class PictureResult(SuccessResult):
    """Picture taking result"""
    data: PictureData

class MyDevice(Device):
    """Custom device class"""
    device_name = "my_device"

    @action("take_picture")
    def take_picture(self, params: TakePictureParams) -> PictureResult:
        """
        Execute picture taking action

        Through the @action decorator, automatically register this method as an MCP standard service
        """
        hw = params.get("horizontal_width", "default")
        # Execute actual device control logic
        return PictureResult(
            message=f"Picture taken with {self.device_name}",
            data={"image_id": "image_123"}
        )
```

--------------------------------

### Fetch Device Control Project Template

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/README.md

Use the dp-agent command to fetch a project template for device control.

```bash
dp-agent fetch scaffolding --type=device
```

--------------------------------

### Cloud Device Control

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/README.md

Control devices connected to the cloud via MQTT. This example sends a control command to a device and waits for a status update.

```python
from dp.agent.cloud import get_mqtt_cloud_instance

mqtt = get_mqtt_cloud_instance()

request_id = mqtt.send_device_control(
    device_name="device_001",
    device_action="take_picture",
    device_params={"width": "1024"}
)

response = mqtt.wait_for_status_update(request_id, timeout=30)
if response:
    print(f"Result: {response['result']}")

mqtt.stop()
```

--------------------------------

### Fetch Device Control Agent Scaffolding

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/README.md

Use this command to fetch the basic scaffolding for creating a device control agent. Refer to device-api.md and cloud-mqtt-api.md for more details.

```bash
dp-agent fetch scaffolding --type device
```

--------------------------------

### DeviceTwin.__init__

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Initializes the device twin, setting up MQTT client and message handlers for device control and status communication.

```APIDOC
## DeviceTwin.__init__

### Description
Initialize the device twin. Loads MQTT configuration, initializes the MQTT client, and sets up message handlers for incoming control commands.

### Method
__init__

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **device** (Device or Callable) - Required - Device instance or action dispatcher function
- **env_path** (str) - Optional - Path to .env file; defaults to current directory or SDK defaults

### Request Example
```python
from dp.agent.device import DeviceTwin

device = MyDevice()
twin = DeviceTwin(device, env_path=".env")
twin.run()
```

### Response
#### Success Response (None)
None

#### Response Example
None
```

--------------------------------

### Session Management - Auto-Reconnection

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Provides an example of how to implement robust client behavior that automatically handles connection errors and attempts to reconnect, ensuring continuous operation.

```APIDOC
## Session Management - Auto-Reconnection

### Description
Provides an example of how to implement robust client behavior that automatically handles connection errors and attempts to reconnect, ensuring continuous operation.

### Example:
```python
async def robust_client():
    client = MCPClient("server.py")
    
    while True:
        try:
            session = await client.get_session()
            result = await session.call_tool(...) # Replace with actual tool call
            print(result)
        except Exception as e:
            print(f"Connection error: {e}")
            # Client automatically cleans up and reconnects on next call
            await asyncio.sleep(5)
```
```

--------------------------------

### Device Project main.py Structure

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Provides the structure for `main.py` in a device project, showing how to define a device class with actions and run it in either cloud or device mode.

```python
from dp.agent.device.device import Device, action
from dp.agent.cloud import mcp, register_mcp_tools
from dp.agent.device import DeviceTwin

class MyDevice(Device):
    device_name = "my_device"
    
    @action("do_something")
    def do_something(self, params):
        # Implementation
        pass

if __name__ == "__main__":
    import sys
    mode = sys.argv[1] if len(sys.argv) > 1 else "device"
    
    if mode == "cloud":
        device = MyDevice()
        register_mcp_tools(mcp, device)
        mcp.run(transport="sse")
    else:  # device/lab mode
        device = MyDevice()
        twin = DeviceTwin(device)
        twin.run()
```

--------------------------------

### MQTTCloud Constructor

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Initializes MQTTCloud with optional parameters. Defaults are loaded from environment variables.

```python
MQTTCloud(
    instance_id: Optional[str] = None,
    endpoint: Optional[str] = None,
    device_id: Optional[str] = None,
    group_id: Optional[str] = None,
    port: Optional[str] = None,
    access_key: Optional[str] = None,
    secret_key: Optional[str] = None,
    device_control_topic: Optional[str] = None,
    device_status_topic: Optional[str] = None,
    redis_config: Optional[Dict[str, str]] = None
)
```

--------------------------------

### DeviceTwin Constructor

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Initializes DeviceTwin with a device instance or dispatcher function and an optional environment file path.

```python
DeviceTwin(
    device: Union[Device, Callable],  # Device instance or dispatcher function
    env_path: str = None              # Path to .env file (optional)
)
```

--------------------------------

### Device __init__ Method

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/device-api.md

Initializes the device and automatically registers all action methods decorated with @action during instantiation.

```python
def __init__(self) -> None
```

--------------------------------

### run tool device

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Runs the agent in device control mode for lab environments, starting a local device twin and listening for MQTT control messages.

```APIDOC
## `run tool device`

Run the agent in device control mode (lab environment).

### Usage

```bash
dp-agent run tool device
```

### Behavior

- Executes `main.py device` subprocess
- Starts local device twin
- Listens for MQTT control messages
- Executes device actions locally

### Environment

Local device twin with MQTT communication

### Example

```bash
dp-agent run tool device
# Starts device twin listening for control commands
```
```

--------------------------------

### CalculationMCPServer Constructor

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Initialize a CalculationMCPServer. Accepts server name, executor, storage, and a preprocessor function.

```python
CalculationMCPServer(
    name: str,                          # Server name
    executor: Optional[dict] = None,    # Executor configuration
    storage: Optional[dict] = None,     # Storage configuration
    preprocessor: Optional[Callable] = None  # Config preprocessor function
)
```

--------------------------------

### Initialize Storage from Configuration

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/executor-storage-api.md

Creates a storage instance from a configuration dictionary. The configuration must include a 'type' field.

```python
from dp.agent.server.calculation_mcp_server import init_storage

config = {"type": "bohrium"}
storage_type, storage = init_storage(config)
```

--------------------------------

### CalculationMCPServer Initialization

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Initializes the CalculationMCPServer with a name and optional configurations for executor, storage, and a preprocessor function.

```APIDOC
## CalculationMCPServer Initialization

### Description
Initializes the CalculationMCP Server.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Constructor Signature
```python
CalculationMCPServer(name: str, executor: Optional[dict] = None, storage: Optional[dict] = None, preprocessor: Optional[Callable] = None)
```

### Parameters
- **name** (str) - Required - Name of the MCP server
- **executor** (dict) - Optional - Executor configuration dict with "type" field
- **storage** (dict) - Optional - Storage configuration dict with "type" field
- **preprocessor** (Callable) - Optional - Function to preprocess executor/storage configs
```

--------------------------------

### Initialize MQTTCloud Client

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Instantiate the MQTTCloud client, which is used for sending device control messages and receiving status updates. Configuration can be provided via parameters or environment variables.

```python
from dp.agent.cloud import MQTTCloud

mqtt_client = MQTTCloud(
    instance_id="mqtt-instance-123",
    endpoint="mqtt.example.com",
    device_id="device_001",
    group_id="group_001",
    access_key="ak_123",
    secret_key="sk_456"
)
```

--------------------------------

### Run Agent in Debug Mode

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cli-api.md

Executes the agent in debug mode, enabling verbose logging and starting with debugging endpoints. This is useful for troubleshooting and provides detailed error messages.

```bash
dp-agent run debug
# Starts with verbose output
```

--------------------------------

### Initialize MQTT Cloud Client

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/configuration.md

Instantiate the MQTTCloud client to establish a connection. Checks if the client was successfully initialized.

```python
from dp.agent.cloud import MQTTCloud

mqtt = MQTTCloud()
if mqtt.mqtt_client:
    print("MQTT client initialized")
else:
    print("MQTT configuration error")
```

--------------------------------

### Get MCPClient Session

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/mcp-client-api.md

Obtain a ClientSession to interact with the MCP server. This method handles connection establishment, reuse, and cleanup, ensuring a valid session for tool discovery and execution.

```python
import asyncio
from dp.agent.client import MCPClient

async def use_client():
    client = MCPClient("server.py")
    session = await client.get_session()
    
    # Use session to list tools
    tools = await session.list_tools()
    print(tools)

asyncio.run(use_client())
```

--------------------------------

### Module Exports

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/cloud-mqtt-api.md

Imports available from the dp.agent.cloud module, including the global FastMCP instance, a function to retrieve it, the MQTTCloud client class, and a function to get the global MQTT instance.

```python
from dp.agent.cloud import (
    mcp,
    get_mcp_instance,
    MQTTCloud,
    get_mqtt_cloud_instance
)
```

--------------------------------

### Create a Calculation Tool for MCP Server

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/module-index.md

This pattern demonstrates how to create a calculation tool that can be registered with an MCP server. The tool takes a Path as input and returns a Path. Ensure CalculationMCPServer and Path are imported.

```python
from dp.agent.server import CalculationMCPServer
from pathlib import Path

mcp = CalculationMCPServer("MyServer")

@mcp.tool()
def my_tool(input_file: Path) -> Path:
    return Path("output.txt")

mcp.run(transport="sse")
```

--------------------------------

### Configure Local Storage

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/calculation-mcp-api.md

Sets up CalculationMCPServer to use local filesystem for artifact storage. This is the default configuration.

```python
mcp = CalculationMCPServer(
    "MyServer",
    storage={"type": "local"}
)
```

--------------------------------

### init_storage

Source: https://github.com/dptech-corp/bohr-agent-sdk/blob/master/_autodocs/executor-storage-api.md

Creates a storage instance from a configuration dictionary. It returns the storage type and the storage instance.

```APIDOC
## init_storage

### Description
Create storage from configuration dictionary.

### Method
```python
def init_storage(storage_config: Optional[dict] = None) -> Tuple[str, BaseStorage]
```

### Parameters
#### Path Parameters
- **storage_config** (dict) - Optional - Config with "type" field specifying the storage type.

### Returns
- **Tuple[str, BaseStorage]** - A tuple containing the storage type string and the storage instance.
```