| name | plugin-function |
| description | Creates in-process NemoFunction surfaces for NeMo Platform plugins. Use when adding a function, declaring spec_schema, mounting function routes with add_function_routes, understanding the two CLI verbs (run / submit), or streaming NDJSON frames. Trigger keywords - function, NemoFunction, spec_schema, add_function_routes, nemo_platform_plugin.functions, two verbs, run, submit, streaming, NDJSON, FunctionContext. |
Plugin Functions (NemoFunction)
A NemoFunction is the third primitive on a plugin, alongside NemoResource and NemoJob. It's an in-process request handler — no scheduler, no backend dispatch — that the platform exposes as both a CLI subcommand and an HTTP route automatically.
nemo <plugin> <fn> run [--spec '{...}' | --spec-file FILE] [--workspace W] [<spec-flag>...]
nemo <plugin> <fn> submit [--spec '{...}' | --spec-file FILE] \
[--base-url URL | --cluster URL] \
[--workspace W] [--request-id ID] [<spec-flag>...]
run is local (in-process); submit POSTs to the plugin service's auto-derived route. Two verbs only — no explain. A function's only schema is spec_schema, and --help is the introspection surface.
CLI introspection — auto-generated per-field flags
Every scalar leaf in spec_schema becomes a Typer flag automatically. Nested submodels recurse with dotted paths (--target.url, --target.timeout-seconds). For a function with spec_schema = GreetSpec(name: str):
$ nemo my-plugin greet run --help
...
Function Spec:
--name <NAME> Name to greet.
Spec Source:
--spec <SPEC> Spec as a JSON string. [default: {}]
--spec-file <SPEC_FILE> Path to a YAML or JSON spec file (used as base; per-flag values override).
--workspace <WORKSPACE> Workspace identity passed to the function as ctx.workspace. [default: default]
...
Function Spec flags are generated from the GreetSpec Pydantic schema. Precedence: --spec-file (base) → --spec JSON (overlay) → per-flag values (top).
Precedence at runtime is --spec-file (base) → --spec JSON (overlay) → per-field flag (top), then validation against spec_schema happens once on the merged value. Lists, dicts, discriminated unions, and other types the walker can't render fall through to --spec / --spec-file. The plumbing lives in nemo_platform_plugin._spec_flags and is shared with the jobs CLI; if a leaf is missing from --help, check that the field type is str / int / float / bool (or Optional of one of those) and that its name doesn't collide with a static verb flag.
Class Signature
from typing import ClassVar
from nemo_platform_plugin.function import NemoFunction
from pydantic import BaseModel
class GreetSpec(BaseModel):
name: str
class GreetResponse(BaseModel):
message: str
class GreetFunction(NemoFunction[GreetSpec]):
name: ClassVar[str] = "greet"
description: ClassVar[str] = "Say hello to a name."
spec_schema: ClassVar[type[BaseModel]] = GreetSpec
async def run(self, spec: GreetSpec) -> GreetResponse:
return GreetResponse(message=f"Hello, {spec.name}!")
Required: name, spec_schema, async def run(). Optional: description, endpoint.
Method colours — run is always async def
NemoFunction is single-coloured. run is the request handler, so it must be async def to avoid blocking the FastAPI event loop while a stream is open. The base class enforces this at class-definition time:
class BadFunction(NemoFunction[GreetSpec]):
name = "bad"
spec_schema = GreetSpec
def run(self, spec):
return {}
Both shapes are accepted:
| Shape | What it returns | Wire content type |
|---|
async def run(self, spec) | A value (typically a Pydantic model) | application/json |
async def run(self, spec): yield ... | An AsyncIterator of frames | application/x-ndjson |
Pure sync work goes through await asyncio.to_thread(...):
async def run(self, spec: GreetSpec) -> GreetResponse:
rendered = await asyncio.to_thread(_blocking_template_render, spec.name)
return GreetResponse(message=rendered)
There is no streaming: ClassVar[bool] toggle — the route adapter detects the iterator return at request time.
Streaming
Yield Pydantic models from an async def body:
from collections.abc import AsyncIterator
from nemo_platform_plugin.functions.frames import Done, Heartbeat
from pydantic import BaseModel
class TickSpec(BaseModel):
upto: int
class Tick(BaseModel):
kind: str = "tick"
n: int
class CountFunction(NemoFunction[TickSpec]):
name = "count"
spec_schema = TickSpec
async def run(self, spec: TickSpec) -> AsyncIterator[BaseModel]:
for i in range(spec.upto):
yield Tick(n=i)
yield Done()
Each yielded BaseModel becomes one NDJSON line (model_dump_json() + "\n") on the wire. The route adapter wraps the iterator with a heartbeat injector that emits a Heartbeat() frame every 5 seconds of idle time so proxies don't time the connection out. Plugin authors don't need to emit heartbeats themselves — the framework adds them.
Convention frames live in nemo_platform_plugin.functions.frames:
| Class | kind | When to emit |
|---|
Heartbeat | "heartbeat" | Framework injects on idle; plugins rarely emit explicitly |
Done | "done" | Last frame of a successful stream |
Error | "error" | Last frame of a failed stream |
The discriminator is the kind field. Use it in your own frames so consumers can branch with jq -r 'select(.kind == "tick") | .n' etc.
Entry-Point Key Format
Dot-separated: "<plugin-name>.<function-name>".
[project.entry-points."nemo.functions"]
"my-plugin.greet" = "nemo_my_plugin.functions.greet:GreetFunction"
NemoFunction.name is the suffix after the dot. For "my-plugin.greet" the class sets name = "greet".
Mounting function routes — add_function_routes(fn_cls)
This is the one-liner that mounts the function as a POST route on your plugin service:
from fastapi import FastAPI
from nemo_platform_plugin.authz import AuthzScope
from nemo_platform_plugin.functions.routes import add_function_routes
from .functions.greet import GreetFunction
app = FastAPI()
router = add_function_routes(
GreetFunction,
authz=AuthzScope("my-plugin"),
permission_description="Invoke the greet function",
)
app.include_router(
router,
prefix="/apis/my-plugin/v2/workspaces/{workspace}",
)
authz= is required — pass your plugin's AuthzScope or the route is left unruled, and under the default on_invalid_plugin=hard_fail the auth service refuses to build the OPA bundle (the platform 502s) rather than silently fencing the route. See the plugin-authz skill for the full path-rule model.
add_function_routes:
- Validates the body against
spec_schema (422 on bad input).
- Builds a
FunctionContext from the workspace path parameter and the optional X-Request-ID header.
- Resolves keyword-only DI parameters on
run by name — ctx, async_sdk (sync sdk injection lands in a follow-up).
- Awaits
run(spec, **resolved) and serialises the result; returns application/x-ndjson with heartbeat injection if the return is an async iterator.
- Stamps the route's authz from
authz= — a PRINCIPAL @path_rule carrying a write-action invoke permission (<namespace>.<function-name>, e.g. my-plugin.greet) plus @AuthzScope.write. permission_description overrides that permission's description (it defaults to the function's description); passing it without authz= is a ValueError.
Optional kwarg heartbeat_interval_seconds overrides the 5-second default — useful in tests.
Endpoint override
The default route path is /{name} mounted under the workspace prefix — i.e. /apis/<plugin>/v2/workspaces/{workspace}/<name>. Override the trailing segment per-class via endpoint:
class GreetFunction(NemoFunction[GreetSpec]):
name = "greet"
spec_schema = GreetSpec
endpoint: ClassVar[str | None] = "/{name}/v1"
Only {name} is substituted; the workspace placeholder stays as a live FastAPI route parameter. The override doesn't let a function relocate itself outside its plugin's URL namespace — that's intentional.
DI by signature
run accepts framework-managed dependencies as keyword-only parameters. The route adapter and the local CLI both resolve them by parameter name:
from nemo_platform_plugin.function import NemoFunction
from nemo_platform_plugin.function_context import FunctionContext
class WhoamiFunction(NemoFunction[GreetSpec]):
name = "whoami"
spec_schema = GreetSpec
async def run(
self,
spec: GreetSpec,
*,
ctx: FunctionContext,
async_sdk: object | None = None,
) -> dict:
return {
"workspace": ctx.workspace,
"request_id": ctx.request_id,
"sdk_present": async_sdk is not None,
}
| Parameter | Source | Notes |
|---|
ctx: FunctionContext | route adapter / CLI | workspace from URL or --workspace, request_id from X-Request-ID |
async_sdk | route adapter only (today) | Plugin services need app.dependency_overrides[get_sdk_client] to inject a real handle |
sdk | declared but currently bound to None | Sync placeholder lands with the SDK-builder follow-up |
Parameters not declared on run aren't injected — there's no implicit context. Keep run signatures narrow.
Programmatic invocation
import asyncio
from nemo_platform_plugin.discovery import discover_functions
from nemo_platform_plugin.function_context import FunctionContext
fn_cls = discover_functions()["my-plugin.greet"]
spec = fn_cls.spec_schema.model_validate({"name": "Ada"})
result = asyncio.run(fn_cls().run(spec, ctx=FunctionContext(workspace="default")))
discover_functions() returns the typed mapping. There is no FunctionScheduler — the function is the runtime; instantiate and await.
Testing
Unit-test functions directly — no FastAPI, no platform required:
import pytest
@pytest.mark.asyncio
async def test_greet_returns_message():
fn = GreetFunction()
result = await fn.run(GreetSpec(name="Ada"))
assert result.message == "Hello, Ada!"
Route-test with TestClient when you want to exercise validation, DI, and streaming:
from fastapi import FastAPI
from fastapi.testclient import TestClient
from nemo_platform_plugin.authz import AuthzScope
from nemo_platform_plugin.functions.routes import add_function_routes
def test_greet_route_returns_json():
app = FastAPI()
app.include_router(
add_function_routes(GreetFunction, authz=AuthzScope("my-plugin")),
prefix="/apis/my-plugin/v2/workspaces/{workspace}",
)
client = TestClient(app)
response = client.post(
"/apis/my-plugin/v2/workspaces/default/greet",
json={"name": "Ada"},
)
assert response.status_code == 200
assert response.json() == {"message": "Hello, Ada!"}
For streaming functions, set a low heartbeat interval so tests aren't flaky:
add_function_routes(CountFunction, heartbeat_interval_seconds=0, authz=AuthzScope("my-plugin"))
Gotchas
run must be async def: The class enforces this at definition time. Sync work uses await asyncio.to_thread(...).name is the suffix only: For entry-point key "my-plugin.greet", NemoFunction.name = "greet". The class declares the suffix; the dot-prefix lives in pyproject.toml.spec_schema is required: add_function_routes raises TypeError if it's missing. Declare it before mounting.authz= is required: pass your plugin's AuthzScope to add_function_routes, or the route is unruled and the OPA bundle build fails closed under hard_fail. See the plugin-authz skill.- Streaming is detected from the return type, not declared: An
async def with yield is an async generator — the route emits NDJSON. An async def with return is a coroutine — the route emits JSON. Don't add a class-level streaming flag.
endpoint is the trailing segment: Setting endpoint = "/{name}/v1" mounts under the workspace prefix; you can't relocate the function to a different plugin's URL namespace.- Reserved CLI flag:
submit --workspace controls the URL path segment — don't put a workspace field in your spec model with conflicting semantics.
For the architectural framing alongside NemoResource and NemoJob, see the shared resources/jobs/functions design note.
