Universal.Operative.Sdk 3.2.0

About

Universal.Operative.Sdk is a provider-neutral framework for building long-running, agentic LLM-driven applications. It defines the domain model (Conversation, Message, content Blocks), the extension seams (IModel, ITool, ITransformer), and two complementary building blocks that sit on top of them: an Engine, which drives a single agentic turn end-to-end (stream a model response, dispatch any tool calls, feed results back, loop until the model stops), and an Operative, which wraps an Engine around one ongoing Conversation and decides whether a new inbound message starts a fresh turn or steers into one already running. Pair it with Universal.Operative.Sdk.Anthropic or Universal.Operative.Sdk.OpenAI for a concrete IModel.

How to Use

Installation

dotnet add package Universal.Operative.Sdk

Quick start: a single turn

Use Engine directly when there's no ongoing session to manage — a one-shot request/response, a batch job, a single CLI invocation.

using Universal.Operative.Sdk;

IModel model = /* an IModel, e.g. from Universal.Operative.Sdk.Anthropic or .OpenAI */;

IEngine engine = EngineBuilder.FromDefaults(model)
    .AddBaselineTools()
    .Build();

var conversation = new Conversation
{
    System = new List<Block> { new TextBlock("You are a helpful assistant.") },
    Messages = { new Message { Role = MessageRoles.User, Content = { new TextBlock("List the files in this repo.") } } },
};

await foreach (EngineEvent e in engine.ExecuteAsync(conversation))
{
    switch (e)
    {
        case ModelStreamEvent s: break;
        case AssistantMessageEvent a: Console.WriteLine(a.Message.Content); break;
        case ToolCallStartedEvent t: Console.WriteLine($"Running {t.Call.Name}..."); break;
        case TurnCompletedEvent done: Console.WriteLine($"Done: {done.Reason}"); break;
    }
}

AddBaselineTools() registers Read, Write, Edit, Glob, Grep, a shell (Bash and/or PowerShell, by availability), TodoWrite, and NotebookEdit. EngineBuilder.FromDefaults wires a pre-transformer stack (boundary-aware history windowing, tool-result budgeting, history snipping, auto-compaction) so long-running conversations don't blow the model's context window; use new EngineBuilder() instead to opt into transformers individually. conversation.Messages holds the full transcript once the turn ends, including tool calls and results.

Quick start: a long-running session

Use IOperative when something outside your control — a chat UI, a Slack app, a webhook — can send a new message while a turn is already running, and you want that message to steer the in-progress turn instead of queuing behind it.

IOperative bot = OperativeBuilder.Create(engine, conversation).Build();

SubmitResult result = await bot.SubmitAsync(
    new Message { Role = MessageRoles.User, Content = { new TextBlock("hi") } });

await foreach (OperativeEvent e in result.Events)
{
    if (e is EngineEventOccurred occurred)
    {
        // same EngineEvent types as the IEngine example above
    }
}

Whether this call started the turn (result.Outcome == SubmitOutcome.StartedTurn) or steered a message into one already running (SubmitOutcome.Injected), result.Events is the same live remainder of that turn either way — one shared consciousness: every caller who touched a turn sees its real outcome, not just whoever happened to start it. Call bot.ResumeAsync() to continue the conversation with no new input at all — e.g. after a restart, when the loaded conversation already ends in an unanswered message; if a turn is already running this joins it too (SubmitOutcome.AlreadyRunning) rather than starting a redundant second one.

Two OperativeBuilder options worth knowing:

• .WithReloadBeforeStart(reload) — reload the conversation from durable storage immediately before each turn starts, so a previous turn that crashed mid-flight self-heals instead of wedging every future one. Runs only once a turn has actually been claimed, so it can't race a turn already in flight.
• .WithFrameInjectedMessage(frame) — customize (or disable) the note prefixed to a message before it's steered into someone else's in-progress turn.

Core concepts

