diff --git a/eng/net10.0.props b/eng/net10.0.props
index 926dadc14..d2540a543 100644
--- a/eng/net10.0.props
+++ b/eng/net10.0.props
@@ -23,6 +23,7 @@
     <Features>$(Features);FEATURE_METADATA_READER</Features>
     <Features>$(Features);FEATURE_MMAP</Features>
     <Features>$(Features);FEATURE_NATIVE</Features>
+    <Features>$(Features);FEATURE_NET_ASYNC</Features>
     <Features>$(Features);FEATURE_OSPLATFORMATTRIBUTE</Features>
     <Features>$(Features);FEATURE_PIPES</Features>
     <Features>$(Features);FEATURE_PROCESS</Features>
diff --git a/eng/net8.0.props b/eng/net8.0.props
index 926dadc14..d2540a543 100644
--- a/eng/net8.0.props
+++ b/eng/net8.0.props
@@ -23,6 +23,7 @@
     <Features>$(Features);FEATURE_METADATA_READER</Features>
     <Features>$(Features);FEATURE_MMAP</Features>
     <Features>$(Features);FEATURE_NATIVE</Features>
+    <Features>$(Features);FEATURE_NET_ASYNC</Features>
     <Features>$(Features);FEATURE_OSPLATFORMATTRIBUTE</Features>
     <Features>$(Features);FEATURE_PIPES</Features>
     <Features>$(Features);FEATURE_PROCESS</Features>
diff --git a/eng/net9.0.props b/eng/net9.0.props
index 926dadc14..d2540a543 100644
--- a/eng/net9.0.props
+++ b/eng/net9.0.props
@@ -23,6 +23,7 @@
     <Features>$(Features);FEATURE_METADATA_READER</Features>
     <Features>$(Features);FEATURE_MMAP</Features>
     <Features>$(Features);FEATURE_NATIVE</Features>
+    <Features>$(Features);FEATURE_NET_ASYNC</Features>
     <Features>$(Features);FEATURE_OSPLATFORMATTRIBUTE</Features>
     <Features>$(Features);FEATURE_PIPES</Features>
     <Features>$(Features);FEATURE_PROCESS</Features>
