Add typed functions APImain

Introduce an in-memory API for registering typed Python functions and invoking them with JSON. Generate Draft 2020-12 schemas from annotations and document the request formats and trust boundary.

6 files changed, 564 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..e552b03
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+.ruff_cache/
+*.py[cod]
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..d08b0bd
--- /dev/null
+++ b/README.md
@@ -0,0 +1,127 @@
+# Hot functions API
+
+A small HTTP service that registers typed Python functions and calls them with
+JSON. It derives input and output schemas from type annotations; there is no
+separate schema registration step or UI.
+
+## Run locally
+
+Python 3.10 or newer is needed for `int | str` annotations in the examples.
+The `curl` examples use Bash's `$'...'` quoting for multiline source.
+
+```sh
+python -m venv .venv
+. .venv/bin/activate
+pip install -r requirements.txt
+python server.py
+```
+
+The service listens on port 8000. Functions live in memory and disappear when
+the server stops.
+
+## Register and call a function
+
+Send a complete, synchronous function definition. The function name must match
+the URL, and every parameter and the return value must have a type annotation.
+
+```sh
+curl -X PATCH http://localhost:8000/double \
+  --data-binary $'def double(x: int) -> int:\n    return x * 2'
+```
+
+With one parameter, POST either a bare JSON value or a named argument:
+
+```sh
+curl -X POST http://localhost:8000/double \
+  -H 'Content-Type: application/json' -d '4'
+# 8
+
+curl -X POST http://localhost:8000/double \
+  -H 'Content-Type: application/json' -d '{"x":4}'
+# 8
+```
+
+The response is the function's JSON result, without a `result` wrapper. Input
+and output are validated against the annotations in Pydantic's strict mode.
+
+### Multiple arguments and union types
+
+```sh
+curl -X PATCH http://localhost:8000/label \
+  --data-binary $'def label(x: int | str, prefix: str = "item") -> str:\n    return f"{prefix}:{x}"'
+
+curl -X POST http://localhost:8000/label \
+  -H 'Content-Type: application/json' -d '{"x":4,"prefix":"order"}'
+# "order:4"
+
+curl -X POST http://localhost:8000/label \
+  -H 'Content-Type: application/json' -d '{"x":"abc"}'
+# "item:abc"
+```
+
+For multiple parameters, POST a JSON object with parameter names as keys.
+Defaults may be omitted. Unrecognized keys are rejected.
+
+A single parameter may accept any JSON value its annotation permits: an object,
+array, string, number, boolean, or `null`. For example, a function annotated
+`x: dict[str, int]` can receive `{"count":2}` directly or as
+`{"x":{"count":2}}`. If an object also looks like valid named arguments, the
+named-argument form takes precedence. Structured types can use Pydantic models
+defined in the uploaded source.
+
+## Discover functions and schemas
+
+```sh
+curl http://localhost:8000/double
+curl http://localhost:8000/
+```
+
+`GET /double` returns an object containing `name`, `source`, and `schema`.
+`GET /` returns `{"functions": [...]}` with the same objects for all registered
+functions. For `double`, the input schema describes both accepted POST forms:
+
+```json
+{
+  "name": "double",
+  "source": "def double(x: int) -> int:\n    return x * 2",
+  "schema": {
+    "input": {
+      "$schema": "https://json-schema.org/draft/2020-12/schema",
+      "anyOf": [
+        {
+          "type": "object",
+          "properties": {"x": {"title": "X", "type": "integer"}},
+          "required": ["x"],
+          "additionalProperties": false
+        },
+        {"type": "integer"}
+      ]
+    },
+    "output": {
+      "$schema": "https://json-schema.org/draft/2020-12/schema",
+      "type": "integer"
+    }
+  }
+}
+```
+
+The input and output schemas are JSON Schema Draft 2020-12 documents. Replacing
+a function with `PATCH` also regenerates its schemas.
+
+## API responses
+
+| Request | Result |
+| --- | --- |
+| `PATCH /name` | `200` on registration; `400` for an invalid definition |
+| `POST /name` | `200` with the JSON result; `400` for invalid input; `404` if undefined; `500` if execution or output validation fails |
+| `GET /name` | `200` with the function description; `404` if undefined |
+| `GET /` | `200` with all defined functions |
+
+Run the tests with `python -m unittest -v test_server`.
+
+## Trust boundary
+
+Uploaded definitions execute as Python in the server process, including at
+registration time. The server has no authentication, isolation, request-size
+limit, or execution timeout and binds to `0.0.0.0`. Only run it where every
+caller is fully trusted; it is not safe to expose to untrusted clients.
diff --git a/hot.code-workspace b/hot.code-workspace
new file mode 100644
index 0000000..362d7c2
--- /dev/null
+++ b/hot.code-workspace
@@ -0,0 +1,7 @@
+{
+	"folders": [
+		{
+			"path": "."
+		}
+	]
+}
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..5a55153
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,2 @@
+uvicorn[standard]
+pydantic>=2.10,<3
diff --git a/server.py b/server.py
new file mode 100644
index 0000000..3c32863
--- /dev/null
+++ b/server.py
@@ -0,0 +1,265 @@
+import inspect
+import json
+import re
+from typing import Union, get_type_hints
+
+from pydantic import ConfigDict, TypeAdapter, ValidationError, create_model
+
+
+CONTENT_TEXT = b"text/plain; charset=utf-8"
+JSON_SCHEMA_DRAFT = "https://json-schema.org/draft/2020-12/schema"
+
+
+def reject_non_json_constant(value):
+    raise ValueError(f"Invalid JSON value: {value}")
+
+
+def schema_document(schema):
+    return {"$schema": JSON_SCHEMA_DRAFT, **schema}
+
+
+def input_schema_document(adapter, model):
+    schema = adapter.json_schema()
+    definitions = schema.get("$defs", {})
+    for index, choice in enumerate(schema.get("anyOf", [])):
+        reference = choice.get("$ref", "")
+        key = reference.removeprefix("#/$defs/")
+        if (
+            reference.startswith("#/$defs/")
+            and definitions.get(key, {}).get("title") == model.__name__
+        ):
+            definition = definitions.pop(key)
+            definition.pop("title", None)
+            schema["anyOf"][index] = definition
+            if not definitions:
+                schema.pop("$defs")
+            break
+    if schema.get("title") == model.__name__:
+        schema.pop("title")
+    return schema_document(schema)
+
+
+class Route:
+    def __init__(self, name):
+        self.name = name
+        self.source = None
+        self.function = None
+        self.input_model = None
+        self.input_type = None
+        self.value_type = None
+        self.output_type = None
+        self.schema = None
+
+    def patch(self, body):
+        try:
+            source = body.decode("utf-8")
+            namespace = {"__name__": "__main__"}
+            exec(compile(source, "<function>", "exec"), namespace)
+            function = namespace.get(self.name)
+            if not inspect.isfunction(function) or inspect.iscoroutinefunction(
+                function
+            ):
+                raise ValueError(f"Expected a synchronous function named {self.name}")
+            parameters = list(inspect.signature(function).parameters.values())
+            if not parameters or any(
+                parameter.kind
+                not in (
+                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                    inspect.Parameter.KEYWORD_ONLY,
+                )
+                for parameter in parameters
+            ):
+                raise ValueError("Function must have one or more named parameters")
+            hints = get_type_hints(function)
+            if "return" not in hints or any(
+                parameter.name not in hints for parameter in parameters
+            ):
+                raise ValueError(
+                    "All parameters and the return value must be annotated"
+                )
+            fields = {
+                parameter.name: (
+                    hints[parameter.name],
+                    ...
+                    if parameter.default is inspect.Parameter.empty
+                    else parameter.default,
+                )
+                for parameter in parameters
+            }
+            input_model = create_model(
+                f"{self.name}Input",
+                __config__=ConfigDict(extra="forbid", validate_default=True),
+                **fields,
+            )
+            input_type = TypeAdapter(
+                Union[input_model, hints[parameters[0].name]]
+                if len(parameters) == 1
+                else input_model
+            )
+            value_type = (
+                TypeAdapter(hints[parameters[0].name]) if len(parameters) == 1 else None
+            )
+            output_type = TypeAdapter(hints["return"])
+            schema = {
+                "input": input_schema_document(input_type, input_model),
+                "output": schema_document(output_type.json_schema()),
+            }
+        except Exception as error:
+            return 400, str(error).encode(), CONTENT_TEXT
+
+        self.source = source
+        self.function = function
+        self.input_model = input_model
+        self.input_type = input_type
+        self.value_type = value_type
+        self.output_type = output_type
+        self.schema = schema
+        return 200, b"", CONTENT_TEXT
+
+    def get(self):
+        if self.source is None:
+            return 404, b"Function not found\n", CONTENT_TEXT
+        return 200, json.dumps(self.describe()).encode(), b"application/json"
+
+    def describe(self):
+        return {"name": self.name, "source": self.source, "schema": self.schema}
+
+    def post(self, body):
+        if (
+            self.function is None
+            or self.input_model is None
+            or self.input_type is None
+            or self.output_type is None
+        ):
+            return 404, b"Function not found\n", CONTENT_TEXT
+        try:
+            data = json.loads(body, parse_constant=reject_non_json_constant)
+            if self.value_type is None:
+                values = self.input_model.model_validate(data, strict=True)
+            elif isinstance(data, dict):
+                try:
+                    values = self.input_model.model_validate(data, strict=True)
+                except ValidationError:
+                    values = self.value_type.validate_python(data, strict=True)
+            else:
+                values = self.value_type.validate_python(data, strict=True)
+        except (ValueError, UnicodeError, ValidationError) as error:
+            return 400, str(error).encode(), CONTENT_TEXT
+        try:
+            if isinstance(values, self.input_model):
+                arguments = {
+                    name: getattr(values, name)
+                    for name in self.input_model.model_fields
+                }
+            else:
+                arguments = {next(iter(self.input_model.model_fields)): values}
+            result = self.function(**arguments)
+            result = self.output_type.validate_python(result, strict=True)
+            response = self.output_type.dump_json(result)
+        except Exception as error:
+            return 500, str(error).encode(), CONTENT_TEXT
+        return 200, response, b"application/json"
+
+
+class Routes:
+    def __init__(self):
+        self.routes = {}
+
+    def route(self, name):
+        if not re.fullmatch(r"[A-Za-z_]\w*", name):
+            return None
+        return self.routes.setdefault(name, Route(name))
+
+    def patch(self, name, body):
+        route = self.route(name)
+        if route is None:
+            return 400, b"Invalid function name\n", CONTENT_TEXT
+        return route.patch(body)
+
+    def get(self, name):
+        route = self.route(name)
+        if route is None:
+            return 400, b"Invalid function name\n", CONTENT_TEXT
+        return route.get()
+
+    def post(self, name, body):
+        route = self.route(name)
+        if route is None:
+            return 400, b"Invalid function name\n", CONTENT_TEXT
+        return route.post(body)
+
+
+class Server:
+    def __init__(self, routes=None):
+        self.routes = routes or Routes()
+
+    async def read_body(self, receive):
+        chunks = []
+        while True:
+            message = await receive()
+            if message["type"] == "http.disconnect":
+                break
+            chunks.append(message.get("body", b""))
+            if not message.get("more_body", False):
+                break
+        return b"".join(chunks)
+
+    def function_route(self, method, name, body):
+        if method == "PATCH":
+            return self.routes.patch(name, body)
+        if method == "GET":
+            return self.routes.get(name)
+        if method == "POST":
+            return self.routes.post(name, body)
+        return 404, b"Function not found\n", CONTENT_TEXT
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            return
+
+        path = scope["path"]
+        method = scope["method"]
+        body = await self.read_body(receive) if method in {"POST", "PATCH"} else b""
+
+        if method == "GET" and path == "/":
+            functions = [
+                route.describe()
+                for name, route in sorted(self.routes.routes.items())
+                if route.source is not None
+            ]
+            status, response_body, content_type = (
+                200,
+                json.dumps({"functions": functions}).encode(),
+                b"application/json",
+            )
+        elif path == "/":
+            status, response_body, content_type = 200, b"", CONTENT_TEXT
+        elif (
+            method in {"GET", "POST", "PATCH"} and path.count("/") == 1 and path != "/"
+        ):
+            status, response_body, content_type = self.function_route(
+                method, path[1:], body
+            )
+        else:
+            status, response_body, content_type = 404, b"Not found\n", CONTENT_TEXT
+
+        await send(
+            {
+                "type": "http.response.start",
+                "status": status,
+                "headers": [
+                    (b"content-length", str(len(response_body)).encode()),
+                    (b"content-type", content_type),
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": response_body})
+
+
+app = Server()
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run("server:app", host="0.0.0.0", port=8000)
diff --git a/test_server.py b/test_server.py
new file mode 100644
index 0000000..42a666f
--- /dev/null
+++ b/test_server.py
@@ -0,0 +1,159 @@
+import asyncio
+import json
+import unittest
+
+from server import JSON_SCHEMA_DRAFT, Server
+
+
+class FunctionTests(unittest.TestCase):
+    def setUp(self):
+        self.app = Server()
+
+    def request(self, method, path, body=b""):
+        async def call():
+            messages = []
+
+            async def receive():
+                return {"type": "http.request", "body": body, "more_body": False}
+
+            async def send(message):
+                messages.append(message)
+
+            await self.app(
+                {"type": "http", "method": method, "path": path}, receive, send
+            )
+            return messages[0]["status"], messages[1]["body"]
+
+        return asyncio.run(call())
+
+    def test_reflection_validates_calls_and_describes_function(self):
+        source = b"def double(x: int) -> int:\n    return x * 2\n"
+        self.assertEqual(self.request("PATCH", "/double", source)[0], 200)
+        status, body = self.request("GET", "/double")
+        self.assertEqual(status, 200)
+        entry = json.loads(self.request("GET", "/")[1])["functions"][0]
+        self.assertEqual(json.loads(body), entry)
+        self.assertEqual(entry["name"], "double")
+        self.assertEqual(entry["source"], source.decode())
+        input_schema = entry["schema"]["input"]
+        self.assertEqual(input_schema["$schema"], JSON_SCHEMA_DRAFT)
+        self.assertNotIn("$defs", input_schema)
+        self.assertEqual(input_schema["anyOf"][1], {"type": "integer"})
+        self.assertEqual(input_schema["anyOf"][0]["required"], ["x"])
+        self.assertEqual(
+            entry["schema"]["output"],
+            {"$schema": JSON_SCHEMA_DRAFT, "type": "integer"},
+        )
+        self.assertEqual(self.request("POST", "/double", b'{"x": 4}'), (200, b"8"))
+        self.assertEqual(self.request("POST", "/double", b"4"), (200, b"8"))
+        for invalid in (b'{"x": "4"}', b'{"x": 4.5}', b"{}", b"[]", b"NaN"):
+            self.assertEqual(self.request("POST", "/double", invalid)[0], 400)
+        self.assertEqual(self.request("GET", "/double/schema")[0], 404)
+        self.assertEqual(self.request("PUT", "/double/schema", b"{}")[0], 404)
+
+        self.assertEqual(
+            self.request(
+                "PATCH", "/double", b"def double(x: int) -> int:\n    return str(x)"
+            )[0],
+            200,
+        )
+        self.assertEqual(self.request("POST", "/double", b'{"x": 4}')[0], 500)
+
+    def test_invalid_update_keeps_previous_function(self):
+        original = b"def double(x: int) -> int:\n    return x * 2"
+        self.request("PATCH", "/double", original)
+        for invalid in (
+            b"return x * 2",
+            b"def other(x: int) -> int:\n    return x",
+            b"def double(x):\n    return x",
+            b"async def double(x: int) -> int:\n    return x",
+        ):
+            self.assertEqual(self.request("PATCH", "/double", invalid)[0], 400)
+        status, body = self.request("GET", "/double")
+        self.assertEqual(status, 200)
+        self.assertEqual(json.loads(body)["source"], original.decode())
+        self.assertEqual(self.request("POST", "/double", b"3"), (200, b"6"))
+
+    def test_multiple_parameters_union_and_default(self):
+        source = (
+            b'def label(x: int | str, prefix: str = "item") -> str:\n'
+            b'    return f"{prefix}:{x}"'
+        )
+        self.assertEqual(self.request("PATCH", "/label", source)[0], 200)
+        self.assertEqual(
+            self.request("POST", "/label", b'{"x":4,"prefix":"order"}'),
+            (200, b'"order:4"'),
+        )
+        self.assertEqual(
+            self.request("POST", "/label", b'{"x":"abc"}'),
+            (200, b'"item:abc"'),
+        )
+        for invalid in (b'{"x":4.5}', b'{"prefix":"order"}', b'{"x":4,"extra":1}'):
+            self.assertEqual(self.request("POST", "/label", invalid)[0], 400)
+        schema = json.loads(self.request("GET", "/")[1])["functions"][0]["schema"]
+        self.assertEqual(schema["input"]["$schema"], JSON_SCHEMA_DRAFT)
+        self.assertEqual(schema["input"]["required"], ["x"])
+        self.assertEqual(
+            {item["type"] for item in schema["input"]["properties"]["x"]["anyOf"]},
+            {"integer", "string"},
+        )
+        self.assertEqual(schema["input"]["properties"]["prefix"]["default"], "item")
+
+    def test_model_input_and_output(self):
+        source = b"""from pydantic import BaseModel
+class Order(BaseModel):
+    quantity: int
+    unit_price: float
+def total(x: Order) -> Order:
+    return x
+"""
+        self.assertEqual(self.request("PATCH", "/total", source)[0], 200)
+        self.assertEqual(
+            json.loads(
+                self.request(
+                    "POST", "/total", b'{"x":{"quantity":2,"unit_price":3.5}}'
+                )[1]
+            ),
+            {"quantity": 2, "unit_price": 3.5},
+        )
+        self.assertEqual(
+            json.loads(
+                self.request("POST", "/total", b'{"quantity":2,"unit_price":3.5}')[1]
+            ),
+            {"quantity": 2, "unit_price": 3.5},
+        )
+        self.assertEqual(
+            self.request("POST", "/total", b'{"x":{"quantity":"2","unit_price":3.5}}')[
+                0
+            ],
+            400,
+        )
+        entry = json.loads(self.request("GET", "/")[1])["functions"][0]
+        self.assertEqual(entry["schema"]["input"]["$schema"], JSON_SCHEMA_DRAFT)
+        self.assertEqual(
+            entry["schema"]["input"]["$defs"]["Order"]["properties"]["quantity"][
+                "type"
+            ],
+            "integer",
+        )
+
+    def test_single_parameter_accepts_every_json_value(self):
+        source = b"from typing import Any\ndef identity(x: Any) -> Any:\n    return x"
+        self.assertEqual(self.request("PATCH", "/identity", source)[0], 200)
+        for value in (
+            b"4",
+            b'"hello"',
+            b"true",
+            b"false",
+            b"null",
+            b"[1,2]",
+            b'{"item":1}',
+        ):
+            status, response = self.request("POST", "/identity", value)
+            self.assertEqual(status, 200, value)
+            self.assertEqual(json.loads(response), json.loads(value))
+        self.assertEqual(self.request("POST", "/identity", b'{"x":4}'), (200, b"4"))
+
+
+if __name__ == "__main__":
+    unittest.main()