• Conversation / Message / Block — the provider-neutral transcript. Message.Content is a list of Blocks (TextBlock, ToolCallBlock, ToolResultBlock, …); round-tripping a Conversation through JSON is enough to resume it later.
• IModel — the seam to a concrete provider. Implement StreamAsync to translate a Conversation into a provider request and its response stream into StreamEvents; CompleteAsync is provided for free as a fold over the stream.
• ITool / Tool<TArgs> — a callable the model can invoke. Subclass Tool<TArgs> with a typed argument record and a hand-authored JSON-Schema Definition; the base class handles (de)serialization and turns malformed arguments into a model-readable error instead of an exception. See Tools/BaselineTools.cs and Tools/FileSystem/GlobTool.cs for worked examples.
• ITurnContext — host-supplied identity/provenance for one turn (e.g. who sent the message that started it), deliberately opaque — the SDK prescribes nothing about what it carries, only its scope: established once by whichever call starts a turn, held for that turn's entire duration. A message steered into a turn already running does not get its own; the turn's original context still applies. Its only consumer is ITransformer.
• IExecutionContext — the caller-supplied configuration for one IEngine.ExecuteAsync call: an ITurnContext plus an optional mid-loop injection queue, bundled together since both are per-call inputs the engine reads from (as opposed to Conversation, the thing being acted on).
• ITransformer — rewrites the conversation (or the ambient ITurnContext) around a model call — compaction, tool-result pruning, history snipping. Receives an ITransformerContext (the current ITurnContext, plus Iteration/Stage — engine-owned loop-position facts it can read but never replace). Engine runs four stages: BeforeTurnTransformers once before the turn's first iteration, BeforeIterationTransformers before each model call, AfterIterationTransformers after the assistant message lands but before tools run, and AfterTurnTransformers once after the loop ends.
• Engine<TExecutionContext> / IEngine<TExecutionContext> — drives one turn to completion, exactly as shown above. Every step is a protected virtual method, so subclass to override one piece (tool-result formatting, dispatch policy, timeouts) without reimplementing the whole turn. Generic so a host needing ExecuteAsync strongly typed over its own IExecutionContext shape can subclass Engine<TMyExecutionContext> directly and inherit the whole loop; Engine/IEngine (non-generic, equivalent to Engine<IExecutionContext>/IEngine<IExecutionContext>) is what Operative and everything else in this SDK is built against, and what most hosts want. An Engine holds no session state of its own — one Engine (one model/tool/transformer configuration) is meant to be reused across many independent IOperative sessions, the same way one IModel is reused across many engines.
• OperativeBuilder / Operative / IOperative — owns one ongoing Conversation and single-flight-serialises turns against it through an Engine. SubmitAsync starts a fresh turn if none is running, or steers a new message into the turn already in flight if one is; ResumeAsync continues the conversation with no new input. Build one through OperativeBuilder, not the Operative constructor directly.
• EngineEvent — the observable channel for one turn, as driven directly by IEngine.ExecuteAsync: model stream events, the settled assistant message, tool start/complete/error events, the synthesized tool-result message, transformer applications, and a terminal TurnCompletedEvent.
• OperativeEvent — what IOperative.SubmitAsync/ResumeAsync's SubmitResult.Events actually yields. EngineEventOccurred relays an underlying EngineEvent as-is; MessageQueuedForNextTurnEvent is IOperative-only — it fires when a message arrives too late for the engine's own mid-turn drain to catch, so IOperative appends it directly instead of losing it, ready for the next turn to answer. Kept separate from EngineEvent because IEngine has no notion of IOperative and is fully usable without one.
• Session isolation — one IOperative is one conversation; every message submitted to it shares that same history and turn. If two channels or two users need to stay unaware of each other, give each its own IOperative (e.g. a Dictionary<string, IOperative> keyed by user/channel/thread) rather than routing everyone through one — the same IEngine (and its model/tools) can safely back any number of them at once, since IEngine holds no session state of its own.

Writing a custom tool

using System.Text.Json;
using System.Text.Json.Serialization;
using Universal.Operative.Sdk.Tools;

public sealed class WeatherTool : Tool<WeatherTool.Args>
{
    public override string Name => "get_weather";
    public override bool IsConcurrencySafe => true;

    public override ToolDefinition Definition { get; } = new()
    {
        Name = "get_weather",
        Description = "Get the current weather for a city.",
        Parameters = JsonDocument.Parse("""
            { "type": "object", "properties": { "city": { "type": "string" } }, "required": ["city"] }
            """).RootElement.Clone(),
    };

    public sealed record Args
    {
        [JsonPropertyName("city")] public string City { get; init; } = string.Empty;
    }

    protected override async Task<ToolResult> ExecuteAsync(Args args, CancellationToken cancellationToken)
        => ToolResult.Json(new { city = args.City, tempF = 72 });
}

IsConcurrencySafe = true marks a read-only tool as safe to run in parallel with other calls in the same turn; mutating tools default to false and run serially in call order. Register a tool with builder.AddTool(new WeatherTool()) and the harness auto-publishes its Definition on outbound messages and dispatches calls to it by name.

Background tool execution

