.net changes to re-enable xlang support, add subscription apis (#4159)

* add subscription response

* fix send subscription response

* add register agent type response

* adding a test

* working on shaping up a test

* appsettins update for backend

* another appsettings

* fixup aspire hosting

* enable AGENT_HOST var from aspire

* add SendMessageAsync

* remove broken test

* test compiles and runs but is not (yet) correct

* subscriptions grain wireup.

* temp assert true.

* remove DI for SubscriptionGrain

* add xlang python code

* add subscription response

* rebond

* Update to .NET 9.0

* Fix Backend project SDK

* Package updates

* get RegisterAgentTypeRequest working

* fix exceptions

* add error handling for requests

* whoops

* send cloud event message type

* processing cloudevents

* trying tosend proto data - doesn't work

* trying to pack proto_data

* fix (#4238)

* pack the Message from agents_events

* format - not sure why these?

* format

* cleanup, error handling, xlang sample publishes messages that can be heard by .NET and vice versa

* format

* sdk version

* sdk vers

* net8

* back to net8

* remove netstandard2

* fix used

* remove unused

* more cleanup

* remove unneeded package

* I'm terrible at writing tests

* deserialize the cloud events and sent them as events

* comment

* cleanup

* await

* Delete dotnet/samples/Hello/Backend/Backend.csproj

unneeded change

* whoops

* merge main python back into here

* revert back to local

* revert some of the helloAgents changes.

* [.NET] Add happy path test for in-memory agent && Simplify HelloAgent example && some clean-up in extension APIs (#4227)

* add happy path test

* remove unnecessary namespace

* fix build error

* Update AgentBaseTests.cs

* revert changes

---------

* fix busted merge from main

* addressing review comments

* make internal

* case sensitive rename step 1

* case sensitive rename step 2

* remove!

---------

Co-authored-by: Peter Chang <petchang@microsoft.com>
Co-authored-by: Reuben Bond <reuben.bond@gmail.com>
Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com>
Co-authored-by: Xiaoyun Zhang <bigmiao.zhang@gmail.com>

dotnet/Directory.Packages.props

@@ -4,26 +4,25 @@
     <MicrosoftSemanticKernelVersion>1.22.0</MicrosoftSemanticKernelVersion>
     <MicrosoftSemanticKernelExperimentalVersion>1.22.0-alpha</MicrosoftSemanticKernelExperimentalVersion>
     <MicrosoftExtensionsAIVersion>9.0.0-preview.9.24525.1</MicrosoftExtensionsAIVersion>
-
     <NuGetAuditMode>direct</NuGetAuditMode>
   </PropertyGroup>
   <ItemGroup>
-    <PackageVersion Include="Aspire.Hosting" Version="9.0.0-rc.1.24511.1" />
+    <PackageVersion Include="Aspire.Hosting" Version="9.0.0" />
     <PackageVersion Include="AspNetCore.Authentication.ApiKey" Version="8.0.1" />
     <PackageVersion Include="Aspire.Azure.AI.OpenAI" Version="8.0.1-preview.8.24267.1" />
-    <PackageVersion Include="Aspire.Hosting.AppHost" Version="8.2.1" />
+    <PackageVersion Include="Aspire.Hosting.AppHost" Version="9.0.0" />
-    <PackageVersion Include="Aspire.Hosting.Azure.ApplicationInsights" Version="8.2.1" />
+    <PackageVersion Include="Aspire.Hosting.Azure.ApplicationInsights" Version="9.0.0" />
-    <PackageVersion Include="Aspire.Hosting.Azure.CognitiveServices" Version="8.2.1" />
+    <PackageVersion Include="Aspire.Hosting.Azure.CognitiveServices" Version="9.0.0" />
     <PackageVersion Include="Aspire.Hosting.NodeJs" Version="8.2.0" />
-    <PackageVersion Include="Aspire.Hosting.Orleans" Version="8.2.1" />
+    <PackageVersion Include="Aspire.Hosting.Orleans" Version="9.0.0" />
-    <PackageVersion Include="Aspire.Hosting.Qdrant" Version="8.2.1" />
+    <PackageVersion Include="Aspire.Hosting.Qdrant" Version="9.0.0" />
     <PackageVersion Include="Aspire.Hosting.Redis" Version="8.2.0" />
-    <PackageVersion Include="Azure.AI.OpenAI" Version=" 2.1.0-beta.1" />
+    <PackageVersion Include="Azure.AI.OpenAI" Version="2.1.0-beta.2" />
     <PackageVersion Include="Azure.AI.Inference" Version="1.0.0-beta.1" />
     <PackageVersion Include="Azure.Data.Tables" Version="12.9.1" />
-    <PackageVersion Include="Azure.Identity" Version="1.13.0" />
+    <PackageVersion Include="Azure.Identity" Version="1.13.1" />
     <PackageVersion Include="Azure.ResourceManager.ContainerInstance" Version="1.2.1" />
-    <PackageVersion Include="Azure.Storage.Files.Shares" Version="12.20.1" />
+    <PackageVersion Include="Azure.Storage.Files.Shares" Version="12.21.0" />
     <PackageVersion Include="CloudNative.CloudEvents.SystemTextJson" Version="2.7.1" />
     <PackageVersion Include="Elsa" Version="3.1.3" />
     <PackageVersion Include="Elsa.EntityFrameworkCore" Version="3.1.3" />
@@ -43,25 +42,25 @@
     <PackageVersion Include="Microsoft.AspNetCore.App" Version="8.0.4" />
     <PackageVersion Include="Microsoft.AspNetCore.OpenApi" Version="8.0.8" />
     <PackageVersion Include="Microsoft.ApplicationInsights.AspNetCore" Version="2.22.0" />
-    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="9.0.0-rc.2.24473.5" />
+    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Extensions.AI" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.AzureAIInference" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.Ollama" Version="$(MicrosoftExtensionsAIVersion)" />
     <PackageVersion Include="Microsoft.Extensions.AI.OpenAI" Version="$(MicrosoftExtensionsAIVersion)" />
-    <PackageVersion Include="Microsoft.Extensions.Azure" Version="1.7.6" />
+    <PackageVersion Include="Microsoft.Extensions.Azure" Version="1.8.0" />
     <PackageVersion Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.FileExtensions" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.UserSecrets" Version="8.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
+    <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.2" />
+    <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
+    <PackageVersion Include="Microsoft.Extensions.Hosting" Version="9.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="8.10.0" />
+    <PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Debug" Version="8.0.0" />
-    <PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="8.2.1" />
+    <PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="9.0.0" />
     <PackageVersion Include="Microsoft.Orleans.Clustering.Cosmos" Version="8.2.0" />
     <PackageVersion Include="Microsoft.Orleans.CodeGenerator" Version="8.2.0">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
@@ -78,17 +77,17 @@
     <PackageVersion Include="Microsoft.Orleans.Server" Version="8.2.0" />
     <PackageVersion Include="Microsoft.Orleans.Streaming" Version="8.2.0" />
     <PackageVersion Include="Microsoft.Orleans.Streaming.EventHubs" Version="8.2.0" />
-    <PackageVersion Include="Microsoft.SemanticKernel" Version="$(MicrosoftSemanticKernelVersion)" />
+    <PackageVersion Include="Microsoft.SemanticKernel" Version="1.29.0" />
     <PackageVersion Include="Microsoft.SemanticKernel.Agents.Core" Version="$(MicrosoftSemanticKernelExperimentalVersion)" />
-    <PackageVersion Include="Microsoft.SemanticKernel.Connectors.AzureOpenAI" Version="$(MicrosoftSemanticKernelVersion)" />
+    <PackageVersion Include="Microsoft.SemanticKernel.Connectors.AzureOpenAI" Version="1.29.0" />
     <PackageVersion Include="Microsoft.SemanticKernel.Connectors.Qdrant" Version="$(MicrosoftSemanticKernelExperimentalVersion)" />
     <PackageVersion Include="Microsoft.SemanticKernel.Plugins.Memory" Version="$(MicrosoftSemanticKernelExperimentalVersion)" />
     <PackageVersion Include="Microsoft.SemanticKernel.Plugins.Web" Version="$(MicrosoftSemanticKernelExperimentalVersion)" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageVersion Include="Octokit" Version="13.0.1" />
     <PackageVersion Include="Octokit.Webhooks.AspNetCore" Version="2.4.0" />
-    <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.9.0" />
+    <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.10.0" />
-    <PackageVersion Include="OpenTelemetry.Extensions.Hosting" Version="1.9.0" />
+    <PackageVersion Include="OpenTelemetry.Extensions.Hosting" Version="1.10.0" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.9.0" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.Http" Version="1.9.0" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.Runtime" Version="1.9.0" />
@@ -96,28 +95,29 @@
     <PackageVersion Include="PdfPig" Version="0.1.9-alpha-20240324-e7896" />
     <PackageVersion Include="Swashbuckle.AspNetCore" Version="6.9.0" />
     <PackageVersion Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />
-    <PackageVersion Include="System.Data.SqlClient" Version="4.8.6" />
+    <PackageVersion Include="System.Data.SqlClient" Version="4.9.0" />
-    <PackageVersion Include="System.Formats.Asn1" Version="8.0.1" />
+    <PackageVersion Include="System.Formats.Asn1" Version="9.0.0" />
     <PackageVersion Include="System.IdentityModel.Tokens.Jwt" Version="7.7.1" />
-    <PackageVersion Include="System.IO.Packaging" Version="8.0.1" />
+    <PackageVersion Include="System.IO.Packaging" Version="9.0.0" />
-    <PackageVersion Include="System.Memory.Data" Version="8.0.1" />
+    <PackageVersion Include="System.Memory.Data" Version="9.0.0" />
     <PackageVersion Include="JsonSchema.Net.Generation" Version="4.5.1" />
     <PackageVersion Include="Microsoft.DotNet.Interactive" Version="1.0.0-beta.24229.4" />
     <PackageVersion Include="Microsoft.DotNet.Interactive.Jupyter" Version="1.0.0-beta.24229.4" />
     <PackageVersion Include="Microsoft.DotNet.Interactive.PackageManagement" Version="1.0.0-beta.24229.4" />
-    <PackageVersion Include="Google.Cloud.AIPlatform.V1" Version="3.9.0" />
+    <PackageVersion Include="Google.Cloud.AIPlatform.V1" Version="3.10.0" />
     <PackageVersion Include="OpenAI" Version="2.1.0-beta.1" />
-    <PackageVersion Include="System.CodeDom" Version="5.0.0" />
+    <PackageVersion Include="System.CodeDom" Version="9.0.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="4.11.0" />
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="8.0.0" />
     <PackageVersion Include="ApprovalTests" Version="6.0.0" />
-    <PackageVersion Include="FluentAssertions" Version="6.12.1" />
+    <PackageVersion Include="FluentAssertions" Version="6.12.2" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
+    <PackageVersion Include="System.Text.Json" Version="9.0.0" />
     <PackageVersion Include="xunit" Version="2.9.2" />
     <PackageVersion Include="xunit.runner.console" Version="2.9.2" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
     <PackageVersion Include="Moq" Version="4.20.72" />
     <PackageVersion Include="Microsoft.PowerShell.SDK" Version="7.4.5" />
-    <PackageVersion Include="Microsoft.AspNetCore.TestHost" Version="8.0.10" />
+    <PackageVersion Include="Microsoft.AspNetCore.TestHost" Version="8.0.0" />
   </ItemGroup>
 </Project>

dotnet/global.json

@@ -1,6 +1,6 @@
 {
   "sdk": {
-    "version": "8.0.104",
+    "version": "8.0.401",
     "rollForward": "latestMinor"
   }
 }

dotnet/samples/Hello/Backend/Backend.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk.Web">
   <ItemGroup>
     <ProjectReference Include="..\..\..\src\Microsoft.AutoGen\Agents\Microsoft.AutoGen.Agents.csproj" />
   </ItemGroup>

dotnet/samples/Hello/Backend/Program.cs

@@ -1,7 +1,5 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Program.cs
 
-using Microsoft.Extensions.Hosting;
-
 var app = await Microsoft.AutoGen.Agents.Host.StartAsync(local: false, useGrpc: true);
 await app.WaitForShutdownAsync();

dotnet/samples/Hello/Backend/appsettings.json

@@ -5,5 +5,11 @@
       "Microsoft": "Warning",
       "Microsoft.Orleans": "Warning"
     }
+  },
+  "AllowedHosts": "*",
+  "Kestrel": {
+    "EndpointDefaults": {
+      "Protocols": "Http2"
+    }
   }
 }

