Dotnet Grpc worker implementation

dotnet/AutoGen.sln

@@ -118,6 +118,10 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Hello", "Hello", "{F42F9C8E
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AutoGen.Core.Grpc", "src\Microsoft.AutoGen\Core.Grpc\Microsoft.AutoGen.Core.Grpc.csproj", "{3D83C6DB-ACEA-48F3-959F-145CCD2EE135}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GettingStartedGrpc", "samples\GettingStartedGrpc\GettingStartedGrpc.csproj", "{C3740DF1-18B1-4607-81E4-302F0308C848}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Microsoft.AutoGen.Core.Grpc.Tests", "test\Microsoft.AutoGen.Core.Grpc.Tests\Microsoft.AutoGen.Core.Grpc.Tests.csproj", "{23A028D3-5EB1-4FA0-9CD1-A1340B830579}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -306,6 +310,14 @@ Global
 		{AAD593FE-A49B-425E-A9FE-A0022CD25E3D}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{AAD593FE-A49B-425E-A9FE-A0022CD25E3D}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{AAD593FE-A49B-425E-A9FE-A0022CD25E3D}.Release|Any CPU.Build.0 = Release|Any CPU
+		{C3740DF1-18B1-4607-81E4-302F0308C848}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{C3740DF1-18B1-4607-81E4-302F0308C848}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{C3740DF1-18B1-4607-81E4-302F0308C848}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{C3740DF1-18B1-4607-81E4-302F0308C848}.Release|Any CPU.Build.0 = Release|Any CPU
+		{23A028D3-5EB1-4FA0-9CD1-A1340B830579}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{23A028D3-5EB1-4FA0-9CD1-A1340B830579}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{23A028D3-5EB1-4FA0-9CD1-A1340B830579}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{23A028D3-5EB1-4FA0-9CD1-A1340B830579}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -359,6 +371,8 @@ Global
 		{3D83C6DB-ACEA-48F3-959F-145CCD2EE135} = {18BF8DD7-0585-48BF-8F97-AD333080CE06}
 		{AAD593FE-A49B-425E-A9FE-A0022CD25E3D} = {F42F9C8E-7BD9-4687-9B63-AFFA461AF5C1}
 		{F42F9C8E-7BD9-4687-9B63-AFFA461AF5C1} = {CE0AA8D5-12B8-4628-9589-DAD8CB0DDCF6}
+		{C3740DF1-18B1-4607-81E4-302F0308C848} = {CE0AA8D5-12B8-4628-9589-DAD8CB0DDCF6}
+		{23A028D3-5EB1-4FA0-9CD1-A1340B830579} = {F823671B-3ECA-4AE6-86DA-25E920D3FE64}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {93384647-528D-46C8-922C-8DB36A382F0B}

dotnet/samples/GettingStartedGrpc/Checker.cs

@@ -0,0 +1,34 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Checker.cs
+
+using Microsoft.AutoGen.Contracts;
+using Microsoft.AutoGen.Core;
+using Microsoft.Extensions.Hosting;
+using TerminationF = System.Func<int, bool>;
+
+namespace GettingStartedGrpcSample;
+
+[TypeSubscription("default")]
+public class Checker(
+    AgentId id,
+    IAgentRuntime runtime,
+    IHostApplicationLifetime hostApplicationLifetime,
+    TerminationF runUntilFunc
+    ) :
+        BaseAgent(id, runtime, "Modifier", null),
+        IHandle<Events.CountUpdate>
+{
+    public async ValueTask HandleAsync(Events.CountUpdate item, MessageContext messageContext)
+    {
+        if (!runUntilFunc(item.NewCount))
+        {
+            Console.WriteLine($"\nChecker:\n{item.NewCount} passed the check, continue.");
+            await this.PublishMessageAsync(new Events.CountMessage { Content = item.NewCount }, new TopicId("default"));
+        }
+        else
+        {
+            Console.WriteLine($"\nChecker:\n{item.NewCount} failed the check, stopping.");
+            hostApplicationLifetime.StopApplication();
+        }
+    }
+}

dotnet/samples/GettingStartedGrpc/GettingStartedGrpc.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <RootNamespace>getting_started</RootNamespace>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Microsoft.AutoGen\Contracts\Microsoft.AutoGen.Contracts.csproj" />
+    <ProjectReference Include="..\..\src\Microsoft.AutoGen\Core\Microsoft.AutoGen.Core.csproj" />
+    <ProjectReference Include="..\..\src\Microsoft.AutoGen\Core.Grpc\Microsoft.AutoGen.Core.Grpc.csproj" />
+  </ItemGroup>
+
+
+    <ItemGroup>
+    <Protobuf Include="message.proto" GrpcServices="Client" Link="Protos\message.proto" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Grpc.Tools" PrivateAssets="All" />
+  </ItemGroup>
+
+</Project>

dotnet/samples/GettingStartedGrpc/Modifier.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Modifier.cs
+
+using Microsoft.AutoGen.Contracts;
+using Microsoft.AutoGen.Core;
+
+using ModifyF = System.Func<int, int>;
+
+namespace GettingStartedGrpcSample;
+
+[TypeSubscription("default")]
+public class Modifier(
+    AgentId id,
+    IAgentRuntime runtime,
+    ModifyF modifyFunc
+    ) :
+        BaseAgent(id, runtime, "Modifier", null),
+        IHandle<Events.CountMessage>
+{
+
+    public async ValueTask HandleAsync(Events.CountMessage item, MessageContext messageContext)
+    {
+        int newValue = modifyFunc(item.Content);
+        Console.WriteLine($"\nModifier:\nModified {item.Content} to {newValue}");
+
+        var updateMessage = new Events.CountUpdate { NewCount = newValue };
+        await this.PublishMessageAsync(updateMessage, topic: new TopicId("default"));
+    }
+}

dotnet/samples/GettingStartedGrpc/Program.cs

@@ -0,0 +1,36 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Program.cs
+using GettingStartedGrpcSample;
+using Microsoft.AutoGen.Contracts;
+using Microsoft.AutoGen.Core;
+using Microsoft.AutoGen.Core.Grpc;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using ModifyF = System.Func<int, int>;
+using TerminationF = System.Func<int, bool>;
+
+ModifyF modifyFunc = (int x) => x - 1;
+TerminationF runUntilFunc = (int x) =>
+{
+    return x <= 1;
+};
+
+AgentsAppBuilder appBuilder = new AgentsAppBuilder();
+appBuilder.AddGrpcAgentWorker("http://localhost:50051");
+
+appBuilder.Services.TryAddSingleton(modifyFunc);
+appBuilder.Services.TryAddSingleton(runUntilFunc);
+
+appBuilder.AddAgent<Checker>("Checker");
+appBuilder.AddAgent<Modifier>("Modifier");
+
+var app = await appBuilder.BuildAsync();
+await app.StartAsync();
+
+// Send the initial count to the agents app, running on the `local` runtime, and pass through the registered services via the application `builder`
+await app.PublishMessageAsync(new GettingStartedGrpcSample.Events.CountMessage
+{
+    Content = 10
+}, new TopicId("default"));
+
+// Run until application shutdown
+await app.WaitForShutdownAsync();

dotnet/samples/GettingStartedGrpc/message.proto

@@ -0,0 +1,11 @@
+syntax = "proto3";
+
+option csharp_namespace = "GettingStartedGrpcSample.Events";
+
+message CountMessage {
+    int32 content = 1;
+}
+
+message CountUpdate {
+    int32 new_count = 1;
+}

dotnet/src/Microsoft.AutoGen/Core.Grpc/AgentsAppBuilderExtensions.cs

@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// AgentsAppBuilderExtensions.cs
+
+using System.Diagnostics;
+using Grpc.Core;
+using Grpc.Net.Client.Configuration;
+using Microsoft.AutoGen.Contracts;
+using Microsoft.AutoGen.Protobuf;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
+namespace Microsoft.AutoGen.Core.Grpc;
+
+public static class AgentsAppBuilderExtensions
+{
+    private const string _defaultAgentServiceAddress = "http://localhost:53071";
+
+    // TODO: How do we ensure AddGrpcAgentWorker and UseInProcessRuntime are mutually exclusive?
+    public static AgentsAppBuilder AddGrpcAgentWorker(this AgentsAppBuilder builder, string? agentServiceAddress = null)
+    {
+        builder.Services.AddGrpcClient<AgentRpc.AgentRpcClient>(options =>
+        {
+            options.Address = new Uri(agentServiceAddress ?? builder.Configuration.GetValue("AGENT_HOST", _defaultAgentServiceAddress));
+            options.ChannelOptionsActions.Add(channelOptions =>
+            {
+                var loggerFactory = new LoggerFactory();
+                if (Debugger.IsAttached)
+                {
+                    channelOptions.HttpHandler = new SocketsHttpHandler
+                    {
+                        EnableMultipleHttp2Connections = false,
+                        KeepAlivePingDelay = TimeSpan.FromSeconds(200),
+                        KeepAlivePingTimeout = TimeSpan.FromSeconds(100),
+                        KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always
+                    };
+                }
+                else
+                {
+                    channelOptions.HttpHandler = new SocketsHttpHandler
+                    {
+                        EnableMultipleHttp2Connections = true,
+                        KeepAlivePingDelay = TimeSpan.FromSeconds(20),
+                        KeepAlivePingTimeout = TimeSpan.FromSeconds(10),
+                        KeepAlivePingPolicy = HttpKeepAlivePingPolicy.WithActiveRequests
+                    };
+                }
+
+                var methodConfig = new MethodConfig
+                {
+                    Names = { MethodName.Default },
+                    RetryPolicy = new RetryPolicy
+                    {
+                        MaxAttempts = 5,
+                        InitialBackoff = TimeSpan.FromSeconds(1),
+                        MaxBackoff = TimeSpan.FromSeconds(5),
+                        BackoffMultiplier = 1.5,
+                        RetryableStatusCodes = { StatusCode.Unavailable }
+                    }
+                };
+
+                channelOptions.ServiceConfig = new() { MethodConfigs = { methodConfig } };
+                channelOptions.ThrowOperationCanceledOnCancellation = true;
+            });
+        });
+
+        builder.Services.TryAddSingleton(DistributedContextPropagator.Current);
+        builder.Services.AddSingleton<IAgentRuntime, GrpcAgentRuntime>();
+        builder.Services.AddHostedService<GrpcAgentRuntime>(services =>
+        {
+            return (services.GetRequiredService<IAgentRuntime>() as GrpcAgentRuntime)!;
+        });
+
+        return builder;
+    }
+}

dotnet/src/Microsoft.AutoGen/Core.Grpc/CloudEventExtensions.cs

@@ -0,0 +1,43 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// CloudEventExtensions.cs
+
+using Microsoft.AutoGen.Contracts;
+
+namespace Microsoft.AutoGen.Core.Grpc;
+
+internal static class CloudEventExtensions
+{
+    // Convert an ISubscrptionDefinition to a Protobuf Subscription
+    internal static CloudEvent CreateCloudEvent(Google.Protobuf.WellKnownTypes.Any payload, TopicId topic, string dataType, AgentId? sender, string messageId)
+    {
+        var attributes = new Dictionary<string, CloudEvent.Types.CloudEventAttributeValue>
+        {
+            {
+                Constants.DATA_CONTENT_TYPE_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = Constants.DATA_CONTENT_TYPE_PROTOBUF_VALUE }
+            },
+            {
+                Constants.DATA_SCHEMA_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = dataType }
+            },
+            {
+                Constants.MESSAGE_KIND_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = Constants.MESSAGE_KIND_VALUE_PUBLISH }
+            }
+        };
+
+        if (sender != null)
+        {
+            var senderNonNull = (AgentId)sender;
+            attributes.Add(Constants.AGENT_SENDER_TYPE_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = senderNonNull.Type });
+            attributes.Add(Constants.AGENT_SENDER_KEY_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = senderNonNull.Key });
+        }
+
+        return new CloudEvent
+        {
+            ProtoData = payload,
+            Type = topic.Type,
+            Source = topic.Source,
+            Id = messageId,
+            Attributes = { attributes }
+        };
+
+    }
+}

dotnet/src/Microsoft.AutoGen/Core.Grpc/Constants.cs

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Constants.cs
+
+namespace Microsoft.AutoGen.Core.Grpc;
+
+public static class Constants
+{
+    public const string DATA_CONTENT_TYPE_PROTOBUF_VALUE = "application/x-protobuf";
+    public const string DATA_CONTENT_TYPE_JSON_VALUE = "application/json";
+    public const string DATA_CONTENT_TYPE_TEXT_VALUE = "text/plain";
+
+    public const string DATA_CONTENT_TYPE_ATTR = "datacontenttype";
+    public const string DATA_SCHEMA_ATTR = "dataschema";
+    public const string AGENT_SENDER_TYPE_ATTR = "agagentsendertype";
+    public const string AGENT_SENDER_KEY_ATTR = "agagentsenderkey";
+
+    public const string MESSAGE_KIND_ATTR = "agmsgkind";
+    public const string MESSAGE_KIND_VALUE_PUBLISH = "publish";
+    public const string MESSAGE_KIND_VALUE_RPC_REQUEST = "rpc_request";
+    public const string MESSAGE_KIND_VALUE_RPC_RESPONSE = "rpc_response";
+}