A tool whose work can outlive a single call implements IBackgroundTool (most conveniently via BackgroundTool<TArgs>, which mirrors Tool<TArgs>'s typed-argument split). ShellTool (and so BashTool/PowerShellTool) already does this — pass run_in_background: true in a call's arguments and it hands back a handle instead of running to completion inline.

IEngine engine = new EngineBuilder()
    .WithModel(model)
    .AddBaselineTools()
    .WithBackgroundTasks(new InMemoryBackgroundTaskRegistry())
    .Build();

WithBackgroundTasks does two things: it wires Engine.Options.BackgroundDispatch to route any run_in_background: true call on an IBackgroundTool through the given IBackgroundTaskRegistry instead of running it inline, and it registers TaskOutputTool/TaskStopTool against that same registry so the model can poll or stop what it started, by the task id the initial call returns. Without a registry wired in, run_in_background is simply ignored and every tool runs the ordinary way.

HTTP proxy tools

A tool that just calls an external REST API — build a request from the model's arguments, send it, hand the response back — can subclass HttpProxyTool<TArgs> instead of hand-rolling ExecuteAsync. It owns the build → send → format template; a subclass overrides BuildRequestAsync (URI, method, auth headers, body) and, optionally, FormatResponseAsync (default is "{status} {reason}\n{body}"). HttpClientProxyTool<TArgs> is the plain-HttpClient implementation — override CreateHttpClient for custom transport (a handler with custom decompression, for instance), otherwise a lazily-created default HttpClient is reused for the tool's lifetime.

public sealed class WeatherTool(string apiKey) : HttpClientProxyTool<WeatherTool.Args>
{
    public override string Name => "get_weather";
    public override ToolDefinition Definition { get; } = /* ... */;

    protected override Task<HttpRequestMessage> BuildRequestAsync(Args args, CancellationToken ct)
    {
        var request = new HttpRequestMessage(HttpMethod.Get, $"https://api.example.com/weather?city={args.City}");
        request.Headers.Authorization = new("Bearer", apiKey);
        return Task.FromResult(request);
    }

    public sealed record Args(string City);
}

Pair this with Universal.Operative.Sdk.Http's HttpServiceClientProxyTool<TServiceClient, TArgs> when the target API already has a Universal.Common.Net.Http.HttpServiceClient — auth and success/failure handling then live once, on that client, rather than being reimplemented on the tool.

Subagents

AgentTool spawns a subagent — a fresh, single-turn IOperative run against its own scratch Conversation seeded with a prompt — and returns its final answer. It's an IBackgroundTool, so run_in_background: true works the same way as any other background tool.

builder.AddAgentTool(new Dictionary<string, AgentTool.SubagentType>
{
    ["research"] = new("Web research only.",
        Factory: () => new EngineBuilder().WithModel(model).AddTools(webSearchTool, webFetchTool)),
    ["code"] = new("Filesystem and shell only.",
        Factory: () => new EngineBuilder().WithModel(model).AddBaselineTools()),
    ["reviewer"] = new("Reviews code for correctness.",
        Factory: () => new EngineBuilder().WithModel(cheaperModel).AddTools(readTool, grepTool),
        SystemPrompt: "You are a meticulous, skeptical code reviewer..."),
});

Register named AgentTool.SubagentTypes to let the calling model choose what kind of subagent it gets. Each type's Factory is a complete, independent EngineBuilder recipe — model, tools, transformers, timeouts, whatever that type needs, computed fresh on every spawn — rather than a filtered subset of some shared menu, so at least one type must be registered; there's nothing to build an implicit fallback from. SystemPrompt sets that type's own persona (the scratch conversation's system prompt). With only one type registered, subagent_type is omitted from the schema entirely and every spawn just uses it.

Because each type builds its own engine from scratch, the self-recursion guard isn't automatic the way a shared-tool-pool design could make it: unless a type sets AllowRecursion: true, AgentTool strips any tool literally named "Agent" from whatever that type's Factory builds, so a type still can't spawn further subagents just because its own factory closure happened to include this same tool instance.

EngineBuilder.AddAgentTool(types) is a small naming/discoverability convenience alongside AddBaselineTools/WithBackgroundTasks — equivalent to AddTool(new AgentTool(types)). The constructor is still there directly if you'd rather not go through EngineBuilder at all.

AgentTool isn't sealed — like Engine, every step is protected virtual so a subclass overrides one piece without reimplementing the rest: BuildOperativeFactory (how a resolved type becomes engine wiring), BuildConversation (how the scratch conversation gets seeded), BuildDefinition (the schema/prompt shown to the calling model), or ExtractFinalAnswer (how the subagent's last message becomes the tool's return text).

Under the hood, a spawn's turn runs through an IOperativeFactory/IOperative (built by BuildOperativeFactory) rather than driving IEngine directly — ResumeAsync() is called once against a scratch conversation already seeded with the prompt.

From EngineBuilder straight to a long-lived IOperative

EngineBuilder.BuildOperative(conversation, configureOperative?) builds this configuration into an IEngine and wraps it, plus a Conversation, into a ready IOperative in one call — the common "one engine, one session on top of it" case, without a separate OperativeBuilder.Create(builder.Build(), conversation).Build() to keep in sync by hand:

IOperative operative = new EngineBuilder()
    .WithModel(model)
    .AddBaselineTools()
    .BuildOperative(conversation,
        configureOperative: b => b.WithReloadBeforeStart(reload));

The returned IOperative's engine is exactly what this Build() call produced, so it automatically shares this builder's model, tools, and transformers.

License

Free for noncommercial use under the Universal.Operative Noncommercial License. Source is closed; commercial use requires a separate license from Andrew Ong.