dotnet/samples/Hello/Hello.AppHost/Hello.AppHost.csproj

@@ -1,4 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
+
+  <Sdk Name="Aspire.AppHost.Sdk" Version="9.0.0" />
+
   <PropertyGroup>
     <OutputType>Exe</OutputType>
         <TargetFramework>net8.0</TargetFramework>

dotnet/samples/Hello/Hello.AppHost/Program.cs

@@ -1,7 +1,19 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Program.cs
 
+using Microsoft.Extensions.Hosting;
+
 var builder = DistributedApplication.CreateBuilder(args);
-var backend = builder.AddProject<Projects.Backend>("backend");
+var backend = builder.AddProject<Projects.Backend>("backend").WithExternalHttpEndpoints();
-builder.AddProject<Projects.HelloAgent>("client").WithReference(backend).WaitFor(backend);
+builder.AddProject<Projects.HelloAgent>("client")
-builder.Build().Run();
+    .WithReference(backend)
+    .WithEnvironment("AGENT_HOST", $"{backend.GetEndpoint("https").Property(EndpointProperty.Url)}")
+    .WaitFor(backend);
+
+using var app = builder.Build();
+
+await app.StartAsync();
+var url = backend.GetEndpoint("http").Url;
+Console.WriteLine("Backend URL: " + url);
+
+await app.WaitForShutdownAsync();

