Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions python/samples/02-agents/middleware/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,3 +50,45 @@ agent's latest response to a second, external judge chat client on every iterati
or malicious judge endpoint could exfiltrate that data, or return a manipulated verdict/gap
analysis that gets fed back into the loop as feedback — a form of indirect prompt injection. Only
configure a judge client that points at a service you trust as much as the primary model.

### Per-tool execution deadlines

The `per_tool_execution_deadlines.py` sample demonstrates independent
asynchronous deadlines using the existing `FunctionMiddleware` extension
point and Python 3.10-compatible `asyncio` APIs.

Run from the `python` directory:

```bash
python samples/02-agents/middleware/per_tool_execution_deadlines.py
python -m pytest samples/02-agents/middleware/test_per_tool_execution_deadlines.py -v
```

Each configured tool has an independent deadline. Unconfigured tools
execute normally.

- **Read-only tools:** Deadline expiry produces a recoverable result.
- **Side-effecting tools:** Deadline expiry raises `MiddlewareFailure`
because side effects may already have occurred.
- **Tool-generated timeouts:** A tool's own `TimeoutError` propagates
separately from an enforced deadline.
- **Caller cancellation:** Cancellation propagates rather than being
converted into a normal timeout result.
- **Concurrency:** Different tools can have independent budgets.

The sample introduces no new framework API and does not implement
automatic retries or distributed deadline enforcement.

**Cancellation and cleanup limitations:** Async cancellation is
cooperative. Cleanup can take longer than the configured deadline.
Synchronous worker threads or remote operations may continue running.
Deadline expiry does not guarantee rollback, exactly-once execution,
or absence of side effects.

Caller cancellation is propagated after requesting cancellation of the
underlying asynchronous task. That task may continue cooperative cleanup
after the caller has been cancelled. Any eventual task exception is
observed to avoid an unhandled-task warning.

This sample does not provide hard termination, rollback, exactly-once
execution, or automatic retries.
130 changes: 130 additions & 0 deletions python/samples/02-agents/middleware/per_tool_execution_deadlines.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
# Copyright (c) Microsoft. All rights reserved.
"""Per-tool cooperative async deadlines using FunctionMiddleware.

Run this sample with Python 3.10 or later.
"""

import asyncio
import math
from collections.abc import Awaitable, Callable
from dataclasses import dataclass

from agent_framework import FunctionInvocationContext, FunctionMiddleware, MiddlewareFailure


@dataclass(frozen=True)
class DeadlinePolicy:
"""Timeout budget and side-effect classification for one tool."""

seconds: float
read_only: bool = False

def __post_init__(self) -> None:
if not math.isfinite(self.seconds) or self.seconds <= 0:
raise ValueError("deadline must be a finite positive number")


class PerToolDeadlineMiddleware(FunctionMiddleware):
"""Enforce independent deadlines for configured async tools.

Cancellation is cooperative. The middleware waits for task cleanup;
therefore actual elapsed time may exceed the configured deadline.
"""

def __init__(self, policies: dict[str, DeadlinePolicy]) -> None:
self._policies = dict(policies)

@staticmethod
def _observe_task_completion(task: asyncio.Task[None]) -> None:
"""Retrieve exceptions from a task finishing after cancellation."""
if not task.cancelled():
task.exception()

async def process(
self,
context: FunctionInvocationContext,
call_next: Callable[[], Awaitable[None]],
) -> None:
policy = self._policies.get(context.function.name)
if policy is None:
await call_next()
return

task = asyncio.create_task(call_next())
deadline_cancel_requested = False
try:
done, _ = await asyncio.wait({task}, timeout=policy.seconds)

if done:
# Propagates tool-raised TimeoutError and other exceptions.
await task
return

# The deadline expired while the tool task was pending.
deadline_cancel_requested = True
task.cancel()

# Wait for cooperative cleanup without suppressing
# cancellation of the calling middleware task.
try:
await asyncio.shield(task)
except asyncio.CancelledError:
if not task.done():
# Caller cancellation during cleanup must propagate.
raise
if not task.cancelled():
# A completed child may itself have raised cancellation.
raise
except MiddlewareFailure:
# Preserve fatal downstream middleware failures.
raise
except Exception:
# Ordinary cleanup errors do not establish that the
# timed-out operation completed successfully.
pass

if policy.read_only:
context.result = f"Read-only tool '{context.function.name}' exceeded its deadline."
return

raise MiddlewareFailure(
f"Tool '{context.function.name}' exceeded its deadline; remote side effects may have occurred."
)

except asyncio.CancelledError:
# Avoid cancelling the child a second time while it is
# already performing deadline-triggered cleanup.
if not deadline_cancel_requested:
task.cancel()
task.add_done_callback(self._observe_task_completion)
# Preserve caller cancellation. The tool may continue cleanup.
raise


async def main() -> None:
"""Demonstrate recoverable and uncertain deadline expiry."""

from types import SimpleNamespace

middleware = PerToolDeadlineMiddleware({
"lookup": DeadlinePolicy(0.01, read_only=True),
"update": DeadlinePolicy(0.01, read_only=False),
})

async def slow_tool() -> None:
await asyncio.sleep(0.05)

for name in ("lookup", "update"):
context = SimpleNamespace(
function=SimpleNamespace(name=name),
result=None,
)
try:
await middleware.process(context, slow_tool)
print(f"{name}: {context.result}")
except MiddlewareFailure as exc:
print(f"{name}: {exc}")


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