Metadata-Version: 2.4
Name: stupidhuman-func
Version: 0.2.3
Summary: Decorator library for turning plain Python functions into Azure HTTP-triggered Functions
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: azure-functions>=1.24.0
Requires-Dist: azurefunctions-extensions-http-fastapi
Requires-Dist: httpx>=0.24.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Dynamic: requires-python

# stupidhuman-func

A decorator library that turns plain Python functions into Azure Functions — HTTP triggers, streaming responses, queue triggers, static file serving, Azure MCP tool triggers, and self-hosted MCP servers. Handles parameter extraction, type coercion, OAuth2/JWT scope validation, and error responses automatically.

## Installation

```bash
pip install stupidhuman-func
```

In `requirements.txt`:

```
stupidhuman-func==0.2.0
```

## Quick start

```python
# function_app.py
from azure_func import ServiceHandler

sh = ServiceHandler()
app = sh.func_app   # Azure Functions runtime entry point

@sh.service()
def hello(name: str):
    return f"Hello, {name}!"

# GET /hello?name=World  →  {"result": "Hello, World!"}
```

`app` must be the name of the `FunctionApp` instance in `function_app.py` — that is what the Azure Functions runtime discovers.

---

## `ServiceHandler`

The main class. Wraps `azure.functions.FunctionApp` and adds decorators for all trigger types.

```python
from azure_func import ServiceHandler
import azure.functions as func

sh = ServiceHandler(http_auth_level=func.AuthLevel.ANONYMOUS)
app = sh.func_app
```

### Decorators

| Decorator | Trigger type |
|---|---|
| `@sh.service()` | HTTP trigger, scope-protected |
| `@sh.anonymous()` | HTTP trigger, always public |
| `@sh.stream()` | Streaming HTTP trigger (SSE / token streaming) |
| `@sh.queue(name)` | Azure Storage Queue trigger |
| `@sh.static(folder)` | Catch-all GET, serves files from a folder |
| `@sh.tool()` | Azure MCP tool trigger (built-in Azure binding) |
| `@sh.scope(*scopes)` | Attaches required OAuth2 scopes to any handler |
| `@sh.tool_property(arg, desc)` | Adds parameter description to `@sh.tool()` |
| `sh.mcp_server(name, route)` | Creates a self-hosted MCP server at a route |

All decorators return the **original function unchanged**, so functions remain directly callable in tests and other code.

---

## HTTP triggers

### `@sh.service(route=None, methods=None)`

Registers an HTTP trigger. Reads any `@sh.scope()` metadata on the function and enforces it on every request.

### `@sh.anonymous(route=None, methods=None)`

Same as `@sh.service()` but always public — no scope check. Use this to make intent explicit when other endpoints on the same handler are scope-protected.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `route` | `str` | function name | URL path, supports `{param}` route segments |
| `methods` | `list[str]` | `["GET", "POST"]` | Accepted HTTP methods |

```python
@sh.anonymous()
def health():
    return "ok"

@sh.service()
@sh.scope("read:items")
def get_item(item_id: int):
    return {"id": item_id}

@sh.service(route="items/{item_id}", methods=["GET"])
def get_item_by_route(item_id: int):
    return {"id": item_id}

# GET /items/42  →  {"result": {"id": 42}}
```

---

## Parameter handling

Parameters are resolved from the HTTP request in this priority order:

1. **Query string** — `?key=value`
2. **Route params** — `{param}` segments in the route template
3. **JSON body** — `Content-Type: application/json` with `{"key": "value"}`

Query string wins when the same key appears in multiple sources. Parameters without a default value are required; omitting them returns **HTTP 400**.

### Type coercion

| Annotation | Behaviour |
|---|---|
| `str` | no-op |
| `int` | `int(value)` |
| `float` | `float(value)` |
| `bool` | `False` for `"false"`, `"0"`, `"no"`, `""`; `True` for everything else |
| `bytes` | raw binary request body — bypasses query/route/JSON extraction entirely |
| any other callable | called with the string value |

Failed coercion returns **HTTP 400** before the function is invoked.

### Raw binary body

Annotate a parameter with `bytes` to receive the raw request body directly. This is independent of other parameters, which are still resolved normally from query string or route params.

```python
@sh.service(route="files/{filename}", methods=["PUT"])
def upload_file(data: bytes, filename: str):
    store(filename, data)
    return {"size": len(data)}

# PUT /files/report.pdf  (binary body)
# filename comes from the route, data is the raw bytes
```

A `bytes` parameter never causes a 400 — it receives `b""` if the body is empty.

---

## Response format

### Success

```json
{"result": <return value>}
```

HTTP **200**.

### Error responses

| Situation | Status | Body |
|---|---|---|
| Missing required parameter(s) | 400 | `{"error": "Missing required parameter(s): x, y"}` |
| Type coercion failure | 400 | `{"error": "Invalid value for 'n': expected int, got 'abc'", "detail": "..."}` |
| Missing or insufficient scope | 403 | `{"error": "Insufficient scope", "required": ["scope1"]}` |
| Unhandled exception | 500 | `{"error": "Internal server error", "detail": "..."}` |

