Langertha-Knarr
view release on metacpan or search on metacpan
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
<capabilities>
## Capability Queries
Every engine reports its capabilities via `Langertha::Role::Capabilities`
(composed by `Role::Chat`, so present on every engine):
```perl
$engine->supports('tool_choice_named') or die "engine cannot force named tool";
$engine->supports('response_format_json_schema') # safe to pass json_schema response_format
$engine->supports('streaming') # chat_stream_request wired up
$engine->supports('tools_native') # accepts a tools array on the wire
$engine->supports('tools_hermes') # Hermes XML-tag tool path
my $caps = $engine->engine_capabilities;
# { chat=>1, streaming=>1, tools_native=>1, tool_choice_named=>1,
# response_format_json_schema=>1, embedding=>1, transcription=>1, ... }
```
The flag set is derived from which capability roles the engine composes
(central role→flag map in `Role::Capabilities`); engines override via
`around engine_capabilities` for wire-reality corrections.
</capabilities>
<chat-f>
## chat_f — Single-Turn with Named Args
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
Engines compose feature roles:
| Role | Feature |
|------|---------|
| `Langertha::Role::Capabilities` | `engine_capabilities` registry + `supports($cap)` |
| `Langertha::Role::Chat` | `simple_chat`, `simple_chat_f`, `chat_f` (named args), `aggregate_tool_calls` |
| `Langertha::Role::Tools` | `chat_with_tools_f` (MCP loop) |
| `Langertha::Role::HermesTools` | XML-tag tool calling for models without native support |
| `Langertha::Role::ParallelToolUse` | `parallel_tool_use` boolean (canonical name) |
| `Langertha::Role::Streaming` | SSE/NDJSON streaming |
| `Langertha::Role::Embedding` | Vector embeddings |
| `Langertha::Role::Transcription` | Audio-to-text |
| `Langertha::Role::ImageGeneration` | Image generation |
| `Langertha::Role::SystemPrompt` | System prompt management |
| `Langertha::Role::Temperature` | Sampling temperature |
| `Langertha::Role::Seed` | Deterministic seed (`seed`, `randomize_seed`) |
| `Langertha::Role::ContextSize` | `context_size` parameter |
| `Langertha::Role::ResponseSize` | `response_size` / max_tokens parameter |
| `Langertha::Role::ResponseFormat` | JSON mode / structured output, plus `$self->decode_loose_json($text)` (overridable) |
| `Langertha::Role::Models` | Model listing |
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
<value-objects>
## Value Objects
| Class | Purpose |
|-------|---------|
| `Langertha::Tool` | Canonical tool definition. `from_openai/from_anthropic/from_mcp/from_gemini/from_hash` accept any shape; `to_openai/to_anthropic/to_gemini/to_mcp/to_json_schema` emit per-provider wire payloads. |
| `Langertha::ToolChoice` | Canonical tool-selection policy (`auto`/`any`/`none`/`tool`). `to_openai/to_anthropic/to_gemini/to_perplexity` per-provider serializers. |
| `Langertha::ToolCall` | Tool invocation emitted by an LLM. `name`, `arguments`, `id`, `synthetic`. `from_openai/from_anthropic/from_ollama/from_gemini`; `extract($raw)` pulls every call out of any known response shape. |
| `Langertha::Content::Image` | Provider-agnostic vision input. `from_url/from_file/from_data`; `to_openai/to_anthropic/to_gemini`. |
| `Langertha::Response` | LLM response with metadata. Stringifies to `content`. `tool_calls` is `ArrayRef[Langertha::ToolCall]` — single source of truth. |
| `Langertha::Stream::Chunk` | Single streaming chunk. Optional `tool_calls` for engines that emit them mid-stream; `Role::Chat::aggregate_tool_calls(\@chunks)` flattens. |
Use these instead of hand-rolled hashes when normalizing across
providers. `Tool->from_hash` auto-detects MCP camelCase, Anthropic
snake_case, OpenAI envelope, and Gemini-flat shapes.
</value-objects>
- Tracing flush is async via Net::Async::HTTP. The previous
LWP::UserAgent call blocked the IO::Async event loop on every
end_trace; the new flush fires the request and returns
immediately, with a warn-on-fail logger attached.
- Streaming pump consolidated into Knarr::Stream::from_callback.
Removes ~50 lines of duplicated queue/pending/finished/error
bookkeeping from Handler::Engine and Handler::Router.
- supports()-aware streaming detection. The old
$engine->can('simple_chat_stream_realtime_f') heuristic now
defers to $engine->supports('streaming') when available
(Langertha 0.500+) and falls back to the can()-check for older
engines.
- Dead steerboard attribute removed from all six Protocol classes
(never read), and Knarr.pm no longer threads $self into
protocol-object construction. Knarr::PSGI's constructor argument
renamed steerboard => knarr.
- 'steerboard' string fallbacks replaced with 'unknown' (model
default) and 'knarr-code' / 'knarr-raider' (handler defaults).
* Passthrough with and without tracing enabled
- Documentation updated: CLAUDE.md, README.md, bin/knarr POD
reflect new architecture, --from-env, repeatable --port,
KNARR_DEBUG, raw passthrough behavior.
1.000 2026-04-10 00:00:20Z
- MAJOR REWRITE. Mojolicious is gone; the new core is built on
IO::Async + Net::Async::HTTP::Server + Future::AsyncAwait for
native async streaming and seamless integration with Langertha
engines. Existing knarr.yaml configs and the knarr CLI commands
(start, container, models, check, init) keep working.
- Six wire protocols loaded by default on every listening port:
OpenAI (/v1/chat/completions, /v1/models, SSE), Anthropic
(/v1/messages, named-event SSE), Ollama (/api/chat, /api/tags,
NDJSON streaming), A2A (Google Agent2Agent JSON-RPC at / with
/.well-known/agent.json discovery), ACP (BeeAI/IBM /agents,
/runs), and AG-UI (CopilotKit /awp event stream).
- Pluggable Handler architecture replaces the old Knarr::Proxy
classes:
Knarr::Handler::Router — model→engine via Knarr::Router
Knarr::Handler::Engine — single Langertha engine wrapper
Knarr::Handler::Raider — per-session Langertha::Raider
Knarr::Handler::Code — coderef-backed for tests/fakes
Knarr::Handler::Passthrough — raw HTTP forward to upstream
reconfiguration.
- proxy_api_key authentication enforced by the new dispatcher.
Clients must present 'Authorization: Bearer <token>' or
'x-api-key: <token>' when KNARR_API_KEY (or yaml proxy_api_key)
is set. The /.well-known/agent.json discovery route stays
anonymous.
- PSGI adapter (Knarr::PSGI) for deploying behind any Plack
server (Starman, Twiggy, mod_perl, ...). Streaming is buffered
in this mode; use the native server for real-time streaming.
- Universal protocol translator pattern: an OpenAI-fronted Knarr
with a Handler::A2AClient backend exposes a remote A2A agent to
OpenAI clients (and similarly for ACP).
- Drops the Knarr namespace facades for Langertha utility code
(Knarr::Metrics, Knarr::Input, Knarr::Output, Knarr::Input::Tools,
Knarr::Output::Tools). Use Langertha::Usage / Tool / ToolCall /
ToolChoice / Pricing / Cost / UsageRecord directly from
Langertha core.
HTTP::Message, File::ShareDir::ProjectDistDir; adds IO::Async,
Net::Async::HTTP, Net::Async::HTTP::Server, Future::AsyncAwait,
Moose, Path::Tiny (test), Capture::Tiny (test).
- Bump Langertha floor to 0.400 for the new value object API.
- dist.ini sets irc = #langertha (on irc.perl.org).
- Test suite expanded from 200 to 320 unit tests + author POD
syntax. New coverage:
* end-to-end live streaming for Anthropic / A2A / ACP / AG-UI
(OpenAI and Ollama already covered)
* Handler::Passthrough sync + streaming through a real backend
* Handler::Router with passthrough fallback
* Multi-listen on real sockets
* Tracing and RequestLog decorators
* CLI smoketest for knarr models / check / init
* proxy_api_key auth enforcement
0.007 2026-03-10 20:05:00Z
- Resolve configured engines through both `Langertha::Engine::*` and
`LangerthaX::Engine::*` (including fully-qualified class names).
t/00-load.t
t/10-config.t
t/10_openai_protocol.t
t/15_response.t
t/20-router.t
t/20_handlers.t
t/25_chat_f_params.t
t/26_tool_calls_routing.t
t/27_usage_routing.t
t/30_all_protocols.t
t/40_streaming_live.t
t/41_streaming_protocols_live.t
t/50_translator.t
t/60_psgi.t
t/70_passthrough.t
t/71_router_passthrough_fallback.t
t/72_multi_listen.t
t/73_tracing_handler.t
t/74_requestlog_handler.t
t/75_cli_smoketest.t
t/76_auth.t
t/77_raw_passthrough_tracing.t
lib/Langertha/Knarr.pm view on Meta::CPAN
Ollama SDKs, and the agent ecosystems around A2A, ACP, and AG-UI.
By default a single running Knarr answers OpenAI
C</v1/chat/completions>, Anthropic C</v1/messages>, Ollama
C</api/chat>, A2A's C</.well-known/agent.json> plus JSON-RPC C</>,
ACP's C</runs>, and AG-UI's C</awp> simultaneously on every listening
port. The same handler implementation drives all of them.
Knarr 1.000 is built on L<IO::Async> and L<Net::Async::HTTP::Server>
with native L<Future::AsyncAwait> integration into Langertha engines,
so streaming works end-to-end token-by-token without any thread or
event-loop bridges.
=head1 ARCHITECTURE
Three pluggable layers:
=over
=item B<Protocols>
lib/Langertha/Knarr.pm view on Meta::CPAN
L<Langertha::Knarr::Handler::Code> (coderef-backed for tests). Implement
L<Langertha::Knarr::Handler> to write your own. Decorators
(L<Langertha::Knarr::Handler::Tracing>,
L<Langertha::Knarr::Handler::RequestLog>) wrap any inner handler and
add behavior on top — they themselves consume the Handler role and
compose freely.
=item B<Transport>
Default is L<Net::Async::HTTP::Server> with chunked SSE / NDJSON
streaming on one or more listen sockets. For Plack deployments,
L<Langertha::Knarr::PSGI> wraps the same Knarr instance into a PSGI
app (buffered — see its docs for the streaming caveat).
=back
=head2 handler
Required. An object consuming L<Langertha::Knarr::Handler>.
=head2 listen
ArrayRef of C<host:port> strings or C<< { host => ..., port => ... } >>
lib/Langertha/Knarr/Handler.pm view on Meta::CPAN
our $VERSION = '1.100';
use Moose::Role;
use Future::AsyncAwait;
use Langertha::Knarr::Stream;
use Langertha::Knarr::Response;
requires 'handle_chat_f';
requires 'list_models';
# Default streaming = run handle_chat_f and emit one chunk.
# Handlers that natively stream should override.
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my $r = Langertha::Knarr::Response->coerce( await $self->handle_chat_f($session, $request) );
return Langertha::Knarr::Stream->from_list( $r->content );
}
1;
__END__
lib/Langertha/Knarr/Handler/Code.pm view on Meta::CPAN
with 'Langertha::Knarr::Handler';
has code => (
is => 'ro',
isa => 'CodeRef',
required => 1,
);
# Optional: separate generator for streaming. Returns a coderef that itself
# returns next-chunk strings (undef = done) when called.
has stream_code => (
is => 'ro',
isa => 'Maybe[CodeRef]',
default => sub { undef },
);
has models => (
is => 'ro',
isa => 'ArrayRef',
lib/Langertha/Knarr/Handler/Code.pm view on Meta::CPAN
},
stream_code => sub {
my @parts = ('hel', 'lo');
return sub { @parts ? shift @parts : undef };
},
);
=head1 DESCRIPTION
The simplest possible handler: pass coderefs that return strings (or
chunk generators for streaming) and you get a working Knarr handler.
Useful for tests, fakes, smoketests, and "fake LLM" demos.
=head2 code
Required. Coderef called as C<< $code->($session, $request) >> for
non-streaming requests; must return a scalar string.
=head2 stream_code
Optional. Coderef returning another coderef that yields the next chunk
per call, C<undef> to signal end.
=head2 models
Optional. Arrayref of model descriptors. Defaults to a single
C<knarr-code> entry.
lib/Langertha/Knarr/Handler/Engine.pm view on Meta::CPAN
async sub handle_chat_f {
my ($self, $session, $request) = @_;
my $response = await $self->engine->chat_f( $request->chat_f_args($self->engine) );
my $r = Langertha::Knarr::Response->coerce($response);
return $r->clone_with( model => $r->model // $self->_model_id );
}
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my $engine = $self->engine;
unless ( _supports_streaming($engine) ) {
# Engine doesn't support native streaming — fall back to single-chunk.
my $r = await $self->handle_chat_f($session, $request);
return Langertha::Knarr::Stream->from_list( $r->content );
}
return Langertha::Knarr::Stream->from_callback( sub {
my ($emit, $done, $fail) = @_;
my $cb = sub {
my ($chunk) = @_;
my $text = ref $chunk && $chunk->can('content') ? $chunk->content : "$chunk";
$emit->($text);
};
my $f = $engine->simple_chat_stream_realtime_f( $cb, @{ $request->messages } );
$f->on_done( $done );
$f->on_fail( $fail );
$f->retain;
});
}
sub _supports_streaming {
my ($engine) = @_;
return $engine->supports('streaming') if $engine->can('supports');
return $engine->can('simple_chat_stream_realtime_f') && $engine->can('chat_stream_request');
}
sub list_models {
my ($self) = @_;
return [ { id => $self->_model_id, object => 'model' } ];
}
__PACKAGE__->meta->make_immutable;
1;
lib/Langertha/Knarr/Handler/Engine.pm view on Meta::CPAN
);
my $handler = Langertha::Knarr::Handler::Engine->new(
engine => $engine,
model_id => 'groq-llama-3.3-70b',
);
=head1 DESCRIPTION
Wraps a single L<Langertha::Engine::*> instance and exposes it as a
Knarr handler. Non-streaming requests are dispatched via
C<< $engine->chat_f >> with the full set of generation parameters
(C<tools>, C<tool_choice>, C<response_format>, C<temperature>,
C<max_tokens>) forwarded from the client request — subject to the
engine's reported capabilities. Streaming requests use
C<simple_chat_stream_realtime_f> for native token-by-token delivery;
engines that don't support streaming fall back to a single-chunk
emission.
For routing across multiple engines based on model name, use
L<Langertha::Knarr::Handler::Router> with a L<Langertha::Knarr::Router>
config instead.
=head2 engine
Required. Any object consuming L<Langertha::Role::Chat>. Streaming
support is detected via C<< $engine->supports('streaming') >> (Langertha
0.500+) or the presence of C<simple_chat_stream_realtime_f>.
=head2 model_id
Optional. The id reported by L</list_models> and surfaced in responses.
Defaults to the engine's C<chat_model>, falling back to a derived name
from the engine class.
=head1 SUPPORT
lib/Langertha/Knarr/Handler/Passthrough.pm view on Meta::CPAN
}
}
if ( my $auth = $self->default_auth ) {
$http_req->header( Authorization => $auth ) unless $http_req->header('Authorization');
}
$http_req->content( $self->_json->encode($body) );
return $http_req;
}
# Extract assistant text from an upstream response body for the protocol it
# came from. Bare-minimum extractor for sync mode; the streaming path
# forwards bytes verbatim and doesn't need this.
sub _extract_text {
my ($self, $protocol_name, $resp_body) = @_;
my $data = eval { $self->_json->decode($resp_body) };
return '' unless ref $data eq 'HASH';
if ( $protocol_name eq 'openai' ) {
return $data->{choices}[0]{message}{content} // '';
}
if ( $protocol_name eq 'anthropic' ) {
my $bits = '';
lib/Langertha/Knarr/Handler/Passthrough.pm view on Meta::CPAN
},
);
=head1 DESCRIPTION
Forwards the original wire-format request verbatim to a real upstream
API. The protocol's parser already turned the body into a
L<Langertha::Knarr::Request>; Passthrough rebuilds the upstream JSON
from C<$request-E<gt>raw> and re-POSTs it.
Both sync and streaming requests are supported. For streaming, the
upstream's protocol-native chunks are extracted into plain text deltas
which the front-side protocol then re-frames — keeping symmetry even
when client and upstream use the same protocol.
This is the building block behind Knarr's classic "configure your API
keys once, point everything at me" use case.
=head2 upstreams
Required. HashRef mapping protocol name (C<openai>, C<anthropic>,
lib/Langertha/Knarr/Handler/RequestLog.pm view on Meta::CPAN
my $rlog = Langertha::Knarr::RequestLog->new(config => $config);
$handler = Langertha::Knarr::Handler::RequestLog->new(
wrapped => $handler,
request_log => $rlog,
);
=head1 DESCRIPTION
Decorator handler that writes a structured per-request log entry for
every chat or stream request via L<Langertha::Knarr::RequestLog>.
Sync requests log a single line with the result; streaming requests
accumulate every delta and log one line with the assembled output
when the stream closes.
C<knarr start> mounts this automatically when
C<KNARR_LOG_FILE> / C<KNARR_LOG_DIR> (or the YAML C<logging:> section)
is set.
=head2 wrapped
Required. The inner L<Langertha::Knarr::Handler> being decorated.
lib/Langertha/Knarr/Handler/Router.pm view on Meta::CPAN
}
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my ($engine) = $self->_resolve( $request->model );
unless ( $engine ) {
return await $self->passthrough->handle_stream_f( $session, $request );
}
unless ( _supports_streaming($engine) ) {
my $r = await $self->handle_chat_f($session, $request);
return Langertha::Knarr::Stream->from_list( $r->content );
}
return Langertha::Knarr::Stream->from_callback( sub {
my ($emit, $done, $fail) = @_;
my $cb = sub {
my ($chunk) = @_;
my $text = ref $chunk && $chunk->can('content') ? $chunk->content : "$chunk";
$emit->($text);
};
my $f = $engine->simple_chat_stream_realtime_f( $cb, @{ $request->messages } );
$f->on_done( $done );
$f->on_fail( $fail );
$f->retain;
});
}
sub _supports_streaming {
my ($engine) = @_;
return $engine->supports('streaming') if $engine->can('supports');
return $engine->can('simple_chat_stream_realtime_f') && $engine->can('chat_stream_request');
}
sub list_models {
my ($self) = @_;
my $models = $self->router->list_models;
return [ map { ref $_ eq 'HASH' ? $_ : { id => "$_", object => 'model' } } @{ $models || [] } ];
}
__PACKAGE__->meta->make_immutable;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
package Langertha::Knarr::PSGI;
# ABSTRACT: PSGI adapter for Langertha::Knarr (buffered, no streaming)
our $VERSION = '1.100';
use Moose;
use JSON::MaybeXS;
use Langertha::Knarr::Request;
# Wraps a Langertha::Knarr instance and returns a PSGI app coderef.
# Streaming requests are coerced into buffered responses: the full body is
# assembled (open + chunks + close + done) before being returned to the
# PSGI server. Use the native Net::Async::HTTP::Server entrypoint
# (Langertha::Knarr->run) if you need real streaming.
has knarr => ( is => 'ro', required => 1 );
has _json => (
is => 'ro',
default => sub { JSON::MaybeXS->new( utf8 => 1, canonical => 1 ) },
);
sub to_app {
my ($self) = @_;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
[ $self->_json->encode({ error => { message => "unknown action $action" } }) ] ];
}
my $body = $self->_read_body($env);
my $fake_http = Langertha::Knarr::PSGI::FakeReq->new( $env );
my $sb_req = $proto->parse_chat_request( $fake_http, \$body );
my $session = $sb->session( $sb_req->session_id );
my $handler = $sb->handler;
if ( $sb_req->stream ) {
# Buffered streaming: drive the stream to completion, concatenate frames.
my $stream = $handler->handle_stream_f( $session, $sb_req )->get;
my $out = $proto->format_stream_open($sb_req);
while ( defined( my $delta = $stream->next_chunk_f->get ) ) {
$out .= $proto->format_stream_chunk( $delta, $sb_req );
}
$out .= $proto->format_stream_close($sb_req);
$out .= $proto->format_stream_done($sb_req);
return [ 200, [ 'Content-Type' => $proto->stream_content_type ], [ $out ] ];
}
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
1;
__END__
=pod
=encoding UTF-8
=head1 NAME
Langertha::Knarr::PSGI - PSGI adapter for Langertha::Knarr (buffered, no streaming)
=head1 VERSION
version 1.100
=head1 SYNOPSIS
use Langertha::Knarr;
use Langertha::Knarr::PSGI;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
# Run with any PSGI server:
# plackup -s Starman -p 8088 app.psgi
=head1 DESCRIPTION
Adapter that wraps a L<Langertha::Knarr> instance and exposes it as a
PSGI app, so you can deploy Knarr behind any Plack server (Starman,
Twiggy, Gazelle, mod_perl, etc.) instead of running its native
L<Net::Async::HTTP::Server> loop.
B<Streaming responses are buffered.> The PSGI streaming protocol's
delayed-response form does work in theory but mixes badly with
L<IO::Async> in the same process; for honesty's sake this adapter
just drives the inner stream to completion in a blocking loop and
returns the full assembled body. Use the native
L<Langertha::Knarr/run> entry point if you need real-time streaming.
=head2 knarr
Required. The L<Langertha::Knarr> instance to expose.
=head2 to_app
Returns the PSGI coderef.
=head1 SUPPORT
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
my ($self) = @_;
return "data: [DONE]\n\n";
}
# Optional lifecycle hooks for protocols that need to frame the stream
# (Anthropic message_start/stop, A2A status events, ACP run.created, AGUI RUN_STARTED).
# Default: empty — protocols like OpenAI / Ollama don't need them.
sub format_stream_open { '' }
sub format_stream_close { '' }
# Content-Type for streaming responses. Default is SSE; Ollama overrides.
sub stream_content_type { 'text/event-stream' }
# format_models_response(\@models) -> ($status, \%headers, $body)
sub format_models_response {
my ($self, $models) = @_;
return ( 200, { 'Content-Type' => 'application/json' }, '{"data":[]}' );
}
1;
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
via the matched protocol's parser/formatter.
Knarr ships with six concrete protocols, all loaded by default:
=over
=item * L<Langertha::Knarr::Protocol::OpenAI> — C</v1/chat/completions>, SSE
=item * L<Langertha::Knarr::Protocol::Anthropic> — C</v1/messages>, named SSE events
=item * L<Langertha::Knarr::Protocol::Ollama> — C</api/chat>, NDJSON streaming
=item * L<Langertha::Knarr::Protocol::A2A> — Google Agent2Agent JSON-RPC
=item * L<Langertha::Knarr::Protocol::ACP> — IBM/BeeAI Agent Communication Protocol
=item * L<Langertha::Knarr::Protocol::AGUI> — CopilotKit AG-UI event protocol
=back
=head2 protocol_name
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
Required. Returns a L<Langertha::Knarr::Request>.
=head2 format_chat_response
my ($status, \%headers, $body) = $proto->format_chat_response($response, $request);
Required. Returns the HTTP response triple for sync mode.
=head2 format_stream_open / format_stream_chunk / format_stream_close / format_stream_done
Lifecycle hooks for streaming responses. Defaults are no-ops where the
protocol doesn't need framing — Anthropic/A2A/ACP/AG-UI override these
to emit their named events around the chunk stream.
=head2 stream_content_type
Returns the HTTP C<Content-Type> for streaming responses. Default
C<text/event-stream>; Ollama overrides to C<application/x-ndjson>.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
with 'Langertha::Knarr::Protocol';
# --- A2A overview ---
# Discovery: GET /.well-known/agent.json -> AgentCard JSON
# Method bus: POST / body: JSON-RPC 2.0 envelope
# methods: "tasks/send", "tasks/sendSubscribe", "tasks/get",
# "tasks/cancel", "tasks/pushNotification/set", ...
#
# tasks/send (sync): returns Task JSON in JSON-RPC result
# tasks/sendSubscribe (streaming): returns SSE stream of JSON-RPC responses
# Each event is a full JSON-RPC response wrapping a TaskStatusUpdateEvent or
# TaskArtifactUpdateEvent. Final event has status.state = "completed".
#
# Streaming model on the wire:
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","status":{"state":"working","message":{...}}, "final":false}}\n\n
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","artifact":{"parts":[{"type":"text","text":"Hi"}], "index":0,"append":true}, "final":false}}\n\n
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","status":{"state":"completed"},"final":true}}\n\n
# ---------------------------------------------------------------------------
has _json => ( is => 'ro', default => sub { JSON::MaybeXS->new( utf8 => 1, canonical => 1 ) } );
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
);
sub _build_agent_card {
my ($self) = @_;
return {
name => 'Langertha Steerboard Agent',
description => 'Steerboard-exposed agent',
url => '/',
version => '0.0.1',
capabilities => {
streaming => JSON::MaybeXS::true(),
pushNotifications => JSON::MaybeXS::false(),
stateTransitionHistory => JSON::MaybeXS::false(),
},
defaultInputModes => ['text'],
defaultOutputModes => ['text'],
skills => [],
};
}
sub protocol_name { 'a2a' }
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
=over
=item * C<GET /.well-known/agent.json> — agent card discovery (anonymous)
=item * C<POST /> — JSON-RPC 2.0 method bus
=back
Supported methods: C<tasks/send> (sync) and C<tasks/sendSubscribe>
(streaming). Streaming wraps every chunk in a JSON-RPC envelope with
the original C<id> preserved, transitions C<status.state> from
C<working> to C<completed>, and emits artifact append events for the
text deltas.
=head2 agent_card
The HashRef returned by the discovery endpoint. Defaults to a generic
"Knarr Agent" card with C<streaming: true>; override to advertise
specific skills, version, etc.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
use Moose;
use JSON::MaybeXS;
use Data::UUID;
use Time::HiRes qw( time );
use Langertha::Knarr::Request;
use Langertha::Knarr::Response;
with 'Langertha::Knarr::Protocol';
# --- AG-UI overview ---
# CopilotKit-driven event protocol meant for agent <-> UI streaming.
# Transport-agnostic; SSE is the canonical wire form.
# Endpoint convention: POST /awp (Agent Wire Protocol) — body is the run input,
# response is an SSE stream of typed events.
#
# Event types (subset, all framed as SSE "data: { ... }" with "type" field):
# RUN_STARTED { type, threadId, runId }
# TEXT_MESSAGE_START { type, messageId, role:"assistant" }
# TEXT_MESSAGE_CONTENT { type, messageId, delta:"..." }
# TEXT_MESSAGE_END { type, messageId }
# TOOL_CALL_START { type, toolCallId, toolCallName, parentMessageId }
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
my $data = $self->_json->decode( $$body_ref || '{}' );
my @msgs;
for my $m ( @{ $data->{messages} || [] } ) {
push @msgs, { role => $m->{role} // 'user', content => $m->{content} // '' };
}
return Langertha::Knarr::Request->new(
protocol => 'agui',
raw => $data,
model => $data->{model},
messages => \@msgs,
stream => 1, # AG-UI is always streaming
session_id => $data->{threadId},
extra => {
thread_id => $data->{threadId} // $self->_uuid->create_str,
run_id => $data->{runId} // $self->_uuid->create_str,
message_id => $self->_uuid->create_str,
},
);
}
sub format_chat_response {
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
version 1.100
=head1 DESCRIPTION
Implements the CopilotKit AG-UI agent-to-UI event protocol on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /awp> — Agent Wire Protocol endpoint, always streaming
=back
Streaming emits the typed AG-UI event sequence:
C<RUN_STARTED>, C<TEXT_MESSAGE_START>, C<TEXT_MESSAGE_CONTENT>×N,
C<TEXT_MESSAGE_END>, C<RUN_FINISHED>. Sync mode synthesizes the same
sequence into a single response body for non-streaming clients.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/Anthropic.pm view on Meta::CPAN
role => 'assistant',
model => $r->model // $request->model // 'unknown',
content => \@blocks,
stop_reason => $stop_reason,
stop_sequence => undef,
usage => $usage,
};
return ( 200, { 'Content-Type' => 'application/json' }, $self->_json->encode($payload) );
}
# Anthropic streaming uses named SSE events. We render full event blocks.
sub _sse_event {
my ($self, $event, $data) = @_;
return "event: $event\ndata: " . $self->_json->encode($data) . "\n\n";
}
sub format_stream_open {
my ($self, $request) = @_;
my $id = _msg_id();
my $model = $request->model // 'unknown';
return join( '',
lib/Langertha/Knarr/Protocol/Anthropic.pm view on Meta::CPAN
version 1.100
=head1 DESCRIPTION
Implements the Anthropic Messages wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /v1/messages> — sync and named-event SSE streaming
=back
Streaming emits the full event sequence the Anthropic SDK expects:
C<message_start>, C<content_block_start>, C<content_block_delta>×N,
C<content_block_stop>, C<message_delta>, C<message_stop>.
=head1 SUPPORT
=head2 Issues
lib/Langertha/Knarr/Protocol/Ollama.pm view on Meta::CPAN
version 1.100
=head1 DESCRIPTION
Implements the Ollama wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /api/chat>, C<POST /api/generate> — chat with NDJSON streaming
=item * C<GET /api/tags> — model listing
=item * C<GET /api/version> — version probe
=back
Streaming uses newline-delimited JSON (NDJSON) rather than SSE — the
C<Content-Type> is C<application/x-ndjson> and each chunk is a single
JSON object per line. The final chunk has C<done: true>.
lib/Langertha/Knarr/Protocol/OpenAI.pm view on Meta::CPAN
version 1.100
=head1 DESCRIPTION
Implements the OpenAI Chat Completions wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default in every
L<Langertha::Knarr> instance.
=over
=item * C<POST /v1/chat/completions> — sync and SSE streaming
=item * C<GET /v1/models> — model listing
=back
Streaming uses the standard SSE chunk format with C<data: [DONE]> as
the terminator. C<tools>, C<tool_choice>, and C<response_format> are
extracted into L<Langertha::Knarr::Request> attributes and forwarded to
the engine via C<chat_f>. Tool-call responses are serialised into
C<message.tool_calls> with C<finish_reason: "tool_calls">.
lib/Langertha/Knarr/Request.pm view on Meta::CPAN
=head2 model
Optional model id from the request body.
=head2 messages
ArrayRef of message hashes (C<< { role => ..., content => ... } >>).
=head2 stream
Boolean. Whether the client requested streaming.
=head2 temperature, max_tokens, tools, tool_choice, response_format, system
Optional generation parameters and tool definitions, if the protocol
extracted them. C<tool_choice> and C<response_format> are passed to
L<Langertha::Engine> via C<chat_f> in their canonical form; Langertha
normalizes them to the target engine's wire format.
=head2 session_id
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
package Langertha::Knarr::Stream;
# ABSTRACT: Async chunk iterator returned by streaming Knarr handlers
our $VERSION = '1.100';
use Moose;
use Future;
sub from_callback {
my ($class, $setup) = @_;
my @queue;
my $pending;
my $finished = 0;
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
1;
__END__
=pod
=encoding UTF-8
=head1 NAME
Langertha::Knarr::Stream - Async chunk iterator returned by streaming Knarr handlers
=head1 VERSION
version 1.100
=head1 SYNOPSIS
use Langertha::Knarr::Stream;
# From a fixed list of strings
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
source => sub { $next_chunk_future },
);
# Drain it
while ( defined( my $chunk = $stream->next_chunk_f->get ) ) {
print $chunk;
}
=head1 DESCRIPTION
The chunk iterator that streaming Knarr handlers return. Supports two
construction modes: a sync C<generator> coderef that returns the next
chunk string each call (or C<undef> for end), or a C<source> coderef
that returns a L<Future> resolving to the next chunk string. The
Future form is the one real async backends like L<Net::Async::HTTP>
use; the generator form is for tests and simple cases.
=head2 generator
Optional. CodeRef returning the next chunk synchronously.
t/40_streaming_live.t view on Meta::CPAN
$sb->start;
# Discover the actual port we bound to
my $sock = $sb->_server->read_handle;
my $port = $sock->sockport;
ok( $port, "bound to port $port" );
my $http = Net::Async::HTTP->new;
$loop->add($http);
# --- 1) non-streaming OpenAI roundtrip ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/v1/chat/completions" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
is( $resp->code, 200, 'non-stream 200' );
my $data = $json->decode( $resp->decoded_content );
is( $data->{object}, 'chat.completion', 'non-stream object' );
is( $data->{choices}[0]{message}{content}, 'hello world', 'non-stream content' );
}
# --- 2) streaming OpenAI ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/v1/chat/completions" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
is( $resp->code, 200, 'stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE content type' );
my $body_text = $resp->decoded_content;
my @data_lines = grep { /^data:/ } split /\n/, $body_text;
ok( scalar(@data_lines) >= 4, 'multiple SSE data lines emitted' );
like( $body_text, qr/hel/, 'first chunk present' );
like( $body_text, qr/ld/, 'last chunk present' );
}
# --- 3) streaming Ollama (NDJSON) ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/api/chat" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
t/41_streaming_protocols_live.t view on Meta::CPAN
$loop->add($http);
sub post {
my ($path, $body) = @_;
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port$path" );
$req->header( 'Content-Type' => 'application/json' );
$req->content( $json->encode($body) );
return $http->do_request( request => $req )->get;
}
# --- Anthropic streaming: named SSE events ---
{
my $resp = post( '/v1/messages', {
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
is( $resp->code, 200, 'anthropic stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/event: message_start/, 'message_start frame' );
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/event: content_block_stop/, 'block_stop frame' );
like( $body, qr/event: message_stop/, 'message_stop frame' );
like( $body, qr/"text":"hel"/, 'first chunk text in delta' );
like( $body, qr/"text":"ld"/, 'last chunk text in delta' );
# Count content_block_delta events — should equal number of stream chunks (4).
my $deltas = () = $body =~ /event: content_block_delta/g;
is( $deltas, 4, 'four content_block_delta events' );
}
# --- A2A streaming via JSON-RPC tasks/sendSubscribe ---
{
my $resp = post( '/', {
jsonrpc => '2.0',
id => 1,
method => 'tasks/sendSubscribe',
params => {
id => 'task-1',
message => { role => 'user', parts => [ { type => 'text', text => 'hi' } ] },
},
});
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/"text":"hel"/, 'first chunk text' );
like( $body, qr/"text":"ld"/, 'last chunk text' );
like( $body, qr/"final":true/, 'final flag set on close' );
# Each delta should be a JSON-RPC envelope with the original id.
my $first_delta = (grep { /artifact/ } @data)[0];
ok( $first_delta, 'has artifact frame' );
like( $first_delta, qr/"id":1/, 'jsonrpc id preserved' );
}
# --- ACP streaming ---
{
my $resp = post( '/runs', {
agent_name => 'm',
mode => 'stream',
input => [ { parts => [ { content_type => 'text/plain', content => 'hi' } ] } ],
});
is( $resp->code, 200, 'acp stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/event: run\.created/, 'run.created' );
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/event: message\.part/, 'message.part frames' );
like( $body, qr/event: message\.completed/, 'message.completed' );
like( $body, qr/event: run\.completed/, 'run.completed' );
like( $body, qr/"content":"hel"/, 'first chunk content' );
like( $body, qr/"content":"ld"/, 'last chunk content' );
my $parts = () = $body =~ /event: message\.part/g;
is( $parts, 4, 'four message.part events' );
}
# --- AG-UI streaming ---
{
my $resp = post( '/awp', {
threadId => 'th-1',
runId => 'run-1',
messages => [ { role => 'user', content => 'hi' } ],
});
is( $resp->code, 200, 'agui stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/"type":"RUN_STARTED"/, 'RUN_STARTED' );
t/60_psgi.t view on Meta::CPAN
handler => Langertha::Knarr::Handler::Code->new(
code => sub { 'psgi-said-hi' },
stream_code => sub { my @p = ('a','b','c'); sub { @p ? shift @p : undef } },
),
port => 0,
);
my $app = Langertha::Knarr::PSGI->new( knarr => $sb )->to_app;
my $test = Plack::Test->create($app);
# Non-streaming OpenAI
{
my $res = $test->request(
POST '/v1/chat/completions',
Content_Type => 'application/json',
Content => $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
}),
);
is( $res->code, 200, 'psgi non-stream 200' );
my $d = $json->decode($res->decoded_content);
is( $d->{choices}[0]{message}{content}, 'psgi-said-hi', 'psgi content' );
}
# Buffered streaming OpenAI
{
my $res = $test->request(
POST '/v1/chat/completions',
Content_Type => 'application/json',
Content => $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
}),
);
t/70_passthrough.t view on Meta::CPAN
use JSON::MaybeXS;
use Langertha::Knarr;
use Langertha::Knarr::Handler::Code;
use Langertha::Knarr::Handler::Passthrough;
my $json = JSON::MaybeXS->new( utf8 => 1, canonical => 1 );
my $loop = IO::Async::Loop->new;
# --- Backend: a Knarr that answers OpenAI/Anthropic/Ollama with a known string,
# plus token-by-token streaming so the passthrough's stream-extractor can
# unwrap the protocol-native chunks.
my $backend_handler = Langertha::Knarr::Handler::Code->new(
code => sub {
my ($s, $r) = @_;
my $u = $r->messages->[-1] // {};
return "BACKEND: " . ( $u->{content} // '' );
},
stream_code => sub {
my @parts = ('BACK', 'END:', ' hi');
return sub { @parts ? shift @parts : undef };
t/70_passthrough.t view on Meta::CPAN
{
my $resp = post_json(
"http://127.0.0.1:$fport/v1/messages",
{ model => 'm', messages => [ { role => 'user', content => 'sup' } ] },
);
is( $resp->code, 200, 'anthropic passthrough sync 200' );
my $d = $json->decode($resp->decoded_content);
is( $d->{content}[0]{text}, 'BACKEND: sup', 'anthropic content via passthrough' );
}
# --- 4) OpenAI streaming passthrough ---
{
my $resp = post_json(
"http://127.0.0.1:$fport/v1/chat/completions",
{ model => 'm', messages => [ { role => 'user', content => 'hi' } ], stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'openai passthrough stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE content type' );
my $body = $resp->decoded_content;
my @data = grep { /^data:/ && !/\[DONE\]/ } split /\n/, $body;
ok( scalar(@data) >= 3, 'multiple SSE chunks forwarded' );
like( $body, qr/BACK/, 'first chunk text present' );
like( $body, qr/END:/, 'second chunk text present' );
}
# --- 5) Ollama streaming passthrough ---
{
my $resp = post_json(
"http://127.0.0.1:$fport/api/chat",
{ model => 'm', messages => [ { role => 'user', content => 'hi' } ], stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'ollama passthrough stream 200' );
is( $resp->header('Content-Type'), 'application/x-ndjson', 'NDJSON content type' );
my @lines = grep { length } split /\n/, $resp->decoded_content;
ok( scalar(@lines) >= 4, 'multiple NDJSON lines forwarded' );
my $last = $json->decode($lines[-1]);
t/77_raw_passthrough_tracing.t view on Meta::CPAN
is( $start->{kind}, 'start', 'first event is start' );
is( $start->{engine}, 'passthrough', 'engine tagged as passthrough' );
is( $start->{model}, 'gpt-mystery', 'model recorded' );
is( $start->{format}, 'openai', 'format recorded' );
my $end = $tracer->events->[1];
is( $end->{kind}, 'end', 'second event is end' );
is( $end->{id}, $start->{id}, 'end matches start id' );
}
# --- 3) Raw passthrough streaming ---
{
$tracer->events->@* = (); # reset
my $resp = post_json(
"http://127.0.0.1:$fport/v1/chat/completions",
{ model => 'gpt-unknown', messages => [ { role => 'user', content => 'stream' } ],
stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'raw passthrough stream 200' );
like( $resp->decoded_content, qr/UP/, 'stream chunk present' );
like( $resp->decoded_content, qr/STREAM/, 'stream chunk 2 present' );
# Trace was created for streaming too
ok( scalar @{ $tracer->events } >= 2, 'tracer has stream events' );
is( $tracer->events->[0]{kind}, 'start', 'stream trace start' );
is( $tracer->events->[0]{model}, 'gpt-unknown', 'stream model recorded' );
is( $tracer->events->[-1]{kind}, 'end', 'stream trace end' );
}
# --- 4) Anthropic raw passthrough with auth headers ---
{
$tracer->events->@* = ();
( run in 1.682 second using v1.01-cache-2.11-cpan-600a1bdf6e4 )