dotnet/samples/Hello/HelloAgent/Program.cs

@@ -19,7 +19,7 @@ var app = await AgentsApp.PublishMessageAsync("HelloAgents", new NewMessageRecei
 {
     Message = "World"
 }, local: true);
-
+//var app = await AgentsApp.StartAsync();
 await app.WaitForShutdownAsync();
 
 namespace Hello
@@ -33,7 +33,8 @@ namespace Hello
             ISayHello,
             IHandleConsole,
             IHandle<NewMessageReceived>,
-            IHandle<ConversationClosed>
+            IHandle<ConversationClosed>,
+            IHandle<Shutdown>
     {
         public async Task Handle(NewMessageReceived item)
         {
@@ -50,13 +51,14 @@ namespace Hello
         public async Task Handle(ConversationClosed item)
         {
             var goodbye = $"*********************  {item.UserId} said {item.UserMessage}  ************************";
-            var evt = new Output
+            var evt = new Output { Message = goodbye };
-            {
+            await PublishMessageAsync(evt).ConfigureAwait(true);
-                Message = goodbye
+            await PublishMessageAsync(new Shutdown()).ConfigureAwait(false);
-            };
+        }
-            await PublishMessageAsync(evt).ConfigureAwait(false);
 
-            // Signal shutdown.
+        public async Task Handle(Shutdown item)
+        {
+            Console.WriteLine("Shutting down...");
             hostApplicationLifetime.StopApplication();
         }
 

dotnet/samples/Hello/HelloAgentState/Program.cs

@@ -9,7 +9,7 @@ using Microsoft.AutoGen.Agents;
 var app = await AgentsApp.PublishMessageAsync("HelloAgents", new NewMessageReceived
 {
     Message = "World"
-}, local: true);
+}, local: false);
 
 await app.WaitForShutdownAsync();
 