diff --git a/src/core/IronPython/Compiler/Ast/AstMethods.cs b/src/core/IronPython/Compiler/Ast/AstMethods.cs
index 611bf3bb1..ac64c8eb3 100644
--- a/src/core/IronPython/Compiler/Ast/AstMethods.cs
+++ b/src/core/IronPython/Compiler/Ast/AstMethods.cs
@@ -80,6 +80,11 @@ internal static class AstMethods {
         public static readonly MethodInfo FormatString = GetMethod((Func<CodeContext, string, object, string>)PythonOps.FormatString);
         public static readonly MethodInfo GeneratorCheckThrowableAndReturnSendValue = GetMethod((Func<object, object>)PythonOps.GeneratorCheckThrowableAndReturnSendValue);
         public static readonly MethodInfo MakeCoroutine = GetMethod((Func<PythonFunction, MutableTuple, object, PythonCoroutine>)PythonOps.MakeCoroutine);
+#if FEATURE_NET_ASYNC
+        public static readonly MethodInfo AsTaskForAwait = GetMethod((Func<object, System.Threading.Tasks.Task<object>>)PythonOps.AsTaskForAwait);
+        public static readonly MethodInfo MakeAsyncCoroutine = GetMethod((Func<PythonFunction, Func<System.Threading.Tasks.Task<object>>, System.Threading.CancellationTokenSource, System.Runtime.CompilerServices.StrongBox<System.Exception>, PythonCoroutine>)PythonOps.MakeAsyncCoroutine);
+        public static readonly MethodInfo MakeAsyncGenerator = GetMethod((Func<PythonFunction, System.Collections.Generic.IAsyncEnumerable<object>, System.Runtime.CompilerServices.StrongBox<object>, System.Runtime.CompilerServices.StrongBox<System.Exception>, System.Threading.CancellationTokenSource, PythonAsyncGenerator>)PythonOps.MakeAsyncGenerator);
+#endif
 
         // builtins
         public static readonly MethodInfo Format = GetMethod((Func<CodeContext, object, string, string>)PythonOps.Format);
diff --git a/src/core/IronPython/Compiler/Ast/AwaitExpression.cs b/src/core/IronPython/Compiler/Ast/AwaitExpression.cs
index 94b960aae..88072d342 100644
--- a/src/core/IronPython/Compiler/Ast/AwaitExpression.cs
+++ b/src/core/IronPython/Compiler/Ast/AwaitExpression.cs
@@ -6,15 +6,41 @@
 
 using MSAst = System.Linq.Expressions;
 
+using IronPython.Runtime.Operations;
+
 using AstUtils = Microsoft.Scripting.Ast.Utils;
 
 namespace IronPython.Compiler.Ast {
     using Ast = MSAst.Expression;
 
     /// <summary>
-    /// Represents an await expression. Implemented as yield from expr.__await__().
+    /// Represents <c>await expr</c>. Under <c>FEATURE_NET_ASYNC</c> this compiles directly to a DLR async suspension point.
+    /// Otherwise it is desugared into <c>yield from expr.__await__()</c> against the enclosing generator-shaped coroutine state machine.
     /// </summary>
     public class AwaitExpression : Expression {
+#if FEATURE_NET_ASYNC
+        public AwaitExpression(Expression expression) {
+            Expression = expression;
+        }
+
+        public Expression Expression { get; }
+
+        public override MSAst.Expression Reduce() {
+            // await x -> AsyncHelpers-driven suspension on the Task produced by
+            // PythonOps.AsTaskForAwait(x).
+            return AstUtils.Await(
+                Ast.Call(
+                    AstMethods.AsTaskForAwait,
+                    AstUtils.Convert(Expression, typeof(object))));
+        }
+
+        public override void Walk(PythonWalker walker) {
+            if (walker.Walk(this)) {
+                Expression?.Walk(walker);
+            }
+            walker.PostWalk(this);
+        }
+#else
         private readonly Statement _statement;
         private readonly NameExpression _result;
 
@@ -60,6 +86,7 @@ public override void Walk(PythonWalker walker) {
             }
             walker.PostWalk(this);
         }
+#endif
 
         public override string NodeName => "await expression";
     }
diff --git a/src/core/IronPython/Compiler/Ast/FunctionDefinition.cs b/src/core/IronPython/Compiler/Ast/FunctionDefinition.cs
index 900ecb86f..97489e908 100644
--- a/src/core/IronPython/Compiler/Ast/FunctionDefinition.cs
+++ b/src/core/IronPython/Compiler/Ast/FunctionDefinition.cs
@@ -8,6 +8,7 @@
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 
 using IronPython.Runtime;
 using IronPython.Runtime.Operations;
@@ -117,7 +118,25 @@ internal override int KwOnlyArgCount {
 
         public Expression ReturnAnnotation { get; internal set; }
 
+#if FEATURE_NET_ASYNC
+        // Under .NET-async, async functions are compiled directly to a Task<object?> via the DLR's AsyncExpression
+        // rather than reused through the generator state machine, so IsAsync does not imply generator-shaped emission.
+        internal override bool IsGeneratorMethod => IsGenerator;
+
+        // Async-generator (PEP 525) channels. The StrongBox *values* are per-async-generator instance,
+        // allocated per stack frame at runtime in the function body. 
+        // Declared/assigned in the body, captured by the generator (its yields read them through Parent),
+        // and handed to the PythonAsyncGenerator wrapper, which writes them before each resume:
+        //   AsyncSendSlot  — the value of `x = yield z` (asend(v); None for __anext__/async for).
+        //   AsyncThrowSlot — an exception to rethrow at the yield resume point (athrow/aclose).
+        private readonly MSAst.ParameterExpression _asyncSendSlot = MSAst.Expression.Variable(typeof(StrongBox<object>), "$asyncSend");
+        private readonly MSAst.ParameterExpression _asyncThrowSlot = MSAst.Expression.Variable(typeof(StrongBox<Exception>), "$asyncThrow");
+
+        internal MSAst.ParameterExpression AsyncSendSlot => _asyncSendSlot;
+        internal MSAst.ParameterExpression AsyncThrowSlot => _asyncThrowSlot;
+#else
         internal override bool IsGeneratorMethod => IsGenerator || IsAsync;
+#endif
 
         /// <summary>
         /// The function is a generator
@@ -182,9 +201,15 @@ internal override FunctionAttributes Flags {
                     fa |= FunctionAttributes.ContainsTryFinally;
                 }
 
+#if FEATURE_NET_ASYNC
+                if (IsGenerator) {
+                    fa |= FunctionAttributes.Generator;
+                }
+#else
                 if (IsGenerator || IsAsync) {
                     fa |= FunctionAttributes.Generator;
                 }
+#endif
 
                 if (IsAsync) {
                     fa |= FunctionAttributes.Coroutine;
@@ -357,7 +382,13 @@ internal MSAst.Expression MakeFunctionExpression() {
                                 annotations
                             )
                         ),
+#if FEATURE_NET_ASYNC
+                    // Async generators are lowered via AsyncEnumerableExpression in the body,
+                    // so they must not be wrapped as a PythonGenerator here — only plain (non-async) generators are.
+                    (IsGenerator && !IsAsync) ?
+#else
                     (IsGenerator || IsAsync) ?
+#endif
                         (MSAst.Expression)new PythonGeneratorExpression(code, GlobalParent.PyContext.Options.CompilationThreshold, IsAsync) :
                         (MSAst.Expression)code
                 );
@@ -659,7 +690,13 @@ private LightLambdaExpression CreateFunctionLambda() {
             // For generators/coroutines, we need to do a check before the first statement for Generator.Throw() / Generator.Close().
             // The exception traceback needs to come from the generator's method body, and so we must do the check and throw
             // from inside the generator.
+#if FEATURE_NET_ASYNC
+            // Async generators have no backing PythonGenerator (they lower to IAsyncEnumerable via AsyncEnumerableExpression),
+            // so skip the $generator.CheckThrowable() prologue for them.
+            if (IsGenerator && !IsAsync) {
+#else
             if (IsGenerator || IsAsync) {
+#endif
                 MSAst.Expression s1 = YieldExpression.CreateCheckThrowExpression(SourceSpan.None);
                 statements.Add(s1);
             }
@@ -681,6 +718,59 @@ private LightLambdaExpression CreateFunctionLambda() {
             body = Ast.Block(body, AstUtils.Empty());
             body = AddReturnTarget(body);
 
+#if FEATURE_NET_ASYNC
+            // Under .NET-async, an `async def` body returns a PythonCoroutine wrapping a Task<object?>.
+            // We pre-allocate a CancellationTokenSource and a StrongBox<Exception?> here
+            // so the same instances are shared with both AsyncExpression, which threads them into AsyncHelpers.DriveAsync
+            // and PythonCoroutine, which uses them to implement coro.throw(exc) on a running coroutine:
+            // write the exception to the box, cancel the CTS, and DriveAsync surfaces that exception in place of OperationCanceledException.
+            if (IsAsync) {
+                var cts = MSAst.Expression.Variable(typeof(CancellationTokenSource), "$cts");
+                var excBox = MSAst.Expression.Variable(typeof(StrongBox<Exception>), "$cancelExc");
+                var ctToken = MSAst.Expression.Property(cts, nameof(CancellationTokenSource.Token));
+                if (IsGenerator) {
+                    // Async generator: the body has both `await` and `yield`. Lower it to an
+                    // IAsyncEnumerable<object?> via AsyncEnumerableExpression, sharing the generator label so
+                    // the body's yields and the rewritten awaits land in one generator, then wrap it in a
+                    // PythonAsyncGenerator. The send/throw slots are per-generator StrongBoxes captured by the
+                    // generator (the body's yields read them) AND handed to the wrapper, which writes them
+                    // before each resume — see AsyncSendSlot / AsyncThrowSlot.
+                    var sendSlot = AsyncSendSlot;
+                    var throwSlot = AsyncThrowSlot;
+                    body = MSAst.Expression.Block(
+                        [cts, excBox, sendSlot, throwSlot],
+                        MSAst.Expression.Assign(cts, MSAst.Expression.New(typeof(CancellationTokenSource))),
+                        MSAst.Expression.Assign(excBox, MSAst.Expression.New(typeof(StrongBox<Exception>))),
+                        MSAst.Expression.Assign(sendSlot, MSAst.Expression.New(typeof(StrongBox<object>))),
+                        MSAst.Expression.Assign(throwSlot, MSAst.Expression.New(typeof(StrongBox<Exception>))),
+                        Ast.Call(
+                            AstMethods.MakeAsyncGenerator,
+                            _functionParam,
+                            AstUtils.AsyncEnumerable(Name, body, GeneratorLabel, ctToken, excBox),
+                            sendSlot,
+                            throwSlot,
+                            cts));
+                } else {
+                    // Plain async def: the body returns a PythonCoroutine wrapping a Task<object?>.
+                    // Lazy start: hand MakeAsyncCoroutine a thunk (Func<Task<object?>>) instead of an already-running Task,
+                    // so the body doesn't execute until the coroutine is first driven (send/AsTask).
+                    // This makes calling an async def side-effect-free (PEP 492) and lets the body's first await capture the driver's SynchronizationContext
+                    // rather than whatever context happened to be current at construction.
+                    body = MSAst.Expression.Block(
+                        [cts, excBox],
+                        MSAst.Expression.Assign(cts, MSAst.Expression.New(typeof(CancellationTokenSource))),
+                        MSAst.Expression.Assign(excBox, MSAst.Expression.New(typeof(StrongBox<Exception>))),
+                        Ast.Call(
+                            AstMethods.MakeAsyncCoroutine,
+                            _functionParam,
+                            MSAst.Expression.Lambda<Func<Task<object>>>(
+                                AstUtils.Async(Name, body, ctToken, excBox)),
+                            cts,
+                            excBox));
+                }
+            }
+#endif
+
             MSAst.Expression bodyStmt = body;
             if (localContext != null) {
                 var createLocal = CreateLocalContext(_parentContext);
diff --git a/src/core/IronPython/Compiler/Ast/ReturnStatement.cs b/src/core/IronPython/Compiler/Ast/ReturnStatement.cs
index f8393f8b2..0896d004d 100644
--- a/src/core/IronPython/Compiler/Ast/ReturnStatement.cs
+++ b/src/core/IronPython/Compiler/Ast/ReturnStatement.cs
@@ -2,6 +2,8 @@
 // The .NET Foundation licenses this file to you under the Apache 2.0 License.
 // See the LICENSE file in the project root for more information.
 
+using System.Diagnostics;
+
 using MSAst = System.Linq.Expressions;
 
 namespace IronPython.Compiler.Ast {
@@ -17,6 +19,16 @@ public ReturnStatement(Expression expression) {
 
         public override MSAst.Expression Reduce() {
             if (Parent.IsGeneratorMethod) {
+#if FEATURE_NET_ASYNC
+                // An async generator (`async def` with `yield`) lowers through the DLR generator via AsyncEnumerableExpression,
+                // which doesn't understand IronPython's -2 "generator-return" marker.
+                // `return` there is always bare (`return value` is a SyntaxError in async generators)
+                // and simply ends the async iteration — map it to a YieldBreak without a value.
+                if (Parent is FunctionDefinition { IsAsync: true }) {
+                    Debug.Assert(Expression == null, "async generators should not have a return value");
+                    return GlobalParent.AddDebugInfo(AstUtils.YieldBreak(GeneratorLabel), Span);
+                }
+#endif
                 // Reduce to a yield return with a marker of -2, this will be interpreted as a yield break with a return value
                 return GlobalParent.AddDebugInfo(AstUtils.YieldReturn(GeneratorLabel, TransformOrConstantNull(Expression, typeof(object)), -2), Span);
             }
diff --git a/src/core/IronPython/Compiler/Ast/YieldExpression.cs b/src/core/IronPython/Compiler/Ast/YieldExpression.cs
index 266a157aa..37728d8e2 100644
--- a/src/core/IronPython/Compiler/Ast/YieldExpression.cs
+++ b/src/core/IronPython/Compiler/Ast/YieldExpression.cs
@@ -7,6 +7,8 @@
 using MSAst = System.Linq.Expressions;
 
 using System;
+using System.Runtime.CompilerServices;
+using System.Runtime.ExceptionServices;
 using System.Diagnostics;
 using Microsoft.Scripting;
 using Microsoft.Scripting.Runtime;
@@ -19,6 +21,13 @@ namespace IronPython.Compiler.Ast {
     //    x = yield z
     // The return value (x) is provided by calling Generator.Send()
     public class YieldExpression : Expression {
+#if FEATURE_NET_ASYNC
+        private static readonly System.Reflection.MethodInfo s_captureMethod
+            = typeof(ExceptionDispatchInfo).GetMethod(nameof(ExceptionDispatchInfo.Capture))!;
+        private static readonly System.Reflection.MethodInfo s_throwMethod
+            = typeof(ExceptionDispatchInfo).GetMethod(nameof(ExceptionDispatchInfo.Throw), Type.EmptyTypes)!;
+#endif
+
         public YieldExpression(Expression? expression) {
             Expression = expression;
         }
@@ -43,16 +52,40 @@ internal static MSAst.Expression CreateCheckThrowExpression(SourceSpan span) {
         }
 
         public override MSAst.Expression Reduce() {
+            MSAst.Expression yieldValue = Expression == null ? AstUtils.Constant(null) : AstUtils.Convert(Expression, typeof(object));
+
+#if FEATURE_NET_ASYNC
+            // An async generator (`async def` with `yield`) is lowered via AsyncEnumerableExpression
+            // and has no backing PythonGenerator, so there is no `$generator` to call CheckThrowable() on.
+            // Instead the resume reads two per-generator cells that PythonAsyncGenerator writes before advancing:
+            //   AsyncThrowSlot — if set (athrow/aclose), rethrow it here (preserving stack);
+            //                    cleared first so a body that catches it and yields again doesn't re-throw on the next resume.
+            //   AsyncSendSlot  — the value of the yield expression: the asend(v) value, or None.
+            if (Parent is FunctionDefinition { IsAsync: true } fd) {
+                MSAst.ParameterExpression sendSlot = fd.AsyncSendSlot;
+                MSAst.ParameterExpression throwSlot = fd.AsyncThrowSlot;
+                MSAst.ParameterExpression pending = Ast.Variable(typeof(Exception), "$athrow");
+                return Ast.Block(
+                    typeof(object),
+                    [pending],
+                    AstUtils.YieldReturn(GeneratorLabel, yieldValue),
+                    Ast.Assign(pending, Ast.Field(throwSlot, nameof(StrongBox<Exception>.Value))),
+                    Ast.Assign(Ast.Field(throwSlot, nameof(StrongBox<Exception>.Value)), Ast.Constant(null, typeof(Exception))),
+                    Ast.IfThen(
+                        Ast.ReferenceNotEqual(pending, Ast.Constant(null, typeof(Exception))),
+                        Ast.Call(Ast.Call(s_captureMethod, pending), s_throwMethod)),
+                    Ast.Field(sendSlot, nameof(StrongBox<object>.Value))
+                );
+            }
+#endif
+
             // (yield z) becomes:
             // .comma (1) {
             //    .void ( .yield_statement (_expression) ),
-            //    $gen.CheckThrowable() // <-- has return result from send            
+            //    $gen.CheckThrowable() // <-- has return result from send
             //  }
             return Ast.Block(
-                AstUtils.YieldReturn(
-                    GeneratorLabel,
-                    Expression == null ? AstUtils.Constant(null) : AstUtils.Convert(Expression, typeof(object))
-                ),
+                AstUtils.YieldReturn(GeneratorLabel, yieldValue),
                 CreateCheckThrowExpression(Span) // emits ($gen.CheckThrowable())
             );
         }
diff --git a/src/core/IronPython/Compiler/Parser.cs b/src/core/IronPython/Compiler/Parser.cs
index 502658084..5c7f8a6b6 100644
--- a/src/core/IronPython/Compiler/Parser.cs
+++ b/src/core/IronPython/Compiler/Parser.cs
@@ -1999,10 +1999,16 @@ private Expression ParseAtomExpr() {
                 if (current is null || !current.IsAsync) {
                     ReportSyntaxError("'await' outside async function");
                 }
+#if !FEATURE_NET_ASYNC
+                // Under the generator-based async path, `await` desugars to `yield from`,
+                // so the enclosing function must be marked a generator.
+                // Under FEATURE_NET_ASYNC, the body is lowered through AsyncExpression and is *not* a Python generator
+                // so this mark assignment is being skipped here.
                 if (current is not null) {
                     current.IsGenerator = true;
                     current.GeneratorStop = GeneratorStop;
                 }
+#endif
             }
 
             Expression ret = ParseAtom();
diff --git a/src/core/IronPython/Runtime/AsyncGenerator.cs b/src/core/IronPython/Runtime/AsyncGenerator.cs
new file mode 100644
index 000000000..78302dbe7
--- /dev/null
+++ b/src/core/IronPython/Runtime/AsyncGenerator.cs
@@ -0,0 +1,160 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+#nullable enable
+
+#if FEATURE_NET_ASYNC
+
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.Scripting.Runtime;
+
+using IronPython.Runtime.Exceptions;
+using IronPython.Runtime.Operations;
+using IronPython.Runtime.Types;
+
+namespace IronPython.Runtime {
+    /// <summary>
+    ///   PEP 525 async generator object — what an <c>async def</c> with <c>yield</c> returns.
+    /// </summary>
+    /// <remarks>
+    ///   Wrapper over the DLR's <c>IAsyncEnumerable&lt;object&gt;</c> (produced by AsyncEnumerableExpression /
+    ///   AsyncHelpers.DriveAsyncEnumerable) plus the channels that let the consumer drive it:
+    ///   <list type="bullet">
+    ///     <item><c>_sendSlot</c> — the value delivered into the body at the next <c>yield</c> resume
+    ///       (<c>x = yield z</c>). Set by <see cref="asend"/> / <see cref="__anext__"/> before advancing.</item>
+    ///     <item><c>_throwSlot</c> — an exception rethrown at the next <c>yield</c> resume (athrow/aclose).</item>
+    ///     <item><c>_cts</c> — backs the underlying enumerator's cancellation.</item>
+    ///   </list>
+    /// </remarks>
+    [PythonType("async_generator")]
+    public sealed class PythonAsyncGenerator {
+        private readonly IAsyncEnumerator<object?> _enumerator;
+        private readonly StrongBox<object?> _sendSlot;
+        private readonly StrongBox<Exception?> _throwSlot;
+        private readonly CancellationTokenSource _cts;
+        private readonly string _name;
+        private bool _started;
+        private bool _closed;
+
+        internal PythonAsyncGenerator(IAsyncEnumerable<object?> source,
+                                       StrongBox<object?> sendSlot,
+                                       StrongBox<Exception?> throwSlot,
+                                       CancellationTokenSource cts,
+                                       string name) {
+            _enumerator = source.GetAsyncEnumerator(cts.Token);
+            _sendSlot = sendSlot;
+            _throwSlot = throwSlot;
+            _cts = cts;
+            _name = name;
+        }
+
+        public PythonAsyncGenerator __aiter__() => this;
+
+
+        /// <summary>
+        ///   Advance the generator with no value sent in (equivalent to <c>asend(None)</c>).
+        ///   Returns an awaitable yielding the next produced value or raising StopAsyncIteration.
+        /// </summary>
+        public object __anext__() {
+            _sendSlot.Value = null;
+            return new AsyncEnumeratorAwaitable<object?>(_enumerator);
+        }
+
+
+        /// <summary>
+        ///   Send a value into the generator; it becomes the value of the <c>yield</c> the body is suspended
+        ///   at (<c>x = yield z</c>). The first call after creation must send None (the body hasn't reached a
+        ///   yield yet), matching CPython.
+        /// </summary>
+        public object asend(object? value) {
+            if (!_started) {
+                _started = true;
+                if (value is not null) {
+                    throw PythonOps.TypeError("can't send non-None value to a just-started async generator");
+                }
+            }
+            _sendSlot.Value = value;
+            return new AsyncEnumeratorAwaitable<object?>(_enumerator);
+        }
+
+
+        [LightThrowing]
+        public object athrow(object? type) => athrow(type, null, null);
+
+        [LightThrowing]
+        public object athrow(object? type, object? value) => athrow(type, value, null);
+
+        /// <summary>
+        ///   Throw an exception into the generator at the suspended <c>yield</c>. Returns an awaitable.
+        /// </summary>
+        /// <remarks>
+        ///   When awaited the exception is rethrown at the resume point:
+        ///   if the body catches it and yields again that value is produced;
+        ///   otherwise the exception propagates (or StopAsyncIteration if the body finishes).
+        /// </remarks>
+        [LightThrowing]
+        public object athrow(object? type, object? value, object? traceback) {
+            // Validate shape (mirrors PythonGenerator/PythonCoroutine.throw).
+            if (type is Exception || type is PythonExceptions.BaseException) {
+                if (value is not null)
+                    return LightExceptions.Throw(PythonOps.TypeError("instance exception may not have a separate value"));
+            } else if (type is PythonType pt && typeof(PythonExceptions.BaseException).IsAssignableFrom(pt.UnderlyingSystemType)) {
+                // ok — class form
+            } else {
+                return LightExceptions.Throw(PythonOps.TypeError(
+                    "exceptions must be classes or instances deriving from BaseException, not {0}",
+                    PythonOps.GetPythonTypeName(type)));
+            }
+
+            Exception ex = PythonOps.MakeExceptionForGenerator(DefaultContext.Default, type, value, traceback, cause: null);
+            _started = true;
+            _throwSlot.Value = ex;
+            _sendSlot.Value = null;
+            return new AsyncEnumeratorAwaitable<object?>(_enumerator);
+        }
+
+        /// <summary>
+        ///   Close the generator: returns an awaitable that injects GeneratorExit at the suspended <c>yield</c>
+        ///   (running the body's try/finally), then disposes the underlying async iterator.
+        /// </summary>
+        public object aclose() {
+            return AcloseAsync();
+        }
+
+        private async Task<object?> AcloseAsync() {
+            if (_closed) return null;
+            _closed = true;
+            if (_started) {
+                _throwSlot.Value = new GeneratorExitException();
+                _sendSlot.Value = null;
+                try {
+                    bool produced = await _enumerator.MoveNextAsync();
+                    if (produced) {
+                        throw PythonOps.RuntimeError("async generator ignored GeneratorExit");
+                    }
+                } catch (GeneratorExitException) {
+                    // expected — the body let GeneratorExit propagate
+                } catch (StopIterationException) {
+                    // body finished — also fine
+                }
+            }
+            await _enumerator.DisposeAsync();
+            _cts.Cancel();
+            return null;
+        }
+
+        public string __name__ => _name;
+
+        public string __qualname__ => _name;
+
+        public bool ag_running => _started && !_closed;
+    }
+}
+
+#endif
diff --git a/src/core/IronPython/Runtime/Coroutine.cs b/src/core/IronPython/Runtime/Coroutine.cs
index e7cfdf575..e5cd2adac 100644
--- a/src/core/IronPython/Runtime/Coroutine.cs
+++ b/src/core/IronPython/Runtime/Coroutine.cs
@@ -4,7 +4,9 @@
 
 #nullable enable
 
+using System;
 using System.Runtime.CompilerServices;
+using System.Threading;
 using System.Threading.Tasks;
 
 using Microsoft.Scripting.Runtime;
@@ -14,6 +16,152 @@
 using IronPython.Runtime.Types;
 
 namespace IronPython.Runtime {
+
+#if FEATURE_NET_ASYNC
+
+    [PythonType("coroutine")]
+    [DontMapIDisposableToContextManager, DontMapIEnumerableToContains]
+    public sealed class PythonCoroutine : ICodeFormattable, IWeakReferenceable {
+        // Under .NET async, `async def` is compiled directly to a Task<object?> via the DLR's AsyncExpression + AsyncHelpers.
+        // PythonCoroutine is a wrapper over the Task with two cancellation channels pre-bound at the codegen level:
+        //
+        //   _cts                   — CancellationTokenSource whose Token was passed into AsyncExpression. throw(exc) cancels this.
+        //   _cancellationException — StrongBox<Exception?> shared with DriveAsync's `cancellationException` parameter;
+        //                            when non-null at cancellation time, that exception is surfaced to the body
+        //                            instead of OperationCanceledException. 
+        //
+        // Lazy start: the body is not executed at construction. _factory is the thunk that creates (and thereby starts driving) the Task on
+        // first access; _task caches the materialized Task. Calling an async def is therefore side-effect free until the coroutine is first
+        // driven (send/AsTask/GetAwaiter) — PEP 492.
+        private readonly Func<Task<object?>> _factory;
+        private Task<object?>? _task;
+        private readonly string _name;
+        private readonly FunctionCode? _code;
+        private readonly CancellationTokenSource _cts;
+        private readonly StrongBox<Exception?> _cancellationException;
+        private WeakRefTracker? _tracker;
+
+        internal PythonCoroutine(Func<Task<object?>> factory, string name, FunctionCode? code,
+                                  CancellationTokenSource cts,
+                                  StrongBox<Exception?> cancellationException) {
+            _factory = factory;
+            _name = name;
+            _code = code;
+            _cts = cts;
+            _cancellationException = cancellationException;
+        }
+
+        private Task<object?> EnsureTask() => _task ??= _factory();
+
+
+        [LightThrowing]
+        public object send(object? value) {
+            // Fast path: task already completed. Dispatch on terminal state without
+            // any try/catch — Result on a Faulted/Canceled task throws, so we must
+            // check IsCanceled / IsFaulted before reading it.
+            //
+            // Slow path: not completed. Block via AsyncWaitHandle.WaitOne() so we do
+            // not pay for exception unwinding (GetAwaiter().GetResult() would observe
+            // and rethrow cancel/fault). After waiting, the same terminal-state
+            // dispatch below applies.
+
+            var task = EnsureTask();
+            if (!task.IsCompleted) {
+                // TODO: coroutines should be steppable via send(), one await at a time.
+                ((IAsyncResult)task).AsyncWaitHandle.WaitOne();
+            }
+
+            return SettleCompletedTask(task);
+        }
+
+
+        [LightThrowing]
+        public object @throw(object? type) => @throw(type, null, null);
+
+        [LightThrowing]
+        public object @throw(object? type, object? value) => @throw(type, value, null);
+
+        [LightThrowing]
+        public object @throw(object? type, object? value, object? traceback) {
+            // Validate the shape of (type, value, traceback) up front. Mirrors PythonGenerator.@throw
+            // so we don't mutate any state on bad input.
+            if (type is Exception || type is PythonExceptions.BaseException) {
+                if (value is not null)
+                    return LightExceptions.Throw(PythonOps.TypeError("instance exception may not have a separate value"));
+            } else if (type is PythonType pt && typeof(PythonExceptions.BaseException).IsAssignableFrom(pt.UnderlyingSystemType)) {
+                // ok — class form, MakeExceptionForGenerator will construct.
+            } else {
+                return LightExceptions.Throw(PythonOps.TypeError(
+                    "exceptions must be classes or instances deriving from BaseException, not {0}",
+                    PythonOps.GetPythonTypeName(type)));
+            }
+
+            // Construct the actual exception object. DefaultContext.Default suffices here — throw()'s
+            // CodeContext role is limited to resolving class-form exception construction, and the
+            // coroutine's own context will re-bind the chain when the exception unwinds inside the body.
+            Exception ex = PythonOps.MakeExceptionForGenerator(
+                DefaultContext.Default, type, value, traceback, cause: null);
+
+            var task = EnsureTask();
+            if (task.IsCompleted) {
+                // Case 1: coroutine has finished. throw() raises the exception immediately —
+                // the body has nothing left to observe it. Matches CPython.
+                return LightExceptions.Throw(ex);
+            }
+
+            // Case 2: body is still running (or suspended at an await). Stash the exception in the
+            // shared StrongBox so DriveAsync surfaces it (instead of OCE) the moment it observes the
+            // cancellation, then cancel. The body sees `ex` rethrown at its next resume point.
+            _cancellationException.Value = ex;
+            _cts.Cancel();
+
+            // Wait for the body to settle. After this the task is in a terminal state — dispatch via
+            // the same logic as send() so the body's reaction (catch-and-return, propagate, etc.) is
+            // reflected back to the caller of throw().
+            ((IAsyncResult)task).AsyncWaitHandle.WaitOne();
+            return SettleCompletedTask(task);
+        }
+
+
+        private static object SettleCompletedTask(Task<object?> task) {
+            if (task.IsCanceled) {
+                // Surfaces as Python CancelledError via the OCE -> CancelledError mapping in PythonExceptions.
+                return LightExceptions.Throw(new TaskCanceledException(task));
+            }
+            if (task.IsFaulted) {
+                return LightExceptions.Throw(task.Exception?.InnerException ?? task.Exception);
+            }
+            return LightExceptions.Throw(new PythonExceptions._StopIteration().InitAndGetClrException(task.Result!));
+        }
+
+
+        [LightThrowing]
+        public object? close() => null;
+
+        public object __await__() => new CoroutineWrapper(this);
+
+        public FunctionCode? cr_code => _code;
+
+        // A not-yet-driven coroutine reports 0 (not running) without materializing the Task.
+        // Merely inspecting cr_running must not start the body.
+        public int cr_running => _task is null || _task.IsCompleted ? 0 : 1;
+
+        public TraceBackFrame? cr_frame => null;
+
+        public string __name__ => _name;
+
+        public string __qualname__ => _name;
+
+        /// <summary>Returns the underlying <see cref="Task{Object}"/>, starting the body on first call.</summary>
+        public Task<object?> AsTask() => EnsureTask();
+
+        /// <summary>Enables <c>await coroutine</c> from C# code.</summary>
+        public TaskAwaiter<object?> GetAwaiter() => EnsureTask().GetAwaiter();
+
+        internal Task<object?> Task => EnsureTask();
+
+#else  // !FEATURE_NET_ASYNC
+
     [PythonType("coroutine")]
     [DontMapIDisposableToContextManager, DontMapIEnumerableToContains]
     public sealed class PythonCoroutine : ICodeFormattable, IWeakReferenceable {
@@ -99,6 +247,8 @@ public string __qualname__ {
 
         internal PythonGenerator Generator => _generator;
 
+#endif  // FEATURE_NET_ASYNC - the rest of the members are shared between both implementations of PythonCoroutine
+
         #region ICodeFormattable Members
 
         public string __repr__(CodeContext context) {
@@ -125,6 +275,7 @@ void IWeakReferenceable.SetFinalizer(WeakRefTracker value) {
         #endregion
     }
 
+
     [PythonType("coroutine_wrapper")]
     public sealed class CoroutineWrapper {
         private readonly PythonCoroutine _coroutine;
diff --git a/src/core/IronPython/Runtime/FunctionCode.cs b/src/core/IronPython/Runtime/FunctionCode.cs
index a790be7e9..f46e8b1b8 100644
--- a/src/core/IronPython/Runtime/FunctionCode.cs
+++ b/src/core/IronPython/Runtime/FunctionCode.cs
@@ -750,7 +750,15 @@ private LambdaExpression GetGeneratorOrNormalLambdaTracing(PythonContext context
                 debugProperties // custom payload
             );
 
+#if FEATURE_NET_ASYNC
+            // Under .NET-async, async bodies are lowered before they reach here and must NOT go through the generator rewriter:
+            //  * plain async (Coroutine, no Generator) via AsyncExpression,
+            //  * async generators (Generator+Coroutine) via AsyncEnumerableExpression.
+            //  Only a plain generator (Generator without Coroutine) is rewritten here.
+            if ((Flags & FunctionAttributes.Generator) == 0 || (Flags & FunctionAttributes.Coroutine) != 0) {
+#else
             if ((Flags & (FunctionAttributes.Generator | FunctionAttributes.Coroutine)) == 0) {
+#endif
                 return context.DebugContext.TransformLambda((LambdaExpression)Compiler.Ast.Node.RemoveFrame(_lambda.GetLambda()), debugInfo);
             }
 
@@ -781,7 +789,13 @@ private LambdaExpression GetGeneratorOrNormalLambdaTracing(PythonContext context
         /// </summary>
         private LightLambdaExpression GetGeneratorOrNormalLambda() {
             LightLambdaExpression finalCode;
+#if FEATURE_NET_ASYNC
+            // See GetGeneratorOrNormalLambda's sibling in MakeDebuggableFunctionCode: only a plain generator
+            // (Generator without Coroutine) is rewritten; plain async and async generators are lowered earlier.
+            if ((Flags & FunctionAttributes.Generator) == 0 || (Flags & FunctionAttributes.Coroutine) != 0) {
+#else
             if ((Flags & (FunctionAttributes.Generator | FunctionAttributes.Coroutine)) == 0) {
+#endif
                 finalCode = Code;
             } else {
                 bool isCoroutine = (Flags & FunctionAttributes.Coroutine) != 0;
diff --git a/src/core/IronPython/Runtime/Operations/InstanceOps.cs b/src/core/IronPython/Runtime/Operations/InstanceOps.cs
index 5c7edca0b..82c479e30 100644
--- a/src/core/IronPython/Runtime/Operations/InstanceOps.cs
+++ b/src/core/IronPython/Runtime/Operations/InstanceOps.cs
@@ -242,6 +242,27 @@ public static object ValueTaskAwaitMethodGeneric<T>(System.Threading.Tasks.Value
         public static object AsyncIterMethod<T>(System.Collections.Generic.IAsyncEnumerable<T> self) {
             return new AsyncEnumeratorWrapper<T>(self.GetAsyncEnumerator());
         }
+
+        /// <summary>
+        /// Provides the implementation of __aenter__ for objects implementing <see cref="System.IAsyncDisposable"/>.
+        /// IAsyncDisposable has no async-enter step — entering just yields the resource itself, wrapped in an
+        /// already-completed Task so the desugared <c>await mgr.__aenter__()</c> resolves to <c>self</c>.
+        /// </summary>
+        public static object AEnterMethod(System.IAsyncDisposable self) {
+            // Task<object> is the same runtime type as Task<object?>, so AsTaskForAwait's
+            // Task<object?> fast path picks it up. (InstanceOps isn't a #nullable context.)
+            return System.Threading.Tasks.Task.FromResult<object>(self);
+        }
+
+        /// <summary>
+        /// Provides the implementation of __aexit__ for objects implementing <see cref="System.IAsyncDisposable"/>.
+        /// Calls <see cref="System.IAsyncDisposable.DisposeAsync"/>; the returned ValueTask is awaited by the
+        /// desugared <c>await mgr.__aexit__(...)</c>. Returns a falsy result (the ValueTask completes with no
+        /// value) so the exception, if any, is not suppressed.
+        /// </summary>
+        public static object AExitMethod(System.IAsyncDisposable self, object exc_type, object exc_value, object exc_back) {
+            return self.DisposeAsync().AsTask();
+        }
 #endif
 
         #endregion
diff --git a/src/core/IronPython/Runtime/Operations/PythonOps.cs b/src/core/IronPython/Runtime/Operations/PythonOps.cs
index b3280ab6e..ee7850c79 100644
--- a/src/core/IronPython/Runtime/Operations/PythonOps.cs
+++ b/src/core/IronPython/Runtime/Operations/PythonOps.cs
@@ -22,6 +22,7 @@
 using System.Runtime.InteropServices;
 using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
 
 using Microsoft.Scripting;
 using Microsoft.Scripting.Actions;
@@ -3196,14 +3197,153 @@ public static PythonGenerator MakeGenerator(PythonFunction function, MutableTupl
             return new PythonGenerator(function, next, data);
         }
 
+
         public static PythonCoroutine MakeCoroutine(PythonFunction function, MutableTuple data, object generatorCode) {
+#if FEATURE_NET_ASYNC
+            throw new System.InvalidOperationException("MakeCoroutine is unused under FEATURE_NET_ASYNC");
+#else
             return new PythonCoroutine(MakeGenerator(function, data, generatorCode));
+#endif
         }
 
+
         public static object MakeCoroutineWrapper(PythonGenerator generator) {
+#if FEATURE_NET_ASYNC
+            throw new System.InvalidOperationException("MakeCoroutineWrapper is unused under FEATURE_NET_ASYNC");
+#else
             return new PythonCoroutine(generator);
+#endif
+        }
+
+
+#if FEATURE_NET_ASYNC
+        /// <summary>
+        /// Converts an arbitrary value yielded into an <c>await</c> expression to a <see cref="Task{T}"/> of object
+        /// that the DLR async runner can drive.
+        /// </summary>
+        public static Task<object?> AsTaskForAwait(object? value) {
+
+            Task<object?>? task = value switch {
+                null                    => Task.FromResult<object?>(null),
+                PythonCoroutine coro    => coro.AsTask(),
+                Task<object?> to        => to,
+                Task t                  => BoxTaskResult(t),
+#if NETCOREAPP
+                // ValueTask / ValueTask<T>: convert to a regular Task via AsTask() and box Result if needed
+                ValueTask<object?> vto  => vto.AsTask(),
+                ValueTask vt            => BoxTaskResult(vt.AsTask()),
+#endif
+                _                       => null,
+            };
+            if (task is not null) return task;
+
+#if NETCOREAPP
+            // ValueTask<T> of arbitrary T
+            Type valueType = value!.GetType();
+            if (valueType.IsGenericType && valueType.GetGenericTypeDefinition() == typeof(ValueTask<>)) {
+                var asTask = (Task)valueType.GetMethod("AsTask")!.Invoke(value, null)!;
+                return BoxTaskResult(asTask);
+            }
+#endif
+
+            // PEP 492 awaitable protocol fallback: any object exposing __await__ that returns an iterator
+            // (user-defined awaitables, and the AsyncEnumeratorAwaitable produced by `async for` over IAsyncEnumerable<T>).
+            // The native Task/coroutine/ValueTask fast paths above mean a plain `await Task` never reaches here.
+            if (TryGetBoundAttr(value, "__await__", out object? awaitMethod)) {
+                object? iterator = PythonCalls.Call(DefaultContext.Default, awaitMethod);
+                return DriveAwaitIterator(iterator);
+            }
+
+            throw TypeError("object of type '{0}' can't be used in 'await' expression", PythonOps.GetPythonTypeName(value));
+        }
+
+
+        /// <summary>
+        ///   Drives a Python awaitable's <c>__await__()</c> iterator to completion,
+        ///   returning a Task{object?} that yields the awaitable's result.
+        /// </summary>
+        /// <remarks>
+        ///   The iterator follows the generator/yield-from protocol: each step yields an awaitable (typically a Task) and the chain
+        ///   terminates with <c>StopIteration</c> whose value is the <c>await</c> result. Mirrors the generator-pump <c>AsTask</c>,
+        ///   but is async and at the Python-iterator level.
+        /// </remarks>
+        private static async Task<object?> DriveAwaitIterator(object? iterator) {
+            CodeContext context = DefaultContext.Default;
+            // Generators feed the awaited result back via send();
+            // plain iterators (e.g. iter([])) only support __next__() and ignore sent values.
+            bool hasSend = TryGetBoundAttr(context, iterator, "send", out object? sendMethod);
+            object? sendValue = null;
+            while (true) {
+                object? yielded;
+                try {
+                    yielded = hasSend
+                        ? PythonCalls.Call(context, sendMethod, sendValue)
+                        : Invoke(context, iterator, "__next__");
+                } catch (StopIterationException e) {
+                    return StopIterationValue(e);
+                }
+                // A [LightThrowing] __next__/send may return a light-exception sentinel rather than throwing — handle both forms.
+                if (LightExceptions.IsLightException(yielded)) {
+                    var clrExc = LightExceptions.GetLightException(yielded!);
+                    if (clrExc is StopIterationException si) return StopIterationValue(si);
+                    throw clrExc;
+                }
+                // The yielded value is itself awaitable — drive it and feed the result back in.
+                sendValue = await AsTaskForAwait(yielded);
+            }
+
+            static object? StopIterationValue(StopIterationException e)
+                => ((IPythonAwareException)e).PythonException is PythonExceptions._StopIteration si ? si.value : null;
+        }
+
+
+        /// <summary>
+        ///   Converts any Task/Task{T} to a Task{object?}, boxing the Result if needed, and propagating cancellation and exceptions.
+        /// </summary>
+        private static Task<object?> BoxTaskResult(Task task) {
+            Debug.Assert(task is not Task<object?>, "BoxTaskResult is suboptimal for Task<object?>");
+            Type tp = task.GetType();
+            var tcs = new TaskCompletionSource<object?>();
+            task.ContinueWith(t => {
+                if (t.IsCanceled)  tcs.SetCanceled();
+                else if (t.IsFaulted) tcs.SetException(t.Exception!.InnerException ?? t.Exception);
+                else tcs.SetResult(!tp.IsGenericType ? null : (object?)tp.GetProperty("Result")!.GetValue(t));
+            }, TaskContinuationOptions.ExecuteSynchronously);
+            return tcs.Task;
+        }
+
+        /// <summary>
+        ///   Wraps an sync function's deferred <see cref="Task"/> thunk in a <see cref="PythonCoroutine"/>
+        ///   for the caller of <c>async def</c>.
+        /// </summary>
+        /// <remarks>
+        ///   The thunk is invoked lazily on first drive (send/AsTask), so calling an <c>async def</c> is side-effect-free until the coroutine
+        ///   is awaited (PEP 492).
+        ///   The supplied <see cref="CancellationTokenSource"/> and exception-override box are pre-bound
+        ///   to <see cref="Microsoft.Scripting.Runtime.AsyncHelpers.DriveAsync"/>'s cancellation channel — the coroutine uses them to implement
+        ///   <c>coro.throw(exc)</c> on a running body (set the box, cancel the CTS).
+        /// </remarks>
+        public static PythonCoroutine MakeAsyncCoroutine(PythonFunction function, Func<Task<object?>> taskFactory, CancellationTokenSource cts, StrongBox<System.Exception?> cancellationException) {
+            return new PythonCoroutine(taskFactory, function.__name__, function.__code__, cts, cancellationException);
         }
 
+
+        /// <summary>
+        ///   Wraps an async-generator's <see cref="IAsyncEnumerable{T}"/> (from
+        ///   AsyncEnumerableExpression / DriveAsyncEnumerable) in a <see cref="PythonAsyncGenerator"/> for the
+        ///   caller of an <c>async def</c> with <c>yield</c>.
+        /// </summary>
+        /// <param name="sendSlot">
+        ///   carries the value of yield to deliver: <c>x = yield z</c> (asend).
+        /// </param>
+        /// <param name="throwSlot">
+        ///   carries an exception to rethrow at the yield resume (athrow/aclose).
+        /// </param>
+        public static PythonAsyncGenerator MakeAsyncGenerator(PythonFunction function, IAsyncEnumerable<object?> source, StrongBox<object?> sendSlot, StrongBox<System.Exception?> throwSlot, CancellationTokenSource cts) {
+            return new PythonAsyncGenerator(source, sendSlot, throwSlot, cts, function.__name__);
+        }
+#endif
+
         public static object MakeGeneratorExpression(object function, object input) {
             PythonFunction func = (PythonFunction)function;
             return ((Func<PythonFunction, object, object>)func.__code__.Target)(func, input);
diff --git a/src/core/IronPython/Runtime/Types/PythonTypeInfo.cs b/src/core/IronPython/Runtime/Types/PythonTypeInfo.cs
index 515e81402..ab908de82 100644
--- a/src/core/IronPython/Runtime/Types/PythonTypeInfo.cs
+++ b/src/core/IronPython/Runtime/Types/PythonTypeInfo.cs
@@ -676,6 +676,8 @@ private class ProtectedMemberResolver : MemberResolver {
                 new OneOffResolver("__await__", AwaitResolver),
                 new OneOffResolver("__aiter__", AsyncIterResolver),
                 new OneOffResolver("__anext__", AsyncNextResolver),
+                new OneOffResolver("__aenter__", AsyncEnterResolver),
+                new OneOffResolver("__aexit__", AsyncExitResolver),
 
                 new OneOffResolver("__complex__", ComplexResolver),
                 new OneOffResolver("__float__", FloatResolver),
@@ -1052,6 +1054,42 @@ internal static MemberGroup GetExtensionMemberGroup(Type type, MemberInfo[] news
             return MemberGroup.EmptyGroup;
         }
 
+        /// <summary>
+        /// Provides a resolution for __aenter__ on types implementing <c>IAsyncDisposable</c>,
+        /// so that <c>async with</c> works seamlessly over .NET async-disposable resources.
+        /// </summary>
+        private static MemberGroup/*!*/ AsyncEnterResolver(MemberBinder/*!*/ binder, Type/*!*/ type) {
+#if NET
+            foreach (Type t in binder.GetContributingTypes(type)) {
+                if (t.GetMember("__aenter__").Length > 0) {
+                    return MemberGroup.EmptyGroup;
+                }
+            }
+            if (typeof(IAsyncDisposable).IsAssignableFrom(type)) {
+                return GetInstanceOpsMethod(type, nameof(InstanceOps.AEnterMethod));
+            }
+#endif
+            return MemberGroup.EmptyGroup;
+        }
+
+        /// <summary>
+        /// Provides a resolution for __aexit__ on types implementing <c>IAsyncDisposable</c>.
+        /// Maps to <c>InstanceOps.AExitMethod</c>, which awaits DisposeAsync().
+        /// </summary>
+        private static MemberGroup/*!*/ AsyncExitResolver(MemberBinder/*!*/ binder, Type/*!*/ type) {
+#if NET
+            foreach (Type t in binder.GetContributingTypes(type)) {
+                if (t.GetMember("__aexit__").Length > 0) {
+                    return MemberGroup.EmptyGroup;
+                }
+            }
+            if (typeof(IAsyncDisposable).IsAssignableFrom(type)) {
+                return GetInstanceOpsMethod(type, nameof(InstanceOps.AExitMethod));
+            }
+#endif
+            return MemberGroup.EmptyGroup;
+        }
+
         /// <summary>
         /// Provides a resolution for __len__
         /// </summary>
diff --git a/tests/IronPython.Tests/Cases/IronPythonCasesManifest.ini b/tests/IronPython.Tests/Cases/IronPythonCasesManifest.ini
index 07fe270f8..ecf47c9ca 100644
--- a/tests/IronPython.Tests/Cases/IronPythonCasesManifest.ini
+++ b/tests/IronPython.Tests/Cases/IronPythonCasesManifest.ini
@@ -4,6 +4,11 @@ WorkingDirectory=$(TEST_FILE_DIR)
 Redirect=false
 Timeout=120000 # 2 minute timeout
 
+[IronPython.test_asyncgen]
+Ignore=true
+RunCondition='$(FRAMEWORK)' <> '.NETFramework,Version=v4.6.2'
+Reason=Requires IAsyncEnumerable & IAsyncDisposable
+
 [IronPython.test_builtin_stdlib]
 RunCondition=NOT $(IS_MONO)
 Reason=Exception on adding DocTestSuite
diff --git a/tests/suite/test_async.py b/tests/suite/test_async.py
index b6c24bd09..22d4f5c43 100644
--- a/tests/suite/test_async.py
+++ b/tests/suite/test_async.py
@@ -94,6 +94,24 @@ async def foo():
             coro.throw(ValueError('boom'))
         self.assertEqual(str(cm.exception), 'boom')
 
+    def test_lazy_evaluation(self):
+        """The body of an async def must not run until the coroutine is first driven."""
+        log = []
+
+        async def noop():
+            return
+
+        async def lazy():
+            log.append('before')
+            await noop()
+            log.append('after')
+            return 'done'
+
+        coro = lazy()
+        self.assertEqual(log, [])  # constructing the coroutine runs nothing
+        self.assertEqual(run_coro(coro), 'done')
+        self.assertEqual(log, ['before', 'after'])
+
 
 class AwaitTest(unittest.TestCase):
     """Tests for await expression."""
@@ -139,6 +157,20 @@ async def test():
 
         self.assertEqual(run_coro(test()), 'done')
 
+    def test_exception_across_await(self):
+        """An exception raised in an awaited coroutine propagates and is catchable."""
+        async def fails():
+            raise ValueError('boom')
+
+        async def test():
+            try:
+                await fails()
+            except ValueError as e:
+                return 'caught: ' + str(e)
+            return 'not caught'
+
+        self.assertEqual(run_coro(test()), 'caught: boom')
+
 
 class AsyncWithTest(unittest.TestCase):
     """Tests for async with statement."""
diff --git a/tests/suite/test_asyncgen.py b/tests/suite/test_asyncgen.py
new file mode 100644
index 000000000..ed3cd8048
--- /dev/null
+++ b/tests/suite/test_asyncgen.py
@@ -0,0 +1,257 @@
+# Licensed to the .NET Foundation under one or more agreements.
+# The .NET Foundation licenses this file to you under the Apache 2.0 License.
+# See the LICENSE file in the project root for more information.
+
+"""Tests for async constructs that lower to .NET IAsyncEnumerable / IAsyncDisposable:
+async generators (PEP 525) and `async with` over a .NET IAsyncDisposable."""
+
+import unittest
+
+from iptest import run_test, skipUnlessIronPython
+
+
+def run_coro(coro):
+    """Run a coroutine to completion, blocking on yielded .NET Tasks."""
+    value = None
+    while True:
+        try:
+            task = coro.send(value)
+            # .NET Task yielded — block on it (test runner is synchronous)
+            task.Wait()
+            value = None
+        except StopIteration as e:
+            return e.value
+
+
+@skipUnlessIronPython()
+class DotNetAsyncDisposableTest(unittest.TestCase):
+    """Tests for async with over a .NET IAsyncDisposable."""
+
+    def test_async_with_dotnet_disposable(self):
+        """async with over a .NET IAsyncDisposable enters with the resource and disposes on exit."""
+        from System.IO import MemoryStream
+        s = MemoryStream()
+        self.assertTrue(hasattr(s, '__aenter__'))
+        self.assertTrue(hasattr(s, '__aexit__'))
+
+        async def test():
+            async with s as entered:
+                self.assertIs(entered, s)
+                self.assertTrue(entered.CanWrite)
+            return s.CanWrite  # disposed on exit -> False
+
+        self.assertFalse(run_coro(test()))
+
+
+@skipUnlessIronPython()
+class AsyncGeneratorTest(unittest.TestCase):
+    """Tests for PEP 525 async generators (await + yield). Bodies await .NET Tasks."""
+
+
+    def test_await_and_yield(self):
+        """An async generator body mixing await and yield, consumed via async for."""
+        from System.Threading.Tasks import Task
+
+        async def agen():
+            yield 1
+            x = await Task.FromResult(10)
+            yield x + 1       # await result feeds the next yield
+            if x > 5:
+                return        # bare return ends the async iteration
+            yield 'unreached'
+
+        async def consume():
+            out = []
+            async for v in agen():
+                out.append(v)
+            return out
+
+        self.assertEqual(run_coro(consume()), [1, 11])
+
+
+    def test_yield_task_is_value_not_await(self):
+        """A yielded Task is the produced value, not a suspension point."""
+        from System.Threading.Tasks import Task
+
+        async def agen():
+            yield Task.FromResult(99)
+
+        async def consume():
+            out = []
+            async for item in agen():
+                out.append(item)
+            return out
+
+        items = run_coro(consume())
+        self.assertEqual(len(items), 1)
+        # The item is the Task itself; if it had been awaited we'd have gotten 99.
+        self.assertEqual(items[0].Result, 99)
+
+
+    def test_async_generator_type(self):
+        async def agen():
+            yield 1
+        g = agen()
+        self.assertEqual(type(g).__name__, 'async_generator')
+
+
+    def test_asend(self):
+        """asend feeds a value into `x = yield z` (PEP 525 two-way communication)."""
+        async def agen():
+            got1 = yield 1
+            got2 = yield ('after1', got1)
+            yield ('after2', got2)
+
+        async def drive():
+            g = agen()
+            out = []
+            out.append(await g.asend(None))    # start -> yields 1
+            out.append(await g.asend('A'))     # got1 == 'A' -> ('after1', 'A')
+            out.append(await g.asend('B'))     # got2 == 'B' -> ('after2', 'B')
+            try:
+                await g.asend('C')
+            except StopAsyncIteration:
+                out.append('stop')
+            return out
+
+        self.assertEqual(run_coro(drive()), [1, ('after1', 'A'), ('after2', 'B'), 'stop'])
+
+
+    def test_athrow(self):
+        """athrow injects an exception at the yield; the body can catch and continue."""
+        async def agen():
+            try:
+                yield 1
+            except ValueError as e:
+                yield ('caught', str(e))
+            yield 3
+
+        async def drive():
+            g = agen()
+            out = []
+            out.append(await g.asend(None))                 # -> 1
+            out.append(await g.athrow(ValueError('boom')))  # caught -> ('caught', 'boom')
+            out.append(await g.asend(None))                 # -> 3
+            try:
+                await g.asend(None)
+            except StopAsyncIteration:
+                out.append('stop')
+            return out
+
+        self.assertEqual(run_coro(drive()), [1, ('caught', 'boom'), 3, 'stop'])
+
+
+    def test_aclose_runs_finally(self):
+        """aclose injects GeneratorExit at the yield so the body's finally runs."""
+        log = []
+
+        async def agen():
+            try:
+                yield 1
+                yield 2
+            finally:
+                log.append('cleanup')
+
+        async def drive():
+            g = agen()
+            first = await g.asend(None)   # -> 1
+            await g.aclose()              # GeneratorExit at the yield -> finally runs
+            return first
+
+        self.assertEqual(run_coro(drive()), 1)
+        self.assertEqual(log, ['cleanup'])
+
+
+@skipUnlessIronPython()
+class NestedAsyncGeneratorTest(unittest.TestCase):
+    """An async generator lexically nested inside another async generator.
+
+    Both functions own the async send/throw slots; their slot variables must stay
+    distinct per function so the inner generator's slots don't collide with the
+    outer's when the inner lambda is nested inside the outer.
+    """
+
+
+    def test_nested_async_generator(self):
+        """Outer async gen consumes a nested async gen via async for and re-yields."""
+        async def outer():
+            async def inner():
+                yield 'a'
+                yield 'b'
+            async for x in inner():
+                yield x.upper()
+
+        async def consume():
+            out = []
+            async for v in outer():
+                out.append(v)
+            return out
+
+        self.assertEqual(run_coro(consume()), ['A', 'B'])
+
+
+    def test_nested_async_generator_with_await(self):
+        """Both nested async gens await a .NET Task between yields."""
+        from System.Threading.Tasks import Task
+
+        async def outer():
+            async def inner():
+                yield 1
+                n = await Task.FromResult(10)
+                yield n + 1
+            async for v in inner():
+                doubled = await Task.FromResult(v * 2)
+                yield doubled
+
+        async def consume():
+            out = []
+            async for v in outer():
+                out.append(v)
+            return out
+
+        self.assertEqual(run_coro(consume()), [2, 22])
+
+
+    def test_nested_inner_asend(self):
+        """Inner nested async gen driven with asend (sendSlot); outer re-yields the pairs."""
+        async def outer():
+            async def inner():
+                a = yield 1
+                b = yield ('a', a)
+                yield ('b', b)
+            g = inner()
+            yield await g.asend(None)    # -> 1
+            yield await g.asend('X')     # a == 'X' -> ('a', 'X')
+            yield await g.asend('Y')     # b == 'Y' -> ('b', 'Y')
+
+        async def consume():
+            out = []
+            async for v in outer():
+                out.append(v)
+            return out
+
+        self.assertEqual(run_coro(consume()), [1, ('a', 'X'), ('b', 'Y')])
+
+
+    def test_nested_inner_athrow(self):
+        """athrow into the inner nested async gen (throwSlot); outer re-yields the recovery."""
+        async def outer():
+            async def inner():
+                try:
+                    yield 1
+                except ValueError as e:
+                    yield ('caught', str(e))
+            g = inner()
+            yield await g.asend(None)                 # -> 1
+            yield await g.athrow(ValueError('boom'))  # inner catches -> ('caught', 'boom')
+
+        async def consume():
+            out = []
+            async for v in outer():
+                out.append(v)
+            return out
+
+        self.assertEqual(run_coro(consume()), [1, ('caught', 'boom')])
+
+
+run_test(__name__)