Python: [Feature]: Sample function middleware for paid tool preflight and receipt

[nonameuserd]

Description

Problem

Agents that call tools which move money (POs, refunds, paid APIs) need an authorize → execute → evidence path outside the LLM. Agent Framework already has function middleware for intercepting tool calls; a sample would show how to wire that hook to an external spend-authorization layer.

Proposal

Add a focused Python sample under python/samples/02-agents/ (middleware/tools) that:

1. Registers a paid tool (e.g. submit_po / issue refund)
2. Uses function middleware to preflight authorization before the tool body runs
3. On allow: runs the tool, then submits completion evidence / receipt
4. On deny: returns an error to the agent without executing the side effect

Dependencies via PEP 723 inline script metadata (e.g. paybond-kit), not added to the root pyproject.toml — matching SAMPLE_GUIDELINES.md.

Why here

Complements existing middleware samples (logging/guardrails) with a concrete “economic side effect” pattern. Does not propose a first-party package under packages/ — sample-only unless maintainers prefer a different placement.

Alternatives considered

• Standalone external repo only (fine as fallback if in-tree third-party deps are unwanted)
• Agent-run middleware instead of function middleware (too coarse for per-tool spend)

Willingness

I can implement and open a PR against main following SAMPLE_GUIDELINES.md and CONTRIBUTING.md once maintainers confirm folder placement.

References

• SAMPLE_GUIDELINES: python/samples/SAMPLE_GUIDELINES.md
• Function middleware docs: https://learn.microsoft.com/en-us/agent-framework/agents/middleware/
• Related pattern (CrewAI): https://github.com/nonameuserd/paybond-crewai-procurement-agent

Code Sample

import asyncio
import os
from collections.abc import Awaitable, Callable, Mapping
from typing import Annotated, Any
from uuid import uuid4

from agent_framework import Agent, FunctionInvocationContext, tool
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import BaseModel, Field

from paybond_kit import Paybond
from paybond_kit.agent.registry import create_paybond_tool_registry
from paybond_kit.spend_guard import PaybondSpendApprovalRequiredError, PaybondSpendDeniedError


@tool(approval_mode="never_require")
def submit_po(
    amount_cents: Annotated[int, Field(description="PO amount in cents.")],
) -> str:
    """Submit a purchase order."""
    return f'{{"status":"completed","cost_cents":{amount_cents},"po_id":"po-{amount_cents}"}}'


def _as_mapping(arguments: BaseModel | Mapping[str, Any]) -> dict[str, Any]:
    if isinstance(arguments, BaseModel):
        return arguments.model_dump()
    return dict(arguments)


async def main() -> None:
    paybond = await Paybond.open(api_key=os.environ["PAYBOND_API_KEY"])
    run = await paybond.agent_run.bind(
        {
            "bootstrap": {
                "kind": "sandbox",
                "operation": "submit_po",
                "requested_spend_cents": 12000,
                "completion_preset": "cost_and_completion",
            },
            "registry": create_paybond_tool_registry(
                {
                    "default_deny": True,
                    "side_effecting": {
                        "submit_po": {
                            "operation": "submit_po",
                            "evidence_preset": "cost_and_completion",
                            "spend_cents": lambda args: int(args["amount_cents"]),
                        }
                    },
                }
            ),
        }
    )

    async def paybond_spend_gate(
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        tool_call_id = str(context.metadata.get("call_id") or uuid4())

        async def execute() -> object:
            await call_next()
            return context.result

        try:
            wrapped = await run.interceptor.wrap_execute(
                tool_name=context.function.name,
                tool_call_id=tool_call_id,
                arguments=_as_mapping(context.arguments),
                execute=execute,
            )
            context.result = wrapped.tool_result
        except (PaybondSpendDeniedError, PaybondSpendApprovalRequiredError) as exc:
            context.result = str(exc)

    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="ProcurementAgent",
            instructions="Submit POs only within policy.",
            tools=submit_po,
            middleware=[paybond_spend_gate],
        ) as agent,
    ):
        result = await agent.run("Submit a PO for $120")
        print(result.text)


if __name__ == "__main__":
    asyncio.run(main())

Language/SDK

Python

[rpelevin]

A sample here would be most useful if it makes the point of no return and receipt contract explicit, not only the allow/deny branch.

I would keep these invariants visible in code and tests:

• Build a canonical spend envelope before call_next(): tool name, tool-call id, normalized arguments hash, amount, currency, payee/resource, policy version, and expiry.
• Bind one authorization reference to exactly that envelope. Persist a non-reusable executing claim before call_next() so retries or concurrent middleware invocations cannot spend twice.
• Denied or approval-required outcomes return a typed not-executed result and never call the tool.
• After call_next(), record succeeded, failed, or execution-unknown; bind the receipt to the authorization ref, execution id, provider ref, actual cost, and evidence digest.
• Changed amount, currency, payee/resource, arguments, policy version, or expired authority requires a fresh authorization. A transport error after provider acceptance must not be reported as a clean denial.

Regression cases I would include:

1. Deny and approval-required paths prove the tool body was never entered.
2. Two concurrent invocations with one authorization produce one execution and one duplicate/no-effect result.
3. A retry with changed spend fields cannot reuse the prior authorization.
4. A tool exception produces an error receipt tied to the execution attempt.
5. A timeout after simulated provider acceptance produces execution-unknown and a reconciliation handle, not an automatic retry.

Because the sample uses approval_mode="never_require", it should also say explicitly that the external middleware is the sole authority surface for this example; otherwise readers may accidentally compose two independent approval systems.

Boundary: architecture and regression-test feedback only. I did not run the sample or validate Agent Framework, middleware behavior, the proposed package, payment behavior, implementation correctness, security, production readiness, official alignment, adoption, partnership, integration, customer interest, or Neura usage.

[nonameuserd]

I agree that the sample should teach more than allow/deny: envelope-bound authorization before call_next(), a non-reusable execution claim, typed not-executed on deny/hold, and post-execution receipts that distinguish success, tool failure, and execution-unknown.

I’ll keep the PR focused on middleware, one paid tool, deny/hold never entering the body, and an explicit note that with approval_mode="never_require" the external gate is the sole authority. The concurrency, auth-reuse, and timeout-unknown cases are good follow-on tests if maintainers want that depth in-tree.

Still waiting on confirmation of sample placement under python/samples/02-agents/ before opening the PR.