dotnet/samples/dev-team/DevTeam.AppHost/DevTeam.AppHost.csproj

@@ -1,5 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
+  <Sdk Name="Aspire.AppHost.Sdk" Version="9.0.0" />
+
   <PropertyGroup>
     <OutputType>Exe</OutputType>
         <TargetFramework>net8.0</TargetFramework>

dotnet/src/AutoGen.Core/Function/FunctionAttribute.cs

@@ -98,7 +98,7 @@ public class FunctionContract
                 [NamespaceKey] = contract.Namespace,
                 [ClassNameKey] = contract.ClassName,
             },
-            Parameters = [.. contract.Parameters?.Select(p => (AIFunctionParameterMetadata)p)],
+            Parameters = [.. contract.Parameters?.Select(p => (AIFunctionParameterMetadata)p)!],
         };
     }
 }

dotnet/src/Microsoft.AutoGen/Abstractions/IAgentBase.cs

@@ -19,4 +19,5 @@ public interface IAgentBase
     Task<T> ReadAsync<T>(AgentId agentId, CancellationToken cancellationToken = default) where T : IMessage, new();
     ValueTask PublishEventAsync(CloudEvent item, CancellationToken cancellationToken = default);
     ValueTask PublishEventAsync(string topic, IMessage evt, CancellationToken cancellationToken = default);
+    List<string> Subscribe(string topic);
 }

dotnet/src/Microsoft.AutoGen/Abstractions/IAgentRuntime.cs

@@ -13,6 +13,7 @@ public interface IAgentRuntime
     ValueTask<AgentState> ReadAsync(AgentId agentId, CancellationToken cancellationToken = default);
     ValueTask SendResponseAsync(RpcRequest request, RpcResponse response, CancellationToken cancellationToken = default);
     ValueTask SendRequestAsync(IAgentBase agent, RpcRequest request, CancellationToken cancellationToken = default);
+    ValueTask SendMessageAsync(Message message, CancellationToken cancellationToken = default);
     ValueTask PublishEventAsync(CloudEvent @event, CancellationToken cancellationToken = default);
     void Update(Activity? activity, RpcRequest request);
     void Update(Activity? activity, CloudEvent cloudEvent);

