From 5af95fdb6c0b3de71152fee8b88afc78866c6a86 Mon Sep 17 00:00:00 2001 From: Emil Date: Sat, 18 Jul 2026 23:42:01 +0300 Subject: [PATCH] feat: add Kimi K3 fleet MCP orchestrator and dashboard --- .gitignore | 4 + Dashboard.html | 14 ++ KimiFleet.csproj | 11 ++ McpStdioServer.cs | 107 +++++++++++ Program.cs | 459 ++++++++++++++++++++++++++++++++++++++++++++++ README.md | 57 ++++++ 6 files changed, 652 insertions(+) create mode 100644 .gitignore create mode 100644 Dashboard.html create mode 100644 KimiFleet.csproj create mode 100644 McpStdioServer.cs create mode 100644 Program.cs create mode 100644 README.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..547a964 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +bin/ +obj/ +.vs/ +*.user diff --git a/Dashboard.html b/Dashboard.html new file mode 100644 index 0000000..5e6b84e --- /dev/null +++ b/Dashboard.html @@ -0,0 +1,14 @@ + +KimiFleet +

KimiFleet

MCP-controlled Kimi K3 agent fleet
LIVE

Agents

Loading agents…

Activity

Waiting for events…
+ diff --git a/KimiFleet.csproj b/KimiFleet.csproj new file mode 100644 index 0000000..f7cf9e4 --- /dev/null +++ b/KimiFleet.csproj @@ -0,0 +1,11 @@ + + + Exe + net9.0 + enable + enable + + + + + diff --git a/McpStdioServer.cs b/McpStdioServer.cs new file mode 100644 index 0000000..12fdb67 --- /dev/null +++ b/McpStdioServer.cs @@ -0,0 +1,107 @@ +using System.Text.Json; + +sealed class McpStdioServer(FleetHost fleet) +{ + private static readonly JsonSerializerOptions Json = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + public async Task RunAsync() + { + string? line; + while ((line = await Console.In.ReadLineAsync()) is not null) + { + try + { + using var document = JsonDocument.Parse(line); + var request = document.RootElement; + if (!request.TryGetProperty("method", out var methodNode)) continue; + var method = methodNode.GetString(); + if (!request.TryGetProperty("id", out var id)) continue; // notifications have no response + object result = method switch + { + "initialize" => new + { + protocolVersion = "2025-03-26", + capabilities = new { tools = new { } }, + serverInfo = new { name = "KimiFleet", version = "0.1.0" }, + instructions = "Use KimiFleet tools to create scoped Kimi K3 agents, assign work, and require peer review." + }, + "tools/list" => new { tools = Tools }, + "tools/call" => await CallToolAsync(request.GetProperty("params")), + _ => throw new InvalidOperationException($"Unsupported MCP method '{method}'.") + }; + await SendAsync(id, result); + } + catch (Exception ex) + { + Console.Error.WriteLine($"[mcp] {ex.Message}"); + await Console.Out.WriteLineAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0", error = new { code = -32603, message = ex.Message } }, Json)); + await Console.Out.FlushAsync(); + } + } + } + + private async Task CallToolAsync(JsonElement call) + { + var name = call.GetProperty("name").GetString() ?? throw new InvalidOperationException("Tool call has no name."); + var arguments = call.TryGetProperty("arguments", out var value) ? value : default; + object payload = name switch + { + "fleet_list_agents" => fleet.ListAgents(), + "fleet_start_agent" => await fleet.StartAgentAsync(new FleetHost.AgentSpec( + Required(arguments, "name"), Required(arguments, "workspace"), + Optional(arguments, "role"), StringArray(arguments, "scopes"))), + "fleet_assign_task" => Assign(Required(arguments, "agent"), Required(arguments, "prompt")), + "fleet_request_review" => Review(Required(arguments, "reviewer"), Required(arguments, "author"), Optional(arguments, "context")), + "fleet_stop_agent" => Stop(Required(arguments, "agent")), + _ => throw new InvalidOperationException($"Unknown KimiFleet tool '{name}'.") + }; + return new { content = new[] { new { type = "text", text = JsonSerializer.Serialize(payload, Json) } } }; + } + + private object Assign(string agent, string prompt) + { + fleet.Assign(agent, prompt); + return new { accepted = true, agent }; + } + + private object Review(string reviewer, string author, string? context) + { + fleet.Review(reviewer, author, context); + return new { accepted = true, reviewer, author }; + } + + private object Stop(string agent) + { + fleet.Stop(agent); + return new { stopped = agent }; + } + + private static string Required(JsonElement input, string name) => + input.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.String && !string.IsNullOrWhiteSpace(value.GetString()) + ? value.GetString()! : throw new InvalidOperationException($"'{name}' is required."); + + private static string? Optional(JsonElement input, string name) => + input.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.String ? value.GetString() : null; + + private static string[]? StringArray(JsonElement input, string name) => + input.TryGetProperty(name, out var values) && values.ValueKind == JsonValueKind.Array + ? values.EnumerateArray().Where(x => x.ValueKind == JsonValueKind.String).Select(x => x.GetString()!).ToArray() : null; + + private static async Task SendAsync(JsonElement id, object result) + { + var json = $"{{\"jsonrpc\":\"2.0\",\"id\":{id.GetRawText()},\"result\":{JsonSerializer.Serialize(result, Json)}}}"; + await Console.Out.WriteLineAsync(json); + await Console.Out.FlushAsync(); + } + + private static readonly object[] Tools = + [ + Tool("fleet_list_agents", "List active Kimi K3 agents, their roles, scope locks and state.", new { type = "object", properties = new { } }), + Tool("fleet_start_agent", "Start a yolo Kimi K3 ACP agent with exclusive source scopes.", new { type = "object", required = new[] { "name", "workspace" }, properties = new { name = new { type = "string" }, workspace = new { type = "string" }, role = new { type = "string" }, scopes = new { type = "array", items = new { type = "string" } } } }), + Tool("fleet_assign_task", "Assign an autonomous task to an active Kimi agent.", new { type = "object", required = new[] { "agent", "prompt" }, properties = new { agent = new { type = "string" }, prompt = new { type = "string" } } }), + Tool("fleet_request_review", "Ask a second agent for read-only peer review of an author's current diff.", new { type = "object", required = new[] { "reviewer", "author" }, properties = new { reviewer = new { type = "string" }, author = new { type = "string" }, context = new { type = "string" } } }), + Tool("fleet_stop_agent", "Stop an agent and release its exclusive scopes.", new { type = "object", required = new[] { "agent" }, properties = new { agent = new { type = "string" } } }) + ]; + + private static object Tool(string name, string description, object inputSchema) => new { name, description, inputSchema }; +} diff --git a/Program.cs b/Program.cs new file mode 100644 index 0000000..50d3606 --- /dev/null +++ b/Program.cs @@ -0,0 +1,459 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Net; +using System.Text; +using System.Text.Json; +using System.Text.Json.Nodes; +using System.Threading.Channels; + +var options = FleetOptions.Parse(args); +var fleet = new FleetHost(options); +Console.CancelKeyPress += (_, e) => +{ + e.Cancel = true; + fleet.Dispose(); + Environment.Exit(0); +}; + +Console.Error.WriteLine($"KimiFleet dashboard: http://127.0.0.1:{options.Port}"); +var webTask = fleet.RunAsync(); +if (options.EnableMcp) + await Task.WhenAny(webTask, new McpStdioServer(fleet).RunAsync()); +else + await webTask; + +sealed record FleetOptions(int Port, string KimiExecutable, string Model, bool EnableMcp) +{ + public static FleetOptions Parse(string[] args) + { + var port = 7373; + var kimi = "kimi"; + var model = "kimi-code/k3"; + var enableMcp = false; + for (var i = 0; i < args.Length; i++) + { + if (args[i] == "--port" && i + 1 < args.Length && int.TryParse(args[++i], out var value)) port = value; + else if (args[i] == "--kimi" && i + 1 < args.Length) kimi = args[++i]; + else if (args[i] == "--model" && i + 1 < args.Length) model = args[++i]; + else if (args[i] == "--mcp") enableMcp = true; + } + return new FleetOptions(port, kimi, model, enableMcp); + } +} + +sealed class FleetHost : IDisposable +{ + private readonly FleetOptions _options; + private readonly HttpListener _listener = new(); + private readonly ConcurrentDictionary _agents = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary _scopeOwners = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary> _eventSubscribers = new(); + private readonly List _recentEvents = []; + private readonly object _eventsLock = new(); + private bool _disposed; + + public FleetHost(FleetOptions options) + { + _options = options; + _listener.Prefixes.Add($"http://127.0.0.1:{options.Port}/"); + } + + public async Task RunAsync() + { + _listener.Start(); + while (!_disposed) + { + HttpListenerContext context; + try { context = await _listener.GetContextAsync(); } + catch (ObjectDisposedException) { break; } + _ = Task.Run(() => HandleAsync(context)); + } + } + + private async Task HandleAsync(HttpListenerContext context) + { + try + { + var path = context.Request.Url?.AbsolutePath.Trim('/') ?? string.Empty; + var parts = path.Split('/', StringSplitOptions.RemoveEmptyEntries); + if (context.Request.HttpMethod == "GET" && path.Length == 0) + await ServeDashboardAsync(context); + else if (context.Request.HttpMethod == "GET" && path == "health") + await RespondJsonAsync(context, new { ok = true, model = _options.Model }); + else if (context.Request.HttpMethod == "GET" && path == "agents") + await RespondJsonAsync(context, _agents.Values.Select(a => a.Snapshot())); + else if (context.Request.HttpMethod == "GET" && path == "events") + await StreamEventsAsync(context); + else if (context.Request.HttpMethod == "POST" && path == "agents") + await StartAgentAsync(context); + else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "prompt") + await PromptAgentAsync(context, parts[1]); + else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "review") + await ReviewAsync(context, parts[1]); + else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "stop") + await StopAgentAsync(context, parts[1]); + else + await RespondJsonAsync(context, new { error = "Unknown endpoint." }, 404); + } + catch (Exception ex) + { + Publish("fleet", "error", ex.Message); + if (context.Response.OutputStream.CanWrite) + await RespondJsonAsync(context, new { error = ex.Message }, 500); + } + } + + private async Task StartAgentAsync(HttpListenerContext context) + { + var request = await ReadJsonAsync(context); + var agent = await StartAgentAsync(new AgentSpec(request.Name, request.Workspace, request.Role, request.Scopes)); + await RespondJsonAsync(context, agent, 201); + } + + private async Task PromptAgentAsync(HttpListenerContext context, string name) + { + var request = await ReadJsonAsync(context); + Assign(name, request.Prompt); + await RespondJsonAsync(context, new { accepted = true, agent = name }); + } + + private async Task ReviewAsync(HttpListenerContext context, string reviewerName) + { + var request = await ReadJsonAsync(context); + Review(reviewerName, request.Author, request.Context); + await RespondJsonAsync(context, new { accepted = true, reviewer = reviewerName, author = request.Author }); + } + + private async Task StopAgentAsync(HttpListenerContext context, string name) + { + RemoveAgent(name); + await RespondJsonAsync(context, new { stopped = name }); + } + + public async Task StartAgentAsync(AgentSpec spec) + { + var name = ValidateName(spec.Name); + var workspace = Path.GetFullPath(spec.Workspace); + if (!Directory.Exists(workspace)) throw new InvalidOperationException($"Workspace does not exist: {workspace}"); + if (_agents.ContainsKey(name)) throw new InvalidOperationException($"Agent '{name}' already exists."); + var scopes = spec.Scopes?.Distinct(StringComparer.OrdinalIgnoreCase).ToArray() ?? []; + foreach (var scope in scopes) + if (_scopeOwners.TryGetValue(scope, out var owner)) throw new InvalidOperationException($"Scope '{scope}' is owned by '{owner}'."); + var agent = new KimiAgent(name, workspace, spec.Role ?? "worker", scopes, _options, Publish); + if (!_agents.TryAdd(name, agent)) throw new InvalidOperationException($"Could not add '{name}'."); + foreach (var scope in scopes) _scopeOwners[scope] = name; + try { await agent.StartAsync(); return agent.Snapshot(); } + catch { RemoveAgent(name); throw; } + } + + public IReadOnlyList ListAgents() => _agents.Values.Select(a => a.Snapshot()).Cast().ToArray(); + public void Assign(string name, string prompt) => _ = GetAgent(name).PromptAsync(prompt); + public void Review(string reviewerName, string authorName, string? context) + { + var reviewer = GetAgent(reviewerName); + var author = GetAgent(authorName); + _ = reviewer.PromptAsync($"Perform peer review for agent '{author.Name}'. Review only; do not edit files. Inspect current git diff and changed files. Check correctness, resource safety, tests, and scope ownership. Report findings by severity with file/line locations. Context: {context ?? "No additional context."}"); + } + public void Stop(string name) => RemoveAgent(name); + + private KimiAgent GetAgent(string name) => + _agents.TryGetValue(name, out var agent) ? agent : throw new InvalidOperationException($"Unknown agent '{name}'."); + + private void RemoveAgent(string name) + { + if (!_agents.TryRemove(name, out var agent)) return; + foreach (var scope in agent.Scopes) _scopeOwners.TryRemove(scope, out _); + agent.Dispose(); + Publish(name, "stopped", "Agent stopped."); + } + + private static string ValidateName(string? name) + { + if (string.IsNullOrWhiteSpace(name) || name.Any(c => !char.IsAsciiLetterOrDigit(c) && c is not '-' and not '_')) + throw new InvalidOperationException("Agent name must contain only ASCII letters, digits, '-' or '_'."); + return name; + } + + private async Task ReadJsonAsync(HttpListenerContext context) + { + using var reader = new StreamReader(context.Request.InputStream, context.Request.ContentEncoding); + var value = await JsonSerializer.DeserializeAsync(reader.BaseStream, JsonOptions); + return value ?? throw new InvalidOperationException("Request body must be valid JSON."); + } + + private static async Task RespondJsonAsync(HttpListenerContext context, object body, int status = 200) + { + var bytes = JsonSerializer.SerializeToUtf8Bytes(body, JsonOptions); + context.Response.StatusCode = status; + context.Response.ContentType = "application/json; charset=utf-8"; + context.Response.ContentLength64 = bytes.Length; + await context.Response.OutputStream.WriteAsync(bytes); + context.Response.Close(); + } + + private static async Task ServeDashboardAsync(HttpListenerContext context) + { + var html = await File.ReadAllBytesAsync(Path.Combine(AppContext.BaseDirectory, "Dashboard.html")); + context.Response.ContentType = "text/html; charset=utf-8"; + context.Response.ContentLength64 = html.Length; + await context.Response.OutputStream.WriteAsync(html); + context.Response.Close(); + } + + private async Task StreamEventsAsync(HttpListenerContext context) + { + context.Response.StatusCode = 200; + context.Response.ContentType = "text/event-stream"; + context.Response.SendChunked = true; + context.Response.Headers.Add("Cache-Control", "no-cache"); + var id = Guid.NewGuid(); + var channel = Channel.CreateUnbounded(); + _eventSubscribers[id] = channel; + try + { + string[] snapshot; + lock (_eventsLock) snapshot = _recentEvents.ToArray(); + foreach (var line in snapshot) await WriteSseAsync(context.Response, line); + await foreach (var line in channel.Reader.ReadAllAsync()) await WriteSseAsync(context.Response, line); + } + catch (HttpListenerException) { } + catch (IOException) { } + finally { _eventSubscribers.TryRemove(id, out _); context.Response.Close(); } + } + + private static async Task WriteSseAsync(HttpListenerResponse response, string json) + { + var data = Encoding.UTF8.GetBytes($"data: {json}\n\n"); + await response.OutputStream.WriteAsync(data); + await response.OutputStream.FlushAsync(); + } + + private void Publish(string agent, string kind, string message) + { + var json = JsonSerializer.Serialize(new { time = DateTimeOffset.UtcNow, agent, kind, message }, JsonOptions); + lock (_eventsLock) + { + _recentEvents.Add(json); + if (_recentEvents.Count > 250) _recentEvents.RemoveAt(0); + } + Console.Error.WriteLine($"[{DateTime.Now:HH:mm:ss}] [{agent}] {kind}: {DescribeEvent(kind, message)}"); + foreach (var subscriber in _eventSubscribers.Values) subscriber.Writer.TryWrite(json); + } + + private static string DescribeEvent(string kind, string message) + { + if (kind != "session/update") return message; + try + { + using var document = JsonDocument.Parse(message); + var update = document.RootElement.GetProperty("update"); + var type = update.GetProperty("sessionUpdate").GetString(); + if (type == "tool_call_update") return $"{update.GetProperty("title").GetString()} — {update.GetProperty("status").GetString()}"; + if (type == "agent_message_chunk") return "Kimi sent a response"; + return type ?? "session update"; + } + catch (JsonException) { return "session update"; } + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _listener.Close(); + foreach (var name in _agents.Keys.ToArray()) RemoveAgent(name); + foreach (var subscriber in _eventSubscribers.Values) subscriber.Writer.TryComplete(); + } + + private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + public sealed record AgentSpec(string Name, string Workspace, string? Role, string[]? Scopes); + private sealed record StartAgentRequest(string Name, string Workspace, string? Role, string[]? Scopes); + private sealed record PromptRequest(string Prompt); + private sealed record ReviewRequest(string Author, string? Context); +} + +sealed class KimiAgent : IDisposable +{ + private readonly FleetOptions _options; + private readonly Action _publish; + private readonly Process _process; + private readonly ConcurrentDictionary> _pending = new(); + private readonly SemaphoreSlim _stdinLock = new(1, 1); + private long _requestId; + private string? _sessionId; + private bool _disposed; + + public string Name { get; } + public string Workspace { get; } + public string Role { get; } + public IReadOnlyList Scopes { get; } + public string State { get; private set; } = "starting"; + + public KimiAgent(string name, string workspace, string role, IReadOnlyList scopes, FleetOptions options, Action publish) + { + Name = name; + Workspace = workspace; + Role = role; + Scopes = scopes; + _options = options; + _publish = publish; + _process = new Process + { + StartInfo = new ProcessStartInfo + { + FileName = options.KimiExecutable, + WorkingDirectory = workspace, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + CreateNoWindow = true + }, + EnableRaisingEvents = true + }; + _process.StartInfo.ArgumentList.Add("--yolo"); + _process.StartInfo.ArgumentList.Add("--model"); + _process.StartInfo.ArgumentList.Add(options.Model); + _process.StartInfo.ArgumentList.Add("acp"); + } + + public async Task StartAsync() + { + if (!_process.Start()) throw new InvalidOperationException($"Could not start Kimi process for '{Name}'."); + _process.Exited += (_, _) => { State = "stopped"; _publish(Name, "exit", $"Kimi exited with {_process.ExitCode}."); }; + _ = ReadOutputAsync(); + _ = ReadErrorAsync(); + var initialized = await RequestAsync("initialize", new + { + protocolVersion = 1, + clientCapabilities = new { fs = new { readTextFile = true, writeTextFile = true } }, + clientInfo = new { name = "KimiFleet", version = "0.1.0" } + }); + _publish(Name, "initialized", initialized.GetProperty("agentInfo").GetProperty("name").GetString() ?? "Kimi Code CLI"); + var session = await RequestAsync("session/new", new { cwd = Workspace, mcpServers = Array.Empty() }); + _sessionId = session.GetProperty("sessionId").GetString() ?? throw new InvalidOperationException("ACP did not return sessionId."); + State = "ready"; + _publish(Name, "ready", $"Role={Role}; scopes={string.Join(", ", Scopes)}"); + } + + public async Task PromptAsync(string prompt) + { + if (string.IsNullOrWhiteSpace(prompt)) throw new InvalidOperationException("Prompt must not be empty."); + if (_sessionId is null) throw new InvalidOperationException($"Agent '{Name}' has no ACP session."); + State = "working"; + _publish(Name, "prompt", prompt.Length > 160 ? prompt[..160] + "…" : prompt); + try + { + await RequestAsync("session/prompt", new { sessionId = _sessionId, prompt = new[] { new { type = "text", text = prompt } } }); + State = "ready"; + _publish(Name, "completed", "Prompt completed."); + } + catch (Exception ex) + { + State = "error"; + _publish(Name, "error", ex.Message); + } + } + + private async Task RequestAsync(string method, object parameters) + { + var id = Interlocked.Increment(ref _requestId); + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + if (!_pending.TryAdd(id, completion)) throw new InvalidOperationException("Could not register ACP request."); + var envelope = JsonSerializer.Serialize(new { jsonrpc = "2.0", id, method, @params = parameters }); + await SendRawAsync(envelope); + using var timeout = new CancellationTokenSource(TimeSpan.FromMinutes(10)); + await using var registration = timeout.Token.Register(() => completion.TrySetException(new TimeoutException($"ACP request '{method}' timed out."))); + return await completion.Task; + } + + private async Task ReadOutputAsync() + { + while (!_process.HasExited) + { + var line = await _process.StandardOutput.ReadLineAsync(); + if (line is null) break; + HandleMessage(line); + } + } + + private async Task ReadErrorAsync() + { + while (!_process.HasExited) + { + var line = await _process.StandardError.ReadLineAsync(); + if (line is null) break; + _publish(Name, "stderr", line); + } + } + + private void HandleMessage(string line) + { + try + { + using var document = JsonDocument.Parse(line); + var root = document.RootElement; + if (root.TryGetProperty("id", out var idNode) && idNode.TryGetInt64(out var id) && _pending.TryRemove(id, out var completion)) + { + if (root.TryGetProperty("error", out var error)) completion.TrySetException(new InvalidOperationException(error.GetRawText())); + else if (root.TryGetProperty("result", out var result)) completion.TrySetResult(result.Clone()); + else completion.TrySetException(new InvalidOperationException("ACP response has neither result nor error.")); + return; + } + if (root.TryGetProperty("method", out var method)) + { + var methodName = method.GetString() ?? "notification"; + if (methodName == "session/request_permission" && root.TryGetProperty("id", out var permissionId) && root.TryGetProperty("params", out var permission)) + { + var optionId = SelectApproval(permission); + _ = RespondToPermissionAsync(permissionId.GetRawText(), optionId); + _publish(Name, "permission", $"Auto-approved '{optionId}' (yolo mode)."); + return; + } + _publish(Name, methodName, root.TryGetProperty("params", out var p) ? p.GetRawText() : string.Empty); + } + else _publish(Name, "protocol", line); + } + catch (JsonException) { _publish(Name, "stdout", line); } + } + + private static string SelectApproval(JsonElement permission) + { + if (!permission.TryGetProperty("options", out var options) || options.ValueKind != JsonValueKind.Array) + throw new InvalidOperationException("ACP permission request has no options."); + var ids = options.EnumerateArray() + .Where(option => option.TryGetProperty("optionId", out _)) + .Select(option => option.GetProperty("optionId").GetString()) + .Where(id => !string.IsNullOrWhiteSpace(id)) + .Cast() + .ToArray(); + return ids.FirstOrDefault(id => id.Contains("approve_always", StringComparison.OrdinalIgnoreCase)) + ?? ids.FirstOrDefault(id => id.Contains("approve", StringComparison.OrdinalIgnoreCase)) + ?? throw new InvalidOperationException("ACP permission request has no approval option."); + } + + private Task RespondToPermissionAsync(string requestId, string optionId) => + SendRawAsync($"{{\"jsonrpc\":\"2.0\",\"id\":{requestId},\"result\":{{\"outcome\":{{\"outcome\":\"selected\",\"optionId\":{JsonSerializer.Serialize(optionId)}}}}}}}"); + + private async Task SendRawAsync(string json) + { + await _stdinLock.WaitAsync(); + try + { + await _process.StandardInput.WriteLineAsync(json); + await _process.StandardInput.FlushAsync(); + } + finally { _stdinLock.Release(); } + } + + public object Snapshot() => new { name = Name, workspace = Workspace, role = Role, scopes = Scopes, state = State, sessionId = _sessionId }; + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + try { if (!_process.HasExited) _process.Kill(entireProcessTree: true); } catch (InvalidOperationException) { } + _stdinLock.Dispose(); + _process.Dispose(); + } +} diff --git a/README.md b/README.md new file mode 100644 index 0000000..43bb25d --- /dev/null +++ b/README.md @@ -0,0 +1,57 @@ +# KimiFleet + +Visible multi-agent orchestrator and MCP server for Kimi Code K3. It controls Kimi through ACP (JSON-RPC over stdio), rather than attempting to drive its terminal UI. + +## Start + +```bash +dotnet run --project KimiFleet.csproj -- --port 7373 +``` + +Open a second terminal and watch live activity: + +```bash +curl -N http://127.0.0.1:7373/events +``` + +Open `http://127.0.0.1:7373` for the dashboard: it shows agents, their scoped ownership, state and a human-readable live activity feed. + +## MCP mode + +Start both the dashboard and an MCP stdio server: + +```bash +dotnet run --project KimiFleet.csproj -- --mcp --port 7373 +``` + +The server implements `fleet_list_agents`, `fleet_start_agent`, `fleet_assign_task`, `fleet_request_review`, and `fleet_stop_agent`. In MCP mode stdout is protocol-only; operational logs are written to stderr and remain visible in the dashboard. + +## API + +Create an agent with exclusive source scopes: + +```bash +curl -X POST http://127.0.0.1:7373/agents \ + -H 'Content-Type: application/json' \ + -d '{"name":"vulkan","workspace":"/home/emil/Desktop/Cortex_Engine","role":"renderer","scopes":["src/Engine.Graphics.Vulkan","src/Engine.Graphics"]}' +``` + +Give it work: + +```bash +curl -X POST http://127.0.0.1:7373/agents/vulkan/prompt \ + -H 'Content-Type: application/json' \ + -d '{"prompt":"Read AGENTS.md, inspect the Vulkan backend, and propose one safe improvement."}' +``` + +Ask another agent to review the current diff without editing: + +```bash +curl -X POST http://127.0.0.1:7373/agents/reviewer/review \ + -H 'Content-Type: application/json' \ + -d '{"author":"vulkan","context":"Review the Vulkan change after its tests finish."}' +``` + +Useful endpoints: `GET /health`, `GET /agents`, `GET /events`, `POST /agents/{name}/stop`. + +All ACP children launch as `kimi --yolo --model kimi-code/k3 acp`. Scope locks prevent two fleet agents from being assigned the same area, but they do not replace a final `git status` check before edits.