Skip to main content
This is part 5 of the Your First Plugin tutorial. Make sure you’ve completed 4. Use Integration in Node first.
Plugins can read and create files. Since plugins run in isolated processes, file I/O goes through the platform’s file helper — the SDK handles all the bridging transparently.

Reading files

Add a node that reads a file input:
from noxus_sdk.files import File


class ParseWeatherFileConfig(NodeConfiguration):
    pass


class ParseWeatherFileNode(BaseNode[ParseWeatherFileConfig]):
    node_name = "parse_weather_file"
    title = "Parse Weather File"
    description = "Reads a CSV file of cities and returns weather data"
    category = NodeCategory.DATA
    color = "#4A90E2"

    inputs = [
        Connector(
            name="file",
            label="Cities File",
            definition=TypeDefinition(data_type=DataType.File),
        )
    ]

    outputs = [
        Connector(
            name="cities",
            label="Cities",
            definition=TypeDefinition(data_type=DataType.str, is_list=True),
        )
    ]

    async def call(self, ctx: RemoteExecutionContext, file: File) -> dict:
        # Read file content through the file helper
        content = await file.get_content(ctx)
        text = content.decode("utf-8")

        # Parse CSV lines
        cities = [line.strip() for line in text.splitlines() if line.strip()]

        return {"cities": cities}

How file reading works

When a File type input arrives, it contains metadata (name, URI, content type) but not the actual bytes. Calling file.get_content(ctx) triggers: You can also access file metadata without downloading:
file.name          # "cities.csv"
file.content_type  # "text/csv"
file.uri           # "spot://..."
file.id            # UUID string

Creating files

class GenerateReportNode(BaseNode[NodeConfiguration]):
    node_name = "generate_weather_report"
    title = "Generate Weather Report"
    description = "Creates a text file with weather data"
    category = NodeCategory.DATA
    color = "#4A90E2"

    inputs = [
        Connector(
            name="report_text",
            label="Report Text",
            definition=TypeDefinition(data_type=DataType.str),
        )
    ]

    outputs = [
        Connector(
            name="report_file",
            label="Report File",
            definition=TypeDefinition(data_type=DataType.File),
        )
    ]

    async def call(self, ctx: RemoteExecutionContext, report_text: str) -> dict:
        # Create a file through the file helper
        report_file = await File.from_bytes(
            ctx,
            data=report_text.encode("utf-8"),
            name="weather_report.txt",
            content_type="text/plain",
        )

        return {"report_file": report_file}
File.from_bytes() uploads the content to the platform’s storage and returns a File object that downstream nodes can use.

Quick reference

OperationCode
Read file contentcontent = await file.get_content(ctx)
Create from bytesawait File.from_bytes(ctx, data=b"...", name="file.txt")
Create from textawait File.from_bytes(ctx, data=text.encode("utf-8"), name="out.txt", content_type="text/plain")
Access file namefile.name
Access content typefile.content_type
Access file URIfile.uri
File type inputTypeDefinition(data_type=DataType.File)
File list inputTypeDefinition(data_type=DataType.File, is_list=True)
Image type inputTypeDefinition(data_type=DataType.Image)

Handling multiple files

Use list types to receive or produce multiple files:
inputs = [
    Connector(
        name="files",
        label="Input Files",
        definition=TypeDefinition(data_type=DataType.File, is_list=True),
    )
]

async def call(self, ctx: RemoteExecutionContext, files: list[File]) -> dict:
    all_text = []
    for f in files:
        content = await f.get_content(ctx)
        all_text.append(content.decode("utf-8"))

    combined = "\n---\n".join(all_text)
    return {"combined_text": combined}

Next: Advanced Techniques →

Config UI controls, dynamic config, list handling, error handling, and deployment.