dotnet/src/Microsoft.AutoGen/Abstractions/IAgentWorker.cs

@@ -8,6 +8,7 @@ public interface IAgentWorker
     ValueTask PublishEventAsync(CloudEvent evt, CancellationToken cancellationToken = default);
     ValueTask SendRequestAsync(IAgentBase agent, RpcRequest request, CancellationToken cancellationToken = default);
     ValueTask SendResponseAsync(RpcResponse response, CancellationToken cancellationToken = default);
+    ValueTask SendMessageAsync(Message message, CancellationToken cancellationToken = default);
     ValueTask StoreAsync(AgentState value, CancellationToken cancellationToken = default);
     ValueTask<AgentState> ReadAsync(AgentId agentId, CancellationToken cancellationToken = default);
 }

dotnet/src/Microsoft.AutoGen/Agents/AgentBase.cs

@@ -114,6 +114,27 @@ public abstract class AgentBase : IAgentBase, IHandle
                 break;
         }
     }
+    public List<string> Subscribe(string topic)
+    {
+        Message message = new()
+        {
+            AddSubscriptionRequest = new()
+            {
+                RequestId = Guid.NewGuid().ToString(),
+                Subscription = new Subscription
+                {
+                    TypeSubscription = new TypeSubscription
+                    {
+                        TopicType = topic,
+                        AgentType = this.AgentId.Key
+                    }
+                }
+            }
+        };
+        _context.SendMessageAsync(message).AsTask().Wait();
+
+        return new List<string> { topic };
+    }
     public async Task StoreAsync(AgentState state, CancellationToken cancellationToken = default)
     {
         await _context.StoreAsync(state, cancellationToken).ConfigureAwait(false);

dotnet/src/Microsoft.AutoGen/Agents/AgentRuntime.cs

@@ -45,6 +45,10 @@ internal sealed class AgentRuntime(AgentId agentId, IAgentWorker worker, ILogger
     {
         await worker.SendRequestAsync(agent, request, cancellationToken).ConfigureAwait(false);
     }
+    public async ValueTask SendMessageAsync(Message message, CancellationToken cancellationToken = default)
+    {
+        await worker.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
+    }
     public async ValueTask PublishEventAsync(CloudEvent @event, CancellationToken cancellationToken = default)
     {
         await worker.PublishEventAsync(@event, cancellationToken).ConfigureAwait(false);

dotnet/src/Microsoft.AutoGen/Agents/Services/AgentWorker.cs

@@ -47,7 +47,7 @@ public class AgentWorker :
     {
         foreach (var (typeName, _) in _agentTypes)
         {
-            if (typeName == "Client") { continue; }
+            if (typeName == nameof(Client)) { continue; }
             var agent = GetOrActivateAgent(new AgentId(typeName, cloudEvent.Source));
             agent.ReceiveMessage(new Message { CloudEvent = cloudEvent });
         }
@@ -63,6 +63,10 @@ public class AgentWorker :
     {
         return _mailbox.Writer.WriteAsync(new Message { Response = response }, cancellationToken);
     }
+    public ValueTask SendMessageAsync(Message message, CancellationToken cancellationToken = default)
+    {
+        return _mailbox.Writer.WriteAsync(message, cancellationToken);
+    }
     public ValueTask StoreAsync(AgentState value, CancellationToken cancellationToken = default)
     {
         var agentId = value.AgentId ?? throw new InvalidOperationException("AgentId is required when saving AgentState.");
@@ -92,6 +96,10 @@ public class AgentWorker :
                 if (message == null) { continue; }
                 switch (message)
                 {
+                    case Message.MessageOneofCase.AddSubscriptionResponse:
+                        break;
+                    case Message.MessageOneofCase.RegisterAgentTypeResponse:
+                        break;
                     case Message msg:
 
                         var item = msg.CloudEvent;

dotnet/src/Microsoft.AutoGen/Agents/Services/AgentWorkerHostingExtensions.cs

@@ -29,7 +29,7 @@ public static class AgentWorkerHostingExtensions
 
     public static IHostApplicationBuilder AddLocalAgentService(this IHostApplicationBuilder builder, bool useGrpc = true)
     {
-        return builder.AddAgentService(local: true, useGrpc);
+        return builder.AddAgentService(local: false, useGrpc);
     }
 
     public static WebApplication MapAgentService(this WebApplication app, bool local = false, bool useGrpc = true)

dotnet/src/Microsoft.AutoGen/Agents/Services/Grpc/GrpcAgentWorker.cs

@@ -85,6 +85,13 @@ public sealed class GrpcAgentWorker(
                             }
                             break;
 
+                        case Message.MessageOneofCase.AddSubscriptionResponse:
+                            if (!message.AddSubscriptionResponse.Success)
+                            {
+                                throw new InvalidOperationException($"Failed to add subscription: '{message.AddSubscriptionResponse.Error}'.");
+                            }
+                            break;
+
                         case Message.MessageOneofCase.CloudEvent:
 
                             // HACK: Send the message to an instance of each agent type
@@ -153,6 +160,13 @@ public sealed class GrpcAgentWorker(
                 item.WriteCompletionSource?.TrySetCanceled();
                 break;
             }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable)
+            {
+                // we could not connect to the endpoint - most likely we have the wrong port or failed ssl
+                // we need to let the user know what port we tried to connect to and then do backoff and retry
+                _logger.LogError(ex, "Error connecting to GRPC endpoint {Endpoint}.", channel.ToString());
+                break;
+            }
             catch (Exception ex) when (!_shutdownCts.IsCancellationRequested)
             {
                 item.WriteCompletionSource?.TrySetException(ex);
@@ -230,6 +244,11 @@ public sealed class GrpcAgentWorker(
         await WriteChannelAsync(new Message { Request = request }, cancellationToken).ConfigureAwait(false);
     }
     // new is intentional
+    public new async ValueTask SendMessageAsync(Message message, CancellationToken cancellationToken = default)
+    {
+        await WriteChannelAsync(message, cancellationToken).ConfigureAwait(false);
+    }
+    // new is intentional
     public new async ValueTask PublishEventAsync(CloudEvent @event, CancellationToken cancellationToken = default)
     {
         await WriteChannelAsync(new Message { CloudEvent = @event }, cancellationToken).ConfigureAwait(false);

dotnet/src/Microsoft.AutoGen/Agents/Services/Grpc/GrpcAgentWorkerHostBuilderExtension.cs

@@ -11,12 +11,12 @@ namespace Microsoft.AutoGen.Agents;
 
 public static class GrpcAgentWorkerHostBuilderExtensions
 {
-    private const string _defaultAgentServiceAddress = "https://localhost:5001";
+    private const string _defaultAgentServiceAddress = "https://localhost:53071";
-    public static IHostApplicationBuilder AddGrpcAgentWorker(this IHostApplicationBuilder builder, string agentServiceAddress = _defaultAgentServiceAddress)
+    public static IHostApplicationBuilder AddGrpcAgentWorker(this IHostApplicationBuilder builder, string? agentServiceAddress = null)
     {
         builder.Services.AddGrpcClient<AgentRpc.AgentRpcClient>(options =>
         {
-            options.Address = new Uri(agentServiceAddress);
+            options.Address = new Uri(agentServiceAddress ?? builder.Configuration["AGENT_HOST"] ?? _defaultAgentServiceAddress);
             options.ChannelOptionsActions.Add(channelOptions =>
             {
 

dotnet/src/Microsoft.AutoGen/Agents/Services/Grpc/GrpcGateway.cs

@@ -16,10 +16,13 @@ public sealed class GrpcGateway : BackgroundService, IGateway
     private readonly IClusterClient _clusterClient;
     private readonly ConcurrentDictionary<string, AgentState> _agentState = new();
     private readonly IRegistryGrain _gatewayRegistry;
+    private readonly ISubscriptionsGrain _subscriptions;
     private readonly IGateway _reference;
     // The agents supported by each worker process.
     private readonly ConcurrentDictionary<string, List<GrpcWorkerConnection>> _supportedAgentTypes = [];
     public readonly ConcurrentDictionary<IConnection, IConnection> _workers = new();
+    private readonly ConcurrentDictionary<string, Subscription> _subscriptionsByAgentType = new();
+    private readonly ConcurrentDictionary<string, List<string>> _subscriptionsByTopic = new();
 
     // The mapping from agent id to worker process.
     private readonly ConcurrentDictionary<(string Type, string Key), GrpcWorkerConnection> _agentDirectory = new();
@@ -33,6 +36,7 @@ public sealed class GrpcGateway : BackgroundService, IGateway
         _clusterClient = clusterClient;
         _reference = clusterClient.CreateObjectReference<IGateway>(this);
         _gatewayRegistry = clusterClient.GetGrain<IRegistryGrain>(0);
+        _subscriptions = clusterClient.GetGrain<ISubscriptionsGrain>(0);
     }
     public async ValueTask BroadcastEvent(CloudEvent evt)
     {
@@ -102,16 +106,54 @@ public sealed class GrpcGateway : BackgroundService, IGateway
             case Message.MessageOneofCase.RegisterAgentTypeRequest:
                 await RegisterAgentTypeAsync(connection, message.RegisterAgentTypeRequest);
                 break;
+            case Message.MessageOneofCase.AddSubscriptionRequest:
+                await AddSubscriptionAsync(connection, message.AddSubscriptionRequest);
+                break;
             default:
-                throw new InvalidOperationException($"Unknown message type for message '{message}'.");
+                // if it wasn't recognized return bad request
+                await RespondBadRequestAsync(connection, $"Unknown message type for message '{message}'.");
+                break;
         };
     }
+    private async ValueTask RespondBadRequestAsync(GrpcWorkerConnection connection, string error)
+    {
+        throw new RpcException(new Status(StatusCode.InvalidArgument, error));
+    }
+    private async ValueTask AddSubscriptionAsync(GrpcWorkerConnection connection, AddSubscriptionRequest request)
+    {
+        var topic = request.Subscription.TypeSubscription.TopicType;
+        var agentType = request.Subscription.TypeSubscription.AgentType;
+        _subscriptionsByAgentType[agentType] = request.Subscription;
+        _subscriptionsByTopic.GetOrAdd(topic, _ => []).Add(agentType);
+        await _subscriptions.Subscribe(topic, agentType);
+        //var response = new AddSubscriptionResponse { RequestId = request.RequestId, Error = "", Success = true };
+        Message response = new()
+        {
+            AddSubscriptionResponse = new()
+            {
+                RequestId = request.RequestId,
+                Error = "",
+                Success = true
+            }
+        };
+        await connection.ResponseStream.WriteAsync(response).ConfigureAwait(false);
+    }
     private async ValueTask RegisterAgentTypeAsync(GrpcWorkerConnection connection, RegisterAgentTypeRequest msg)
     {
         connection.AddSupportedType(msg.Type);
         _supportedAgentTypes.GetOrAdd(msg.Type, _ => []).Add(connection);
 
-        await _gatewayRegistry.RegisterAgentType(msg.Type, _reference);
+        await _gatewayRegistry.RegisterAgentType(msg.Type, _reference).ConfigureAwait(true);
+        Message response = new()
+        {
+            RegisterAgentTypeResponse = new()
+            {
+                RequestId = msg.RequestId,
+                Error = "",
+                Success = true
+            }
+        };
+        await connection.ResponseStream.WriteAsync(response).ConfigureAwait(false);
     }
     private async ValueTask DispatchEventAsync(CloudEvent evt)
     {

dotnet/src/Microsoft.AutoGen/Agents/Services/Host.cs

@@ -14,11 +14,11 @@ public static class Host
         builder.AddServiceDefaults();
         if (local)
         {
-            builder.AddLocalAgentService(useGrpc);
+            builder.AddLocalAgentService(useGrpc: useGrpc);
         }
         else
         {
-            builder.AddAgentService(useGrpc);
+            builder.AddAgentService(useGrpc: useGrpc);
         }
         var app = builder.Build();
         app.MapAgentService(local, useGrpc);

dotnet/src/Microsoft.AutoGen/Agents/Services/HostBuilderExtensions.cs

@@ -15,7 +15,7 @@ namespace Microsoft.AutoGen.Agents;
 
 public static class HostBuilderExtensions
 {
-    private const string _defaultAgentServiceAddress = "https://localhost:5001";
+    private const string _defaultAgentServiceAddress = "https://localhost:53071";
 
     public static IHostApplicationBuilder AddAgent<
         [DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TAgent>(this IHostApplicationBuilder builder, string typeName) where TAgent : AgentBase
@@ -28,12 +28,12 @@ public static class HostBuilderExtensions
     public static IHostApplicationBuilder AddAgent(this IHostApplicationBuilder builder, string typeName, Type agentType)
     {
         builder.Services.AddKeyedSingleton("AgentTypes", (sp, key) => Tuple.Create(typeName, agentType));
-
         return builder;
     }
 
-    public static IHostApplicationBuilder AddAgentWorker(this IHostApplicationBuilder builder, string agentServiceAddress = _defaultAgentServiceAddress, bool local = false)
+    public static IHostApplicationBuilder AddAgentWorker(this IHostApplicationBuilder builder, string? agentServiceAddress = null, bool local = false)
     {
+        agentServiceAddress ??= builder.Configuration["AGENT_HOST"] ?? _defaultAgentServiceAddress;
         builder.Services.TryAddSingleton(DistributedContextPropagator.Current);
 
         // if !local, then add the gRPC client

dotnet/src/Microsoft.AutoGen/Agents/Services/Orleans/ISubscriptionsGrain.cs

@@ -0,0 +1,10 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ISubscriptionsGrain.cs
+
+namespace Microsoft.AutoGen.Agents;
+public interface ISubscriptionsGrain : IGrainWithIntegerKey
+{
+    ValueTask Subscribe(string agentType, string topic);
+    ValueTask Unsubscribe(string agentType, string topic);
+    ValueTask<Dictionary<string, List<string>>> GetSubscriptions(string agentType);
+}

dotnet/src/Microsoft.AutoGen/Agents/Services/Orleans/RegistryGrain.cs

@@ -5,7 +5,7 @@ using Microsoft.AutoGen.Abstractions;
 
 namespace Microsoft.AutoGen.Agents;
 
-public sealed class RegistryGrain : Grain, IRegistryGrain
+internal sealed class RegistryGrain : Grain, IRegistryGrain
 {
     // TODO: use persistent state for some of these or (better) extend Orleans to implement some of this natively.
     private readonly Dictionary<IGateway, WorkerState> _workerStates = new();

dotnet/src/Microsoft.AutoGen/Agents/Services/Orleans/SubscriptionsGrain.cs

@@ -0,0 +1,48 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// SubscriptionsGrain.cs
+
+namespace Microsoft.AutoGen.Agents;
+
+internal sealed class SubscriptionsGrain([PersistentState("state", "PubSubStore")] IPersistentState<SubscriptionsState> state) : Grain, ISubscriptionsGrain
+{
+    private readonly Dictionary<string, List<string>> _subscriptions = new();
+    public ValueTask<Dictionary<string, List<string>>> GetSubscriptions(string agentType)
+    {
+        return new ValueTask<Dictionary<string, List<string>>>(_subscriptions);
+    }
+    public ValueTask Subscribe(string agentType, string topic)
+    {
+        if (!_subscriptions.TryGetValue(topic, out var subscriptions))
+        {
+            subscriptions = _subscriptions[topic] = [];
+        }
+        if (!subscriptions.Contains(agentType))
+        {
+            subscriptions.Add(agentType);
+        }
+        _subscriptions[topic] = subscriptions;
+        state.State.Subscriptions = _subscriptions;
+        state.WriteStateAsync();
+
+        return ValueTask.CompletedTask;
+    }
+    public ValueTask Unsubscribe(string agentType, string topic)
+    {
+        if (!_subscriptions.TryGetValue(topic, out var subscriptions))
+        {
+            subscriptions = _subscriptions[topic] = [];
+        }
+        if (!subscriptions.Contains(agentType))
+        {
+            subscriptions.Remove(agentType);
+        }
+        _subscriptions[topic] = subscriptions;
+        state.State.Subscriptions = _subscriptions;
+        state.WriteStateAsync();
+        return ValueTask.CompletedTask;
+    }
+}
+public sealed class SubscriptionsState
+{
+    public Dictionary<string, List<string>> Subscriptions { get; set; } = new();
+}

dotnet/src/Microsoft.AutoGen/Extensions/AIModelClientHostingExtensions/AIModelClientHostingExtensions.csproj

@@ -14,6 +14,6 @@
     <PackageReference Include="Microsoft.Extensions.AI.OpenAI" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
     <PackageReference Include="Microsoft.Extensions.Hosting" />
-    <PackageReference Include="System.Text.Json" VersionOverride="9.0.0-rc.2.24473.5" />
+    <PackageReference Include="System.Text.Json" />
   </ItemGroup>
 </Project>

dotnet/test/AutoGen.Tests/AutoGen.Tests.csproj

@@ -14,7 +14,7 @@
     <ProjectReference Include="..\..\src\AutoGen\AutoGen.csproj" />
     <ProjectReference Include="..\AutoGen.Test.Share\AutoGen.Tests.Share.csproj" />
     <PackageReference Include="Microsoft.Extensions.AI" />
-    <PackageReference Include="System.Text.Json" VersionOverride="9.0.0-rc.2.24473.5" />
+    <PackageReference Include="System.Text.Json" />
   </ItemGroup>
 
   <ItemGroup>

python/.gitignore

@@ -172,3 +172,6 @@ docs/**/jupyter_execute
 
 # Temporary files
 tmp_code_*.py
+
+# .NET Development settings
+appsettings.Development.json