---

## OAuth2 / JWT scope validation

```python
@sh.service()
@sh.scope("read:items")
@sh.scope("write:items")   # cumulative — both required
def update_item(item_id: int, value: str):
    ...

# Equivalent:
@sh.service()
@sh.scope("read:items", "write:items")
def update_item(item_id: int, value: str):
    ...
```

`@sh.scope()` must be placed **below** `@sh.service()` (closer to the function). Decorators are cumulative — stacking calls accumulates all listed scopes.

The `Authorization: Bearer <jwt>` header is required. Scopes are read from the JWT payload without signature verification — the Azure Functions host or an upstream API gateway is expected to validate the token.

Supported claims:
- `scp` — Azure AD format, space-separated string
- `scope` — standard OAuth2, space-separated string or list

Missing header, missing scopes, or malformed token all return **HTTP 403**.

---

## Streaming responses

`@sh.stream()` registers an async generator as a streaming HTTP trigger using the [`azurefunctions-extensions-http-fastapi`](https://pypi.org/project/azurefunctions-extensions-http-fastapi/) extension. Use this for SSE, LLM token streaming, or any response too large to buffer.

### `@sh.stream(route=None, methods=None, media_type="text/event-stream")`

| Parameter | Type | Default | Description |
|---|---|---|---|
| `route` | `str` | function name | URL path |
| `methods` | `list[str]` | `["GET", "POST"]` | Accepted HTTP methods |
| `media_type` | `str` | `"text/event-stream"` | `Content-Type` of the streamed response |

The decorated function must be an **async generator** (`async def` with `yield`). Parameters and `@sh.scope()` work identically to `@sh.service()`.

```python
@sh.stream()
async def count(n: int):
    for i in range(n):
        yield f"data: {i}\n\n"

@sh.stream(route="openai-chat", methods=["POST"], media_type="text/event-stream")
@sh.scope("api:access")
async def chat(message: str, model: str = "gpt-4o"):
    async for chunk in stream_chat([{"role": "user", "content": message}], model):
        if '"type": "done"' in chunk:
            break
        yield chunk
```

### Required environment variables

```json
{
  "Values": {
    "PYTHON_ENABLE_INIT_INDEXING": "1",
    "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"
  }
}
```

`PYTHON_ENABLE_INIT_INDEXING=1` is required for all deployments that use streaming.  
`PYTHON_ISOLATE_WORKER_DEPENDENCIES=1` is required on Linux Consumption and recommended elsewhere.

### Limitations

- Works on **Consumption** and **Flex Consumption** plans
- Does not work on **Premium** or **Dedicated** plans (known Azure issue)

---

## Queue triggers

### `@sh.queue(queue_name, connection="AzureWebJobsStorage")`

| Parameter | Type | Default | Description |
|---|---|---|---|
| `queue_name` | `str` | — | Name of the storage queue |
| `connection` | `str` | `"AzureWebJobsStorage"` | App setting name for the storage connection string |

```python
import azure.functions as func

@sh.queue("orders")
def process_order(msg: func.QueueMessage):
    order = msg.get_json()
    raw = msg.get_body().decode()

@sh.queue("events", connection="EventsStorageConnection")
def process_event(msg: func.QueueMessage):
    ...
```

Unhandled exceptions are logged and re-raised, triggering Azure's poison-message retry policy.

---

## Static files

`@sh.static(folder)` serves files from a local folder at the root URL.

```python
@sh.static('public')
def serve_static():
    pass

# GET /style.css        →  ./public/style.css
# GET /js/app.js        →  ./public/js/app.js
```

| Situation | Status |
|---|---|
| File found | 200 with correct `Content-Type` |
| File not found | 404 |
| Path traversal attempt | 404 |

The catch-all route `{*filepath}` has lower specificity than named routes, so all `@sh.service()` and MCP endpoints take precedence.

---

## Azure MCP tool triggers

`@sh.tool()` registers a function as an [Azure Functions MCP tool trigger](https://learn.microsoft.com/en-us/azure/azure-functions/functions-bindings-mcp-tool-trigger), turning your function app into a remote MCP server compatible with VS Code Agent Mode, Claude Desktop, and other MCP clients.

Requires `azure-functions >= 1.25.0b2` and the preview extension bundle in `host.json`.

### `@sh.tool()`

The **function name** becomes the tool name, the **docstring** becomes the tool description, and each parameter is registered as an MCP tool property.

### `@sh.tool_property(arg_name, description)`

Attaches a description to a parameter. Must be placed **below** `@sh.tool()`.

```python
@sh.tool()
@sh.tool_property("item_id", "The ID of the item to retrieve.")
def get_item(item_id: int) -> str:
    """Get an item by ID."""
    return str(item_id)

@sh.tool()
@sh.tool_property("name", "The item name.")
@sh.tool_property("price", "The price in USD.")
def create_item(name: str, price: float) -> str:
    """Create a new item."""
    return f"Created {name} at ${price}"
```

### Authentication

Authentication is handled at the Azure host level via the `mcp_extension` system key — `@sh.scope()` is not supported on tools. Clients pass the key as `?code=<key>` or the `x-functions-key` header.

### `host.json`

```json
{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle.Preview",
    "version": "[4.*, 5.0.0)"
  }
}
```

---

## Self-hosted MCP servers

`sh.mcp_server()` creates independent MCP servers exposed as POST routes within the same Function App. Use this when you need to expose different groups of tools at separate endpoints — for example to serve different customers, domains, or agents with only the tools they need.

Unlike `@sh.tool()`, these servers do not require the Azure MCP binding or preview extension bundle, and they support `@sh.scope()` style authentication through bearer tokens (not yet built in — add your own middleware if needed).

### `sh.mcp_server(name, route=None)` → `McpToolServer`

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | — | Server name, returned in `serverInfo` |
| `route` | `str` | `"mcp/<name>"` | URL path for the POST endpoint |

Returns a `McpToolServer` instance. Register tools on it with `@server.tool()`.

### `@server.tool()`

Registers the function as a tool. The **function name** becomes the tool name, the **docstring** becomes the tool description.

### `@server.tool_property(arg_name, description)`

Attaches a description to a parameter. Must be placed **below** `@server.tool()`.

```python
sales    = sh.mcp_server("sales",    route="mcp/sales")
invoices = sh.mcp_server("invoices", route="mcp/invoices")

@sales.tool()
@sales.tool_property("customer_id", "The customer's unique identifier.")
def lookup_customer(customer_id: str) -> dict:
    """Look up a customer account."""
    return db.get_customer(customer_id)

@invoices.tool()
@invoices.tool_property("invoice_id", "The invoice's unique identifier.")
@invoices.tool_property("include_lines", "Include line items in the response.")
def get_invoice(invoice_id: str, include_lines: bool = True) -> dict:
    """Retrieve an invoice by ID."""
    return db.get_invoice(invoice_id, include_lines)

# Tools on different servers can share the same function name:
@sales.tool()
def get_by_id(customer_id: str) -> dict:
    """Get a customer by ID."""
    ...

@invoices.tool()
def get_by_id(invoice_id: str) -> dict:
    """Get an invoice by ID."""
    ...
```

### JSON-RPC 2.0 protocol

Each server handles these methods over `POST /<route>`:

| Method | Description |
|---|---|
| `initialize` | Handshake — returns `serverInfo` and capabilities |
| `notifications/initialized` | Client acknowledgement — returns 204 |
| `tools/list` | Returns the list of tools with input schemas |
| `tools/call` | Invokes a tool by name with arguments |

`tools/list` example response:

```json
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "lookup_customer",
        "description": "Look up a customer account.",
        "inputSchema": {
          "type": "object",
          "properties": {
            "customer_id": {
              "type": "string",
              "description": "The customer's unique identifier."
            }
          },
          "required": ["customer_id"]
        }
      }
    ]
  }
}
```

`tools/call` example request and response:

```json
{ "jsonrpc": "2.0", "id": 2, "method": "tools/call",
  "params": { "name": "lookup_customer", "arguments": { "customer_id": "C123" } } }

{ "jsonrpc": "2.0", "id": 2,
  "result": { "content": [{ "type": "text", "text": "{\"id\": \"C123\", ...}" }] } }
```

Tool errors are returned as `isError: true` in the result (not as JSON-RPC errors) so the LLM can see and handle the failure message.

---

## OpenAI streaming helper

`stream_chat` streams an OpenAI Responses API call as SSE chunks. It handles MCP tool-call continuations automatically — if the model calls a tool without producing text, it re-submits with `previous_response_id` until it presents a final response.

```python
from azure_func import stream_chat

@sh.stream(route="chat", methods=["POST"])
async def chat_endpoint(message: str, model: str = "gpt-4o"):
    async for chunk in stream_chat(
        messages=[{"role": "user", "content": message}],
        model=model,
    ):
        if '"type": "done"' in chunk:
            # Terminal event — contains full assistant text for persistence
            # Do not forward to the client
            break
        yield chunk
```

### SSE event types

| `type` field | Description |
|---|---|
| `content` | A text delta from the model — forward to the client |
| `tool_call` | The model is calling an MCP tool — forward if you want to show progress |
| `tool_done` | The tool call completed — forward if you want to show progress |
| `error` | OpenAI returned an error |
| `done` | Terminal event with full `content` — do **not** forward; use to persist the turn |

### Environment variable

```
OPENAI_API_KEY=sk-...
```

---

## Project layout

```
my_project/
├── function_app.py      ← entry point; register all routes here
├── host.json
├── local.settings.json
└── requirements.txt
```

**`local.settings.json`** (minimum for streaming support):

```json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "PYTHON_ENABLE_INIT_INDEXING": "1",
    "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"
  }
}
```

**`requirements.txt`** (minimum):

```
stupidhuman-func==0.2.0
```

---

## Running tests

```bash
pip install -e '.[dev]'
pytest
```

Tests mock the Azure SDK entirely — no Azure account or Functions runtime needed.
