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

ServiceHandler(http_auth_level=..., raw=False, **kwargs)

raw sets the default for the raw option (see HTTP triggers) on every @sh.service()/@sh.anonymous() route registered through this handler, so you don't have to repeat raw=True on each one:

sh = ServiceHandler(raw=True)

@sh.service(route="items/{item_id}")
def get_item(item_id: int):
    return {"id": item_id}

# GET /items/42  →  {"id": 42}   (no "result" wrapper, inherited from the handler)

@sh.service(route="legacy", raw=False)
def legacy_endpoint():
    return {"ok": True}

# GET /legacy  →  {"result": {"ok": True}}   (explicit raw=False overrides the handler default)

Any raw= passed directly to @sh.service()/@sh.anonymous() always wins over the handler's own setting — the handler-level raw is only a default for routes that don't specify their own.

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, raw=False)

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

@sh.anonymous(route=None, methods=None, raw=False)

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
raw	bool	False	If True, the return value becomes the response body directly, instead of being wrapped as {"result": ...}

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

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

# GET /items/42  →  {"id": 42}   (no "result" wrapper)

raw=True only changes the success response. Framework-level responses — missing/invalid parameters (400), scope errors (401/403), and unhandled exceptions (500) — are unaffected, since those are protocol-level signals, not your function's own return value.

---

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.

Whole JSON body as a dict

A parameter annotated dict receives the entire parsed JSON body directly, instead of being looked up as a named field, when it's the only parameter not covered by a route placeholder. Route params are structural (they come from the URL template, not the body), so they don't count against this:

@sh.service(route="users", methods=["POST"])
async def create_user(user: dict):
    return {"received": user}

# POST /users  {"name": "Ada", "email": "ada@example.com"}
# user == {"name": "Ada", "email": "ada@example.com"}

@sh.service(route="users/{id}", methods=["PUT"])
async def update_user(id: str, payload: dict):
    return {"id": id, "received": payload}

# PUT /users/42  {"name": "Ada"}
# id == "42" (from the route), payload == {"name": "Ada"} (the whole body)

With more than one parameter left over after excluding route params, a dict-annotated argument instead falls back to normal per-field extraction (e.g. def f(user: dict, note: str) with no route placeholders looks up user under a "user" key in the body, same as any other field) — this avoids ambiguity over which parameter means "everything". An invalid or non-object JSON body yields {} rather than a 400.

Any dict subclass works here too, including typing.TypedDict — a TypedDict is a plain dict at runtime, so it gets the whole body with no conversion, just static type-checking/autocomplete on the keys:

from typing import TypedDict

class CreateUserPayload(TypedDict):
    name: str
    email: str

@sh.service(route="users", methods=["POST"])
async def create_user(payload: CreateUserPayload):
    return {"received": payload["name"]}

---

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

Parameter annotations

Tool arguments map straight to a Python parameter of the same name — there's no HTTP-style "whole body" mode for MCP tools, so give every field its own parameter. str, int, float, bool, list, and dict produce a matching "type" in the generated JSON Schema; typing.Literal[...] produces {"type": ..., "enum": [...]}. Any other annotation (a TypedDict, Optional[...], a custom class, ...) is accepted as a parameter type but doesn't add schema type info beyond what @server.tool_property() provides, and the argument value is passed through as received (parsed JSON) rather than being coerced into that type.

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

---

Keyword generation helpers

generate_google_ads_keywords and generate_junglescout_keywords call the Google Ads and JungleScout keyword-ideas APIs, run the results through a shared LLM brand-safety filter, normalize/score/sort, and return a consistent payload shape. Like stream_chat, they are plain async functions meant to be called inside your own @sh.service() function — they are not decorated themselves.

Both raise ValueError on invalid input (a single message joining every failing field, e.g. "language: is required and must be a string, seed: must be one of ..."), and raise a generic Exception wrapping any upstream/pipeline failure. On success they return just the inner payload — {"message", "requestId", "timestamp", "data"} — not a full HTTP envelope; building an outer {success, data, message, timestamp} shape and picking 400 vs 500 based on the exception type is up to your wrapping function.

Because these functions do their own validation instead of relying on @sh.service()'s generic "missing required parameter" 400, give every parameter on your wrapping function a default of None so the framework never intercepts a request before your validation logic runs:

from azure_func import generate_google_ads_keywords

@sh.service(route="google-ads/keywords", methods=["POST"])
async def google_ads_keywords_endpoint(
    language: str = None,
    locations: list = None,
    brandName: str = None,
    seed: str = None,
    keywords: list = None,
    url: str = None,
    site: str = None,
    save: bool = None,
    useAiFilterForOtherBrands: bool = None,
    normalizationType: str = None,
    aiPrompt: str = None,
):
    try:
        data = await generate_google_ads_keywords(
            language=language, locations=locations, brandName=brandName, seed=seed,
            keywords=keywords, url=url, site=site, save=save,
            useAiFilterForOtherBrands=useAiFilterForOtherBrands,
            normalizationType=normalizationType, aiPrompt=aiPrompt,
        )
        return {"success": True, "data": data, "message": data["message"]}
    except ValueError as exc:
        return {"success": False, "data": None, "message": str(exc)}   # 400-equivalent

generate_junglescout_keywords follows the same pattern — every field from its request shape (market, keyword, asins, categories, includeVariants, the min/max volume and count filters, sortBy, pageSize, pageCursor, useAiFilter, brandName, aiPrompt) is its own keyword argument, defaulting to None.

from azure_func import generate_junglescout_keywords

@sh.service(route="junglescout/keywords", methods=["POST"])
async def junglescout_keywords_endpoint(
    market: str = None,
    keyword: str = None,
    asins: list = None,
    brandName: str = None,
    useAiFilter: bool = None,
    aiPrompt: str = None,
):
    try:
        data = await generate_junglescout_keywords(
            market=market, keyword=keyword, asins=asins,
            brandName=brandName, useAiFilter=useAiFilter, aiPrompt=aiPrompt,
        )
        return {"success": True, "data": data, "message": data["message"]}
    except ValueError as exc:
        return {"success": False, "data": None, "message": str(exc)}

Both are Google Ads/JungleScout-specific ports — they don't implement the source API's CSV/Blob Storage export (save is accepted but always takes the "not saved" branch) or its file-based monthly rate limiter, both of which are Azure-infra-specific concerns left for you to add if needed.

Environment variables

GOOGLE_ADS_SERVICE_ACCOUNT_JSON
GOOGLE_ADS_LOGIN_CUSTOMER_ID
GOOGLE_ADS_DEVELOPER_TOKEN
GOOGLE_ADS_API_VERSION            # optional, default "v20"

OPENAI_API_KEY
OPENAI_MODEL                      # optional, default "gpt-5.6-luna"
AI_PROMPT                         # optional, used verbatim (no brandName substitution)

JUNGLESCOUT_API_KEY
JUNGLESCOUT_API_URL
JUNGLESCOUT_API_KEY_NAME

---

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.