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

Decorators are cumulative — stacking calls accumulates all listed scopes. @sh.scope() can be placed above or below @sh.service() in the stack; scope metadata is read fresh on every request, not captured once at decoration time, so order doesn't affect enforcement.

@sh.scope() with no scope arguments (@sh.scope()) still requires a valid, authenticated caller — it means "any authenticated request is fine," not "no check at all." Only never applying @sh.scope() makes a @sh.service() route public; use @sh.anonymous() to make that explicit, or to force a route public even if @sh.scope() also happens to be stacked on it.

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. Can be placed above or below @server.tool() — like @sh.scope(), metadata is read fresh on every call, not captured once at registration, so order doesn't matter. Cumulative, and authenticated identically to @sh.scope() — see OAuth2 / JWT scope validation for how JWT_PUBLIC_KEY and ALLOWED_API_KEYS are checked, and for the "no scope arguments still means *authenticated*, not *public*" rule, which applies here too. 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,
    excludeBrands: list = 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,
            excludeBrands=excludeBrands,
        )
        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, excludeBrands) 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,
    excludeBrands: list = None,
):
    try:
        data = await generate_junglescout_keywords(
            market=market, keyword=keyword, asins=asins,
            brandName=brandName, useAiFilter=useAiFilter, aiPrompt=aiPrompt,
            excludeBrands=excludeBrands,
        )
        return {"success": True, "data": data, "message": data["message"]}
    except ValueError as exc:
        return {"success": False, "data": None, "message": str(exc)}

excludeBrands

Both functions accept an optional excludeBrands: list[str] — any keyword containing one of the listed names as a case-insensitive substring is dropped. This is a deterministic filter (plain substring match), applied before the AI filter and independent of it — unlike brand-safety instructions folded into aiPrompt, it's guaranteed rather than left to the model's judgment, and it doesn't cost any extra LLM tokens since excluded keywords never reach the AI filter call. Use aiPrompt for fuzzier judgment calls (junk words, off-topic terms) that can't be expressed as a fixed list; use excludeBrands for known competitor names you want reliably removed every time.

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.

generate_junglescout_keywords only ever fetches one page from JungleScout — it never follows data.metadata.links.next automatically. JungleScout's own default order (when sortBy isn't set) is ascending by monthly_search_volume_exact, i.e. least-relevant keywords first, so sortBy defaults to "-monthly_search_volume_exact" (descending) instead of being left unset, ensuring the one page you get back actually contains the highest-volume keywords rather than the lowest. Pass your own sortBy (any of the enum values, optionally prefixed with - for descending — this is a JSON:API-style API) to override it. If you need the full result set for a high-volume term, you'll need to paginate yourself: read data.metadata.links.next off the response, extract its page[cursor] value, and call again with pageCursor=... until links.next is absent.

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.