feat(a2a-client): the a2a tool with agents, send, status and cancel, and per-session wiring
This commit is contained in:
EugeneTes
@@ -0,0 +1,59 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Volcanic.AgenticCore.Abstractions.Extensions;
|
||||
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
|
||||
|
||||
namespace Volcanic.AgenticCore.Extensions.A2A;
|
||||
|
||||
/// <summary>
|
||||
/// Per session (the host adds it, e.g. per Telegram chat; never host-wide): the <c>a2a</c> tool bound to the host's
|
||||
/// <paramref name="target"/>, the hop counter from <c>a2a.inbound</c>, and the live-session delivery hook that also
|
||||
/// flushes results kept while the target had no live conversation.
|
||||
/// </summary>
|
||||
public sealed class A2AClientExtension(A2AClientSettings settings, string target, A2AWatchRegistry? registry = null) : IExtension
|
||||
{
|
||||
public const string ExtensionName = "a2a-client";
|
||||
|
||||
/// <summary>Emitted by the A2A server's per-session extension before each inbound prompt: <c>{"hops": n, "caller": "…"}</c>.</summary>
|
||||
public const string InboundChannel = "a2a.inbound";
|
||||
|
||||
private readonly A2AWatchRegistry _registry = registry ?? A2AWatchRegistry.Shared;
|
||||
private volatile int _hops;
|
||||
|
||||
public string Name => ExtensionName;
|
||||
|
||||
public ValueTask ActivateAsync(IExtensionApi api, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(api);
|
||||
api.RegisterTool(new A2ATool(settings, target, _registry, () => _hops));
|
||||
api.Events.On(InboundChannel, data =>
|
||||
{
|
||||
if (data is JsonElement { ValueKind: JsonValueKind.Object } e && e.TryGetProperty("hops", out var h) && h.TryGetInt32(out var n))
|
||||
_hops = n;
|
||||
});
|
||||
api.On<SessionStartEvent>((_, ctx, _) =>
|
||||
{
|
||||
var workspace = ctx.Cwd;
|
||||
_registry.Live.Attach(target, workspace, api);
|
||||
// Not inline: delivering starts a run, which mustn't happen inside session creation.
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await _registry.FlushUndeliveredAsync(target, workspace, CancellationToken.None).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OutOfMemoryException)
|
||||
{
|
||||
api.Logger.LogWarning(ex, "A2A: flushing undelivered results for {Target} failed", target);
|
||||
}
|
||||
});
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
api.On<SessionShutdownEvent>((_, _, _) =>
|
||||
{
|
||||
_registry.Live.Detach(target, api);
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
292
extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs
Normal file
292
extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs
Normal file
@@ -0,0 +1,292 @@
|
||||
using System.ComponentModel;
|
||||
using System.Net;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using A2A;
|
||||
using Volcanic.AgenticCore.Abstractions.Tools;
|
||||
using A2AMessage = global::A2A.Message;
|
||||
|
||||
namespace Volcanic.AgenticCore.Extensions.A2A;
|
||||
|
||||
public sealed record A2AArgs(
|
||||
[property: Description("\"agents\" (list remote agents), \"send\" (message an agent), \"status\" (check a task) or \"cancel\" (cancel a task)")] string Action,
|
||||
[property: Description("Agent name from .volcanic/a2a/agents.json (send, status, cancel)")] string? Agent = null,
|
||||
[property: Description("Text to send (send)")] string? Message = null,
|
||||
[property: Description("Continue an earlier conversation with this agent: the context id from a previous result (send)")] string? ContextId = null,
|
||||
[property: Description("A task id: answer the agent's question (send), or the task to check or cancel (status, cancel)")] string? TaskId = null,
|
||||
[property: Description("Workspace files to attach (send)")] string[]? Files = null);
|
||||
|
||||
/// <summary>The <c>a2a</c> tool: remote agents from <c>agents.json</c>. Errors are thrown with readable, key-free messages.</summary>
|
||||
internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatchRegistry registry, Func<int> inboundHops) : Tool<A2AArgs>
|
||||
{
|
||||
/// <summary>Symlink hops followed when confining a path (a loop fails earlier in the OS).</summary>
|
||||
private const int MaxLinkDepth = 40;
|
||||
|
||||
public override string Name => "a2a";
|
||||
|
||||
public override string Description =>
|
||||
"Talk to other AI agents over the A2A protocol. Agents are configured in .volcanic/a2a/agents.json " +
|
||||
"([{\"name\", \"url\", \"key\"}]). Use action=agents to see them and their skills, action=send to delegate work.";
|
||||
|
||||
public override string? PromptSnippet => "a2a: delegate to other AI agents listed in .volcanic/a2a/agents.json";
|
||||
|
||||
public override IReadOnlyList<string> PromptGuidelines =>
|
||||
[
|
||||
"When a remote agent asks a question (input-required), answer with action=send and its taskId; ask the user first if you don't know the answer.",
|
||||
"Long remote tasks are watched for you: their result arrives later as a message starting with [A2A result from …].",
|
||||
];
|
||||
|
||||
protected override async Task<ToolResult> ExecuteAsync(A2AArgs args, ToolInvocation invocation, CancellationToken cancellationToken)
|
||||
{
|
||||
var workspace = invocation.Context.Cwd;
|
||||
return args.Action?.Trim().ToLowerInvariant() switch
|
||||
{
|
||||
"agents" => ToolResult.Text(await ListAsync(workspace, cancellationToken).ConfigureAwait(false)),
|
||||
"send" => ToolResult.Text(await SendAsync(workspace, args, cancellationToken).ConfigureAwait(false)),
|
||||
"status" => ToolResult.Text(await StatusAsync(workspace, args, cancellationToken).ConfigureAwait(false)),
|
||||
"cancel" => ToolResult.Text(await CancelAsync(workspace, args, cancellationToken).ConfigureAwait(false)),
|
||||
_ => throw new ArgumentException("action must be agents, send, status or cancel."),
|
||||
};
|
||||
}
|
||||
|
||||
private async Task<string> ListAsync(string workspace, CancellationToken ct)
|
||||
{
|
||||
var file = AgentsFile.Read(workspace);
|
||||
if (file.Error is not null) return file.Error;
|
||||
if (file.Agents.Count == 0)
|
||||
{
|
||||
var none = "No remote agents configured. Add them to .volcanic/a2a/agents.json: [{\"name\": \"travel-bot\", \"url\": \"https://…\", \"key\": \"…\"}]";
|
||||
return file.Problems.Count == 0 ? none : none + string.Concat(file.Problems.Select(p => $"\n(skipped: {p})"));
|
||||
}
|
||||
var sb = new StringBuilder("Remote agents:");
|
||||
foreach (var entry in file.Agents)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var agent = await registry.Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false);
|
||||
var card = agent.Card;
|
||||
sb.Append($"\n- {entry.Name}: {entry.Description ?? card.Description} ({card.Name})");
|
||||
var caps = new List<string>();
|
||||
if (agent.Streaming) caps.Add("streaming");
|
||||
if (agent.SupportsPush) caps.Add("push");
|
||||
if (caps.Count > 0) sb.Append($" [{string.Join(", ", caps)}]");
|
||||
foreach (var skill in card.Skills ?? [])
|
||||
{
|
||||
sb.Append($"\n - skill {skill.Id}: {skill.Name} — {skill.Description}");
|
||||
if (skill.Examples is { Count: > 0 } examples) sb.Append($" (e.g. {string.Join("; ", examples.Take(3))})");
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (ex is not OutOfMemoryException && (ex is not OperationCanceledException || !ct.IsCancellationRequested))
|
||||
{
|
||||
sb.Append($"\n- {entry.Name}: unreachable ({Clean(ex.Message, entry)})");
|
||||
}
|
||||
}
|
||||
foreach (var problem in file.Problems) sb.Append($"\n(skipped: {problem})");
|
||||
return sb.ToString();
|
||||
}
|
||||
|
||||
private async Task<string> SendAsync(string workspace, A2AArgs args, CancellationToken ct)
|
||||
{
|
||||
var entry = Entry(workspace, args.Agent);
|
||||
if (string.IsNullOrWhiteSpace(args.Message) && args.Files is not { Length: > 0 })
|
||||
throw new ArgumentException("send needs a message or files.");
|
||||
var parts = new List<Part>();
|
||||
if (!string.IsNullOrWhiteSpace(args.Message)) parts.Add(Part.FromText(args.Message));
|
||||
foreach (var path in args.Files ?? []) parts.Add(FilePart(workspace, path));
|
||||
|
||||
return await CallAsync(entry, ct, async agent =>
|
||||
{
|
||||
var response = await agent.Client.SendMessageAsync(new SendMessageRequest
|
||||
{
|
||||
Message = new A2AMessage
|
||||
{
|
||||
Role = Role.User,
|
||||
MessageId = Guid.NewGuid().ToString("N"),
|
||||
Parts = parts,
|
||||
ContextId = string.IsNullOrWhiteSpace(args.ContextId) ? null : args.ContextId,
|
||||
TaskId = string.IsNullOrWhiteSpace(args.TaskId) ? null : args.TaskId,
|
||||
},
|
||||
Configuration = new SendMessageConfiguration { ReturnImmediately = true },
|
||||
Metadata = new Dictionary<string, JsonElement> { ["volcanic.hops"] = JsonSerializer.SerializeToElement(inboundHops() + 1) },
|
||||
}, ct).ConfigureAwait(false);
|
||||
|
||||
if (response.Task is not { } task)
|
||||
{
|
||||
var message = response.Message;
|
||||
return $"Agent \"{entry.Name}\" replied (context {message?.ContextId}):\n{A2AResults.TextOf(message)}";
|
||||
}
|
||||
task = await WaitAsync(agent, task, ct).ConfigureAwait(false);
|
||||
if (A2AResults.IsSettled(task.Status.State)) return DescribeSettled(workspace, entry.Name, task);
|
||||
await registry.StartAsync(target, workspace, agent, task, settings, ct).ConfigureAwait(false);
|
||||
return A2AResults.Describe(entry.Name, task, []) +
|
||||
"\nIt is still running: the result will be delivered to this conversation when it is ready. You can also check it with action=status.";
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<string> StatusAsync(string workspace, A2AArgs args, CancellationToken ct)
|
||||
{
|
||||
var entry = Entry(workspace, args.Agent);
|
||||
var taskId = Required(args.TaskId, "taskId");
|
||||
return await CallAsync(entry, ct, async agent =>
|
||||
{
|
||||
var task = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
|
||||
return A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []);
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<string> CancelAsync(string workspace, A2AArgs args, CancellationToken ct)
|
||||
{
|
||||
var entry = Entry(workspace, args.Agent);
|
||||
var taskId = Required(args.TaskId, "taskId");
|
||||
// First, so a "canceled" outcome is never delivered as a follow-up.
|
||||
await registry.StopWatchingAsync(target, workspace, taskId).ConfigureAwait(false);
|
||||
return await CallAsync(entry, ct, async agent =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var task = await agent.Client.CancelTaskAsync(new CancelTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
|
||||
return A2AResults.Describe(entry.Name, task, []);
|
||||
}
|
||||
catch (A2AException ex)
|
||||
{
|
||||
// Typically: already finished. Its watch is gone, so hand over the outcome here.
|
||||
var task = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
|
||||
return $"Could not cancel ({Clean(ex.Message, entry)}). " +
|
||||
(A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []));
|
||||
}
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private string DescribeSettled(string workspace, string agent, AgentTask task)
|
||||
{
|
||||
var saved = A2AResults.SaveFiles(workspace, agent, task, settings.MaxFileBytes);
|
||||
return A2AResults.Describe(agent, task, saved.Saved, saved.Skipped);
|
||||
}
|
||||
|
||||
/// <summary>Wait up to WaitSeconds for the task to settle: subscribe (streaming agents) or poll every second.</summary>
|
||||
private async Task<AgentTask> WaitAsync(RemoteAgent agent, AgentTask task, CancellationToken ct)
|
||||
{
|
||||
if (A2AResults.IsSettled(task.Status.State) || settings.WaitSeconds <= 0) return task;
|
||||
using var wait = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
wait.CancelAfter(TimeSpan.FromSeconds(settings.WaitSeconds));
|
||||
try
|
||||
{
|
||||
if (agent.Streaming)
|
||||
{
|
||||
AgentTask current = task;
|
||||
await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false))
|
||||
{
|
||||
current = TaskProjection.Apply(current, ev) ?? current;
|
||||
if (A2AResults.IsSettled(current.Status.State)) return current;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(1), registry.Time, wait.Token).ConfigureAwait(false);
|
||||
var latest = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false);
|
||||
if (A2AResults.IsSettled(latest.Status.State)) return latest;
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (!ct.IsCancellationRequested) { } // waited long enough
|
||||
catch (A2AException) { } // e.g. subscribe on a task that just finished
|
||||
return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = task.Id }, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<string> CallAsync(RemoteAgentEntry entry, CancellationToken ct, Func<RemoteAgent, Task<string>> call)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var agent = await registry.Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false);
|
||||
return await call(agent).ConfigureAwait(false);
|
||||
}
|
||||
catch (HttpRequestException ex) when (ex.StatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden)
|
||||
{
|
||||
throw new InvalidOperationException($"Agent \"{entry.Name}\" rejected the key ({(int)ex.StatusCode!}). Check its key in .volcanic/a2a/agents.json.");
|
||||
}
|
||||
catch (HttpRequestException ex)
|
||||
{
|
||||
throw new InvalidOperationException($"Could not reach agent \"{entry.Name}\": {Clean(ex.Message, entry)}");
|
||||
}
|
||||
catch (A2AException ex)
|
||||
{
|
||||
throw new InvalidOperationException($"Agent \"{entry.Name}\" refused: {Clean(ex.Message, entry)} ({ex.ErrorCode}).");
|
||||
}
|
||||
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
||||
{
|
||||
throw new InvalidOperationException($"Agent \"{entry.Name}\" did not answer in time.");
|
||||
}
|
||||
}
|
||||
|
||||
private static RemoteAgentEntry Entry(string workspace, string? name)
|
||||
{
|
||||
var file = AgentsFile.Read(workspace);
|
||||
if (file.Error is not null) throw new InvalidOperationException(file.Error);
|
||||
if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("agent is required.");
|
||||
return file.Agents.FirstOrDefault(a => a.Name == name)
|
||||
?? throw new ArgumentException($"Unknown agent \"{name}\". Configured: {(file.Agents.Count == 0 ? "none (see .volcanic/a2a/agents.json)" : string.Join(", ", file.Agents.Select(a => a.Name)))}.");
|
||||
}
|
||||
|
||||
private Part FilePart(string workspace, string path)
|
||||
{
|
||||
var full = ConfinedPath(workspace, path);
|
||||
var info = new FileInfo(full);
|
||||
if (!info.Exists) throw new ArgumentException($"{path} does not exist.");
|
||||
if (info.Length > settings.MaxFileBytes) throw new ArgumentException($"{path} is larger than {settings.MaxFileBytes} bytes.");
|
||||
return Part.FromRaw(File.ReadAllBytes(full), MediaType(full), Path.GetFileName(full));
|
||||
}
|
||||
|
||||
/// <summary>The real path of a workspace file: inside the workspace lexically and after every symlink is resolved.</summary>
|
||||
internal static string ConfinedPath(string workspace, string path)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("a file path is empty.");
|
||||
var root = Path.GetFullPath(workspace);
|
||||
var full = Path.GetFullPath(Path.Combine(root, path));
|
||||
if (!IsInside(full, root)) throw new ArgumentException($"{path} is outside the workspace.");
|
||||
if (!IsInside(RealPath(full, 0), RealPath(root, 0))) throw new ArgumentException($"{path} points outside the workspace.");
|
||||
return full;
|
||||
}
|
||||
|
||||
private static bool IsInside(string path, string root)
|
||||
{
|
||||
var prefix = root.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar;
|
||||
return path.StartsWith(prefix, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
/// <summary>Resolves symlinks in every existing component of an absolute path.</summary>
|
||||
private static string RealPath(string full, int depth)
|
||||
{
|
||||
if (depth > MaxLinkDepth) throw new ArgumentException("too many symbolic links.");
|
||||
var current = Path.GetPathRoot(full)!;
|
||||
foreach (var segment in full[current.Length..].Split(Path.DirectorySeparatorChar, StringSplitOptions.RemoveEmptyEntries))
|
||||
{
|
||||
var next = Path.Combine(current, segment);
|
||||
FileSystemInfo info = Directory.Exists(next) ? new DirectoryInfo(next) : new FileInfo(next);
|
||||
// A link's target may itself sit under links: resolve it again from the top.
|
||||
if (info.LinkTarget is not null && info.ResolveLinkTarget(true) is { } resolved) next = RealPath(resolved.FullName, depth + 1);
|
||||
current = next;
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
private static string MediaType(string path) => Path.GetExtension(path).ToLowerInvariant() switch
|
||||
{
|
||||
".png" => "image/png",
|
||||
".jpg" or ".jpeg" => "image/jpeg",
|
||||
".pdf" => "application/pdf",
|
||||
".txt" => "text/plain",
|
||||
".md" => "text/markdown",
|
||||
".json" => "application/json",
|
||||
".csv" => "text/csv",
|
||||
_ => "application/octet-stream",
|
||||
};
|
||||
|
||||
private static string Required(string? value, string name) =>
|
||||
string.IsNullOrWhiteSpace(value) ? throw new ArgumentException($"{name} is required.") : value;
|
||||
|
||||
/// <summary>Exception texts must never echo a key.</summary>
|
||||
private static string Clean(string text, RemoteAgentEntry entry) =>
|
||||
entry.Key is { Length: > 0 } key ? text.Replace(key, "[key]", StringComparison.Ordinal) : text;
|
||||
}
|
||||
Reference in New Issue
Block a user