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

pip install stupidhuman-func

In requirements.txt:

stupidhuman-func==0.2.0

Quick start

# 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.

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

@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.

@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

{"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/invalid credentials	401	{"error": "<reason>"}
Insufficient scope	403	{"error": "Insufficient scope", "required": ["scope1"]}
Unhandled exception	500	{"error": "Internal server error", "detail": "..."}

---

OAuth2 / JWT scope validation

@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.

An Authorization: Bearer <token> header is required. The token is authenticated one of two ways, based on its shape:

JWTs (token contains two . separators)

Verified cryptographically (RS256) against the JWT_PUBLIC_KEY environment variable, which must hold a PEM-encoded RSA public key (literal \n sequences are unescaped automatically, so the key can be stored as a single-line env var). If JWT_PUBLIC_KEY is not set, or the signature doesn't verify, the request is rejected — the token is never trusted unverified.

Scopes are read from the verified payload:

• scp — Azure AD format, space-separated string
• scope — standard OAuth2, space-separated string or list

API keys (any other token shape)

Looked up against the ALLOWED_API_KEYS environment variable, a comma-separated list of key:scope1 scope2 pairs:

ALLOWED_API_KEYS=key1:read:items write:items,key2:read:items

The scopes after : are space-separated and become the scopes that key grants.

Errors

• Missing header, malformed header, invalid JWT signature, or unrecognized API key → HTTP 401
• Valid credentials but missing a required scope → HTTP 403

---

Streaming responses

@sh.stream() registers an async generator as a streaming HTTP trigger using the 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().

@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

{
  "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

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.

@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, 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().

@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

{
  "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 scope-based authentication through @server.scope(), checked against the same JWT_PUBLIC_KEY / ALLOWED_API_KEYS credentials described in OAuth2 / JWT scope validation. This is not available for @sh.tool() — see Authentication above; Azure's MCP tool trigger calls the handler directly with only the tool's own arguments, with no request or headers available to check.

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().

@server.scope(*scopes)

Attaches required scope(s) to a tool, enforced on every tools/call for that tool. Must be placed below @server.tool(). Cumulative, and authenticated identically to @sh.scope() — see OAuth2 / JWT scope validation for how JWT_PUBLIC_KEY and ALLOWED_API_KEYS are checked. tools/list is unaffected — it always lists every tool, scoped or not.

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.")
@sales.scope("read:customers")
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:

{
  "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:

{ "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.

Auth failures on a @server.scope()-protected tool are JSON-RPC errors instead (HTTP 200, per JSON-RPC convention):

Situation	Error code	Message
Missing/invalid credentials	-32001	reason (e.g. "Invalid API key")
Insufficient scope	-32002	"Insufficient scope: missing <scopes>"

---

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.

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):

{
  "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

pip install -e '.[dev]'
pytest

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