langchain-hs-anthropic (empty) → 0.0.6.2
raw patch · 7 files changed
+1413/−0 lines, 7 filesdep +QuickCheckdep +aesondep +async
Dependencies added: QuickCheck, aeson, async, base, bytestring, conduit, containers, http-client, http-client-tls, http-types, langchain-hs-anthropic, langchain-hs-core, mtl, resourcet, stm, tasty, tasty-hunit, tasty-quickcheck, text, wai, warp
Files
- CHANGELOG.md +9/−0
- LICENSE +20/−0
- README.md +42/−0
- langchain-hs-anthropic.cabal +86/−0
- src/Langchain/Provider/Anthropic.hs +856/−0
- test/Spec.hs +15/−0
- test/Test/Langchain/Provider/AnthropicSpec.hs +385/−0
+ CHANGELOG.md view
@@ -0,0 +1,9 @@+# Changelog for `langchain-hs-anthropic`++## 0.0.6.2 - 2026-11-14++- Initial release of native Anthropic Messages API (`/v1/messages`) provider for `langchain-hs`.+- Added `Langchain.Provider.Anthropic` implementing `ChatModel`, `ToolBinder`, and `CacheableChatModel`.+- Added SSE streaming support (`text_delta` and fragmented `input_json_delta` accumulation, 0-arg tool handling, and `TokenUsage` reporting).+- Added prompt caching (`withPromptCaching`), extended thinking budgets (`withThinkingBudget`), and token limits (`withMaxTokens`).+
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2025-2026 Tushar Adhatrao++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,42 @@+# langchain-hs-anthropic++Native Anthropic Claude Messages API (`/v1/messages`) provider for [langchain-hs](https://hackage.haskell.org/package/langchain-hs).++Provides `Langchain.Provider.Anthropic` implementing `ChatModel`, `ToolBinder`, and `CacheableChatModel` without pulling in OpenAI, Gemini, or Ollama dependencies.++## Features++- Native `/v1/messages` client with top-level `system` instructions, multi-modal `text` and `image` blocks, `tool_use` / `tool_result` mapping, prompt caching (`cache_control`), and thinking budgets.+- Real-time SSE streaming (`message_start`, `content_block_start`, `content_block_delta` with `text_delta` & `input_json_delta`, `content_block_stop`, `message_delta`, `message_stop`) emitting `LLMStart`, `LLMChunk`, and `LLMEnd` with `TokenUsage`.+- Automatic `x-api-key` redaction in all error payloads.++## Installation++```yaml+# stack.yaml extra-deps+- langchain-hs-anthropic-0.0.6.2+```++```yaml+# package.yaml / .cabal+dependencies:+ - langchain-hs-anthropic >= 0.0.6.2 && < 0.0.7+ - langchain-hs-core >= 0.0.6.2 && < 0.0.7+```++## Usage++```haskell+{-# LANGUAGE OverloadedStrings #-}++import Control.Monad.Except (runExceptT)+import Langchain.Model+import Langchain.Provider.Anthropic++main :: IO ()+main = do+ let claude = newAnthropic "sk-ant-..." "claude-sonnet-4-5"+ res <- runExceptT $ invoke claude [userMessage "Explain algebraic effects in 2 sentences."] Nothing+ print res+```+
+ langchain-hs-anthropic.cabal view
@@ -0,0 +1,86 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.39.6.+--+-- see: https://github.com/sol/hpack++name: langchain-hs-anthropic+version: 0.0.6.2+synopsis: Anthropic Claude Messages API provider for langchain-hs+description: Native Anthropic Claude Messages API (/v1/messages) ChatModel, ToolBinder, and CacheableChatModel implementation for langchain-hs. Supports multi-modal content, tool_use/tool_result calling, extended thinking budgets, prompt caching, and SSE streaming with incremental tool_use delta accumulation.+category: AI+homepage: https://github.com/tusharad/langchain-hs#readme+bug-reports: https://github.com/tusharad/langchain-hs/issues+author: Tushar Adhatrao+maintainer: tusharadhatrao@gmail.com+copyright: 2025-2026 Tushar Adhatrao+license: MIT+license-file: LICENSE+build-type: Simple+tested-with:+ GHC == 9.12.4+ , GHC == 9.10.3+ , GHC == 9.8.4+extra-source-files:+ README.md+ CHANGELOG.md++source-repository head+ type: git+ location: https://github.com/tusharad/langchain-hs++library+ exposed-modules:+ Langchain.Provider.Anthropic+ other-modules:+ Paths_langchain_hs_anthropic+ hs-source-dirs:+ src+ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints+ build-depends:+ aeson >=2.0 && <3+ , base >=4.17 && <5+ , bytestring >=0.10 && <0.13+ , conduit ==1.3.*+ , containers >=0.6 && <0.9+ , http-client ==0.7.*+ , http-client-tls >=0.3 && <0.5+ , http-types >=0.11 && <0.13+ , langchain-hs-core >=0.0.6.2 && <0.0.7+ , mtl >=2.2 && <2.4+ , resourcet >=1.2 && <1.4+ , text >=1.2 && <3+ default-language: Haskell2010++test-suite langchain-hs-anthropic-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Test.Langchain.Provider.AnthropicSpec+ Paths_langchain_hs_anthropic+ hs-source-dirs:+ test+ ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ QuickCheck >=2.14+ , aeson+ , async+ , base >=4.17 && <5+ , bytestring+ , conduit+ , containers+ , http-client ==0.7.*+ , http-client-tls >=0.3 && <0.5+ , http-types+ , langchain-hs-anthropic+ , langchain-hs-core+ , mtl+ , resourcet+ , stm+ , tasty+ , tasty-hunit+ , tasty-quickcheck+ , text+ , wai+ , warp+ default-language: Haskell2010
+ src/Langchain/Provider/Anthropic.hs view
@@ -0,0 +1,856 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++{- |+Module : Langchain.Provider.Anthropic+Description : Native Anthropic Claude Messages API provider for langchain-hs+Copyright : (c) 2025-2026 Tushar Adhatrao+License : MIT+Maintainer : Tushar Adhatrao <tusharadhatrao@gmail.com>+Stability : experimental++Anthropic Claude Messages API (@/v1/messages@) integration implementing+'ChatModel', 'ToolBinder', and 'CacheableChatModel' with full support for+top-level @system@ prompts, multi-modal images, @tool_use@/@tool_result@ blocks,+prompt caching, extended thinking budgets, and SSE incremental tool streaming.+-}+module Langchain.Provider.Anthropic+ ( -- * Provider & Construction+ Anthropic (..)+ , newAnthropic+ , newAnthropicWithBaseUrl+ , withBaseUrl+ , withAnthropicVersion++ -- * Configuration & Combinators+ , AnthropicTool (..)+ , AnthropicConfig (..)+ , defaultAnthropicConfig+ , withMaxTokens+ , withThinkingBudget+ , withPromptCaching+ , withSystem+ , withTemperature++ -- * Request / Response Conversion & Streaming Helpers+ , messagesToAnthropic+ , redactApiKey+ , splitSseFrames+ ) where++import Control.Exception (SomeException, displayException, try)+import Control.Monad.Except (throwError)+import Control.Monad.IO.Class (liftIO)+import Data.Aeson (FromJSON (..), ToJSON (..), Value (..), object, (.:), (.:?), (.=))+import qualified Data.Aeson as Aeson+import Data.Aeson.Types (Parser, parseEither)+import Data.Bifunctor (first)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as BSC+import qualified Data.ByteString.Lazy as LBS+import Data.Conduit (bracketP, yield)+import qualified Data.List as List+import qualified Data.List.NonEmpty as NonEmpty+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Maybe (catMaybes, fromMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import GHC.Generics (Generic)+import Network.HTTP.Client+ ( BodyReader+ , HttpException+ , Request+ , RequestBody (RequestBodyLBS)+ , Response+ , brConsume+ , brRead+ , httpLbs+ , method+ , newManager+ , parseRequest+ , requestBody+ , requestHeaders+ , responseBody+ , responseClose+ , responseOpen+ , responseStatus+ )+import Network.HTTP.Client.TLS (tlsManagerSettings)+import Network.HTTP.Types.Status (statusCode, statusIsSuccessful, statusMessage)++import Langchain.Cache.Core (CacheableChatModel (..))+import Langchain.Error (LangchainError, llmError)+import Langchain.Model+import Langchain.Stream (ChatStream, StreamEvent (..), TokenUsage (..))+import qualified Langchain.Tool as CoreTool+import Langchain.Tool.Binding (ToolBinder (..))++-- | Anthropic provider configuration for the @/v1/messages@ API.+data Anthropic = Anthropic+ { anthropicApiKey :: !Text+ -- ^ Anthropic API key sent via @x-api-key@ header+ , anthropicModel :: !Text+ -- ^ Target Claude model identifier (e.g. @"claude-sonnet-4-5"@)+ , anthropicBaseUrl :: !Text+ -- ^ API base URL (default: @"https://api.anthropic.com"@)+ , anthropicVersion :: !Text+ -- ^ Value for the @anthropic-version@ header (default: @"2023-06-01"@)+ }+ deriving (Eq, Show, Generic)++-- | Tool schema formatted for the Anthropic Messages API.+data AnthropicTool = AnthropicTool+ { anthropicToolName :: !Text+ , anthropicToolDescription :: !Text+ , anthropicToolInputSchema :: !Value+ }+ deriving (Eq, Show, Generic)++instance ToJSON AnthropicTool where+ toJSON AnthropicTool {..} =+ object+ [ "name" .= anthropicToolName+ , "description" .= anthropicToolDescription+ , "input_schema" .= anthropicToolInputSchema+ ]++instance FromJSON AnthropicTool where+ parseJSON v = case v of+ Object o ->+ AnthropicTool+ <$> o .: "name"+ <*> o .: "description"+ <*> o .: "input_schema"+ _ -> fail "Expected AnthropicTool JSON object"++-- | Runtime request parameters for Anthropic Messages API calls.+data AnthropicConfig = AnthropicConfig+ { configMaxTokens :: !Int+ -- ^ Maximum output tokens (required by Anthropic API, default: 4096)+ , configTemperature :: !(Maybe Double)+ , configTopP :: !(Maybe Double)+ , configTopK :: !(Maybe Int)+ , configStopSequences :: !(Maybe [Text])+ , configSystem :: !(Maybe Text)+ , configTools :: !(Maybe [AnthropicTool])+ , configToolChoice :: !(Maybe Value)+ , configThinkingBudget :: !(Maybe Int)+ -- ^ Optional extended thinking token budget+ , configPromptCaching :: !Bool+ -- ^ Enable ephemeral @cache_control@ markers on system/tools+ }+ deriving (Eq, Show, Generic, ToJSON, FromJSON)++-- | Default configuration with @max_tokens = 4096@ and prompt caching disabled.+defaultAnthropicConfig :: AnthropicConfig+defaultAnthropicConfig =+ AnthropicConfig+ { configMaxTokens = 4096+ , configTemperature = Nothing+ , configTopP = Nothing+ , configTopK = Nothing+ , configStopSequences = Nothing+ , configSystem = Nothing+ , configTools = Nothing+ , configToolChoice = Nothing+ , configThinkingBudget = Nothing+ , configPromptCaching = False+ }++-- | Construct an 'Anthropic' provider using the default Anthropic API base URL.+newAnthropic :: Text -> Text -> Anthropic+newAnthropic key mdl =+ Anthropic+ { anthropicApiKey = key+ , anthropicModel = mdl+ , anthropicBaseUrl = "https://api.anthropic.com"+ , anthropicVersion = "2023-06-01"+ }++-- | Construct an 'Anthropic' provider targeting a custom base URL (e.g. proxy or local mock server).+newAnthropicWithBaseUrl :: Text -> Text -> Text -> Anthropic+newAnthropicWithBaseUrl key mdl url =+ (newAnthropic key mdl) {anthropicBaseUrl = url}++-- | Override the API base URL on an 'Anthropic' instance.+withBaseUrl :: Text -> Anthropic -> Anthropic+withBaseUrl url a = a {anthropicBaseUrl = url}++-- | Override the @anthropic-version@ header on an 'Anthropic' instance.+withAnthropicVersion :: Text -> Anthropic -> Anthropic+withAnthropicVersion ver a = a {anthropicVersion = ver}++-- | Configure @max_tokens@.+withMaxTokens :: Int -> AnthropicConfig -> AnthropicConfig+withMaxTokens n cfg = cfg {configMaxTokens = n}++-- | Configure extended thinking token budget (@thinking.budget_tokens@).+withThinkingBudget :: Int -> AnthropicConfig -> AnthropicConfig+withThinkingBudget n cfg = cfg {configThinkingBudget = Just n}++-- | Enable or disable ephemeral prompt caching (@cache_control@).+withPromptCaching :: Bool -> AnthropicConfig -> AnthropicConfig+withPromptCaching enabled cfg = cfg {configPromptCaching = enabled}++-- | Set explicit system instruction override.+withSystem :: Text -> AnthropicConfig -> AnthropicConfig+withSystem sys cfg = cfg {configSystem = Just sys}++-- | Set sampling temperature.+withTemperature :: Double -> AnthropicConfig -> AnthropicConfig+withTemperature t cfg = cfg {configTemperature = Just t}++-- | Redact every occurrence of the API key from error text.+redactApiKey :: Text -> Text -> Text+redactApiKey key txt+ | T.null key = txt+ | otherwise = T.replace key "[REDACTED]" txt++encodeContentBlock :: ContentBlock -> Value+encodeContentBlock (TextBlock txt) =+ object ["type" .= ("text" :: Text), "text" .= txt]+encodeContentBlock (ImageBlock ImageContent {imageSource = ImageBase64 mMime b64}) =+ object+ [ "type" .= ("image" :: Text)+ , "source"+ .= object+ [ "type" .= ("base64" :: Text)+ , "media_type" .= fromMaybe "image/png" mMime+ , "data" .= b64+ ]+ ]+encodeContentBlock (ImageBlock ImageContent {imageSource = ImageUrl url}) =+ object+ [ "type" .= ("image" :: Text)+ , "source"+ .= object+ [ "type" .= ("url" :: Text)+ , "url" .= url+ ]+ ]+encodeContentBlock (AudioBlock mime b64) =+ object+ [ "type" .= ("text" :: Text)+ , "text" .= ("[audio:" <> mime <> " len=" <> T.pack (show (T.length b64)) <> "]")+ ]+encodeContentBlock (DataBlock bs) =+ object+ ["type" .= ("text" :: Text), "text" .= ("[data:" <> T.pack (show (BS.length bs)) <> " bytes]")]++toRoleAndBlocks :: Message -> Maybe (Text, [Value])+toRoleAndBlocks msg = case messageRole msg of+ System -> Nothing+ Developer -> Nothing+ User -> Just ("user", map encodeContentBlock (NonEmpty.toList (messageContents msg)))+ Assistant ->+ let txt = extractMessageText msg+ txtBlocks = [object ["type" .= ("text" :: Text), "text" .= txt] | not (T.null txt)]+ toolBlocks =+ [ object+ [ "type" .= ("tool_use" :: Text)+ , "id" .= toolCallId tc+ , "name" .= toolCallName tc+ , "input" .= toolCallArguments tc+ ]+ | tc <- fromMaybe [] (messageToolCalls msg)+ ]+ allBlocks = txtBlocks ++ toolBlocks+ nonEmptyBlocks =+ if null allBlocks then [object ["type" .= ("text" :: Text), "text" .= ("" :: Text)]] else allBlocks+ in Just ("assistant", nonEmptyBlocks)+ Tool -> toolResultMsg+ Function -> toolResultMsg+ where+ toolResultMsg =+ Just+ ( "user"+ ,+ [ object+ [ "type" .= ("tool_result" :: Text)+ , "tool_use_id" .= fromMaybe "" (messageToolId msg)+ , "content" .= extractMessageText msg+ ]+ ]+ )++coalesceRoleBlocks :: [(Text, [Value])] -> [(Text, [Value])]+coalesceRoleBlocks = foldr step []+ where+ step (r, bs) ((r', bs') : rest)+ | r == r' = (r, bs ++ bs') : rest+ step item acc = item : acc++-- | Convert Langchain 'Message's into Anthropic top-level @system@ and @messages@ JSON values.+messagesToAnthropic :: Maybe Text -> [Message] -> (Maybe Text, [Value])+messagesToAnthropic explicitSys msgs =+ let msgSystems =+ [ extractMessageText m+ | m <- msgs+ , messageRole m == System || messageRole m == Developer+ , not (T.null (extractMessageText m))+ ]+ sysParts = maybe [] pure explicitSys ++ msgSystems+ combinedSys = if null sysParts then Nothing else Just (T.intercalate "\n\n" sysParts)+ bodyMsgs =+ [ object ["role" .= r, "content" .= bs]+ | (r, bs) <- coalesceRoleBlocks (foldr (\m acc -> maybe acc (: acc) (toRoleAndBlocks m)) [] msgs)+ ]+ in (combinedSys, bodyMsgs)++toolWithOptionalCache :: Bool -> Bool -> AnthropicTool -> Value+toolWithOptionalCache True True AnthropicTool {..} =+ object+ [ "name" .= anthropicToolName+ , "description" .= anthropicToolDescription+ , "input_schema" .= anthropicToolInputSchema+ , "cache_control" .= object ["type" .= ("ephemeral" :: Text)]+ ]+toolWithOptionalCache _ _ tool = toJSON tool++encodeToolsWithCaching :: Bool -> [AnthropicTool] -> Value+encodeToolsWithCaching _ [] = toJSON ([] :: [AnthropicTool])+encodeToolsWithCaching enableCache tools =+ let len = length tools+ encoded = zipWith (\idx t -> toolWithOptionalCache enableCache (idx == len) t) [1 ..] tools+ in toJSON encoded++buildAnthropicBody :: Anthropic -> Bool -> Maybe AnthropicConfig -> [Message] -> Value+buildAnthropicBody Anthropic {..} isStream mbCfg msgs =+ let AnthropicConfig {..} = fromMaybe defaultAnthropicConfig mbCfg+ (mbSys, encodedMsgs) = messagesToAnthropic configSystem msgs+ sysVal = case mbSys of+ Nothing -> Nothing+ Just sTxt+ | configPromptCaching ->+ Just $+ toJSON+ [ object+ [ "type" .= ("text" :: Text)+ , "text" .= sTxt+ , "cache_control" .= object ["type" .= ("ephemeral" :: Text)]+ ]+ ]+ | otherwise -> Just (String sTxt)+ basePairs =+ [ "model" .= anthropicModel+ , "max_tokens" .= configMaxTokens+ , "messages" .= encodedMsgs+ , "stream" .= isStream+ ]+ optPairs =+ catMaybes+ [ ("system" .=) <$> sysVal+ , ("temperature" .=) <$> configTemperature+ , ("top_p" .=) <$> configTopP+ , ("top_k" .=) <$> configTopK+ , ("stop_sequences" .=) <$> configStopSequences+ , (\ts -> "tools" .= encodeToolsWithCaching configPromptCaching ts) <$> configTools+ , ("tool_choice" .=) <$> configToolChoice+ , (\b -> "thinking" .= object ["type" .= ("enabled" :: Text), "budget_tokens" .= b])+ <$> configThinkingBudget+ ]+ in object (basePairs ++ optPairs)++messagesEndpoint :: Text -> Text+messagesEndpoint raw =+ let trimmed = T.dropWhileEnd (== '/') raw+ in if "/v1/messages" `T.isSuffixOf` trimmed+ then trimmed+ else+ if "/v1" `T.isSuffixOf` trimmed+ then trimmed <> "/messages"+ else trimmed <> "/v1/messages"++mkHttpRequest :: Anthropic -> Bool -> Value -> IO (Either LangchainError Request)+mkHttpRequest a@Anthropic {..} isStream body = do+ let urlStr = T.unpack (messagesEndpoint anthropicBaseUrl)+ eBaseReq <- try @SomeException (parseRequest urlStr)+ pure $ case eBaseReq of+ Left ex ->+ Left $+ llmError+ (redactApiKey anthropicApiKey $ "Invalid Anthropic URL: " <> T.pack (displayException ex))+ (Just "Anthropic")+ Nothing+ Right baseReq ->+ let _ = a+ in Right+ baseReq+ { method = "POST"+ , requestHeaders =+ [ ("x-api-key", TE.encodeUtf8 anthropicApiKey)+ , ("anthropic-version", TE.encodeUtf8 anthropicVersion)+ , ("content-type", "application/json")+ , ("accept", if isStream then "text/event-stream" else "application/json")+ ]+ , requestBody = RequestBodyLBS (Aeson.encode body)+ }++parseResponseBlock :: Value -> Parser (Maybe Text, Maybe ToolCall)+parseResponseBlock = Aeson.withObject "AnthropicContentBlock" $ \o -> do+ typ :: Text <- o .: "type"+ case typ of+ "text" -> do+ txt <- o .: "text"+ pure (Just txt, Nothing)+ "tool_use" -> do+ cid <- o .: "id"+ cname <- o .: "name"+ cinput <- o .:? "input"+ let tc = ToolCall cid "function" cname (fromMaybe (object []) cinput)+ pure (Nothing, Just tc)+ _ -> pure (Nothing, Nothing)++parseAnthropicResponse :: Value -> Either Text Message+parseAnthropicResponse =+ first T.pack+ . parseEither+ ( Aeson.withObject "AnthropicResponse" $ \o -> do+ blocks :: [Value] <- o .: "content"+ parsed <- mapM parseResponseBlock blocks+ let txts = [t | (Just t, _) <- parsed]+ tcs = [tc | (_, Just tc) <- parsed]+ fullText = T.concat txts+ mbTools = if null tcs then Nothing else Just tcs+ pure $ (assistantMessage fullText) {messageToolCalls = mbTools}+ )++httpFailureError :: Anthropic -> Response a -> LBS.ByteString -> LangchainError+httpFailureError Anthropic {..} resp bodyBytes =+ let code = statusCode (responseStatus resp)+ reason = TE.decodeUtf8 (statusMessage (responseStatus resp))+ bodyTxt = TE.decodeUtf8 (LBS.toStrict bodyBytes)+ rawMsg =+ "Anthropic HTTP "+ <> T.pack (show code)+ <> " "+ <> reason+ <> ": "+ <> bodyTxt+ in llmError (redactApiKey anthropicApiKey rawMsg) (Just "Anthropic") Nothing++-- | Split a buffered byte chunk into complete SSE frame payloads (delimited by blank lines) and trailing remainder.+splitSseFrames :: BS.ByteString -> ([BS.ByteString], BS.ByteString)+splitSseFrames raw =+ let normalized = BSC.pack (stripCr (BSC.unpack raw))+ in go normalized []+ where+ stripCr [] = []+ stripCr ('\r' : '\n' : xs) = '\n' : stripCr xs+ stripCr (x : xs) = x : stripCr xs++ go buf acc+ | BS.null buf = (reverse acc, BS.empty)+ | otherwise =+ let (frame, restWithSep) = BS.breakSubstring "\n\n" buf+ in if BS.null restWithSep+ then (reverse acc, buf)+ else+ let nextBuf = BS.drop 2 restWithSep+ nextAcc = if BS.null frame then acc else frame : acc+ in go nextBuf nextAcc++data ActiveBlock+ = ActiveText+ | ActiveTool !Text !Text ![Text]++data AnthropicSseState = AnthropicSseState+ { sseRunId :: !Text+ , sseModel :: !Text+ , sseStarted :: !Bool+ , sseEnded :: !Bool+ , sseAccText :: ![Text]+ , sseActiveBlocks :: !(Map Int ActiveBlock)+ , sseTools :: ![ToolCall]+ , sseInputTokens :: !Int+ , sseOutputTokens :: !Int+ }++initialSseState :: Text -> AnthropicSseState+initialSseState mdl =+ AnthropicSseState+ { sseRunId = "anthropic-stream"+ , sseModel = mdl+ , sseStarted = False+ , sseEnded = False+ , sseAccText = []+ , sseActiveBlocks = Map.empty+ , sseTools = []+ , sseInputTokens = 0+ , sseOutputTokens = 0+ }++parseRawSseFrame :: BS.ByteString -> (Maybe Text, Maybe BS.ByteString)+parseRawSseFrame frame =+ let ls = BSC.lines frame+ in List.foldl' step (Nothing, Nothing) ls+ where+ step (mEv, mData) line+ | "event:" `BS.isPrefixOf` line =+ let val = T.strip (TE.decodeUtf8 (BS.drop 6 line))+ in (Just val, mData)+ | "data:" `BS.isPrefixOf` line =+ let chunk = BS.dropWhile (== 32) (BS.drop 5 line)+ nextData = case mData of+ Nothing -> Just chunk+ Just prev -> Just (prev <> "\n" <> chunk)+ in (mEv, nextData)+ | otherwise = (mEv, mData)++buildFinalStreamMessage :: AnthropicSseState -> Message+buildFinalStreamMessage AnthropicSseState {..} =+ let fullTxt = T.concat (reverse sseAccText)+ toolsRev = reverse sseTools+ mbTools = if null toolsRev then Nothing else Just toolsRev+ in (assistantMessage fullTxt) {messageToolCalls = mbTools}++currentTokenUsage :: AnthropicSseState -> TokenUsage+currentTokenUsage AnthropicSseState {..} =+ TokenUsage+ { promptTokens = sseInputTokens+ , completionTokens = sseOutputTokens+ , totalTokens = sseInputTokens + sseOutputTokens+ }++ensureStarted :: AnthropicSseState -> [Message] -> ([StreamEvent], AnthropicSseState)+ensureStarted st@AnthropicSseState {..} inputMsgs+ | sseStarted = ([], st)+ | otherwise =+ ( [LLMStart sseRunId sseModel inputMsgs]+ , st {sseStarted = True}+ )++processSseFrame ::+ Text ->+ [Message] ->+ AnthropicSseState ->+ BS.ByteString ->+ Either LangchainError ([StreamEvent], AnthropicSseState)+processSseFrame apiKey inputMsgs st frame =+ case parseRawSseFrame frame of+ (_, Nothing) -> Right ([], st)+ (mbEvName, Just rawJson) ->+ case Aeson.eitherDecodeStrict rawJson of+ Left err ->+ Left $+ llmError+ (redactApiKey apiKey $ "Invalid Anthropic SSE JSON: " <> T.pack err)+ (Just "Anthropic")+ Nothing+ Right (Object obj) ->+ let evType =+ case parseEither (.: "type") obj of+ Right t -> t+ Left _ -> fromMaybe "" mbEvName+ in handleSseObject apiKey inputMsgs st evType obj+ Right _ ->+ Left $ llmError "Expected JSON object in Anthropic SSE event" (Just "Anthropic") Nothing++handleSseObject ::+ Text ->+ [Message] ->+ AnthropicSseState ->+ Text ->+ Aeson.Object ->+ Either LangchainError ([StreamEvent], AnthropicSseState)+handleSseObject apiKey inputMsgs st evType obj = case evType of+ "ping" -> Right ([], st)+ "error" ->+ let errTxt = case parseEither (\o -> (o .: "error") >>= (.: "message")) obj of+ Right m -> m+ Left _ -> TE.decodeUtf8 (LBS.toStrict (Aeson.encode obj))+ in Left $ llmError (redactApiKey apiKey errTxt) (Just "Anthropic") Nothing+ "message_start" ->+ case parseEither parseMsgStart obj of+ Left err ->+ Left $+ llmError ("Failed to parse Anthropic message_start: " <> T.pack err) (Just "Anthropic") Nothing+ Right (mbId, mbMdl, mbInTok) ->+ let st1 =+ st+ { sseRunId = fromMaybe (sseRunId st) mbId+ , sseModel = fromMaybe (sseModel st) mbMdl+ , sseInputTokens = fromMaybe (sseInputTokens st) mbInTok+ }+ (startEvs, st2) = ensureStarted st1 inputMsgs+ in Right (startEvs, st2)+ "content_block_start" ->+ let (startEvs, st1) = ensureStarted st inputMsgs+ in case parseEither parseBlockStart obj of+ Left err ->+ Left $+ llmError+ ("Failed to parse Anthropic content_block_start: " <> T.pack err)+ (Just "Anthropic")+ Nothing+ Right (idx, bTyp, mbInitText, mbToolMeta) ->+ case bTyp of+ "tool_use" ->+ let (tid, tname) = fromMaybe ("", "") mbToolMeta+ st2 = st1 {sseActiveBlocks = Map.insert idx (ActiveTool tid tname []) (sseActiveBlocks st1)}+ in Right (startEvs, st2)+ _ ->+ let st2 = st1 {sseActiveBlocks = Map.insert idx ActiveText (sseActiveBlocks st1)}+ initEvs = case mbInitText of+ Just txt | not (T.null txt) -> [LLMChunk (sseRunId st2) txt Nothing]+ _ -> []+ st3 = case mbInitText of+ Just txt | not (T.null txt) -> st2 {sseAccText = txt : sseAccText st2}+ _ -> st2+ in Right (startEvs ++ initEvs, st3)+ "content_block_delta" ->+ let (startEvs, st1) = ensureStarted st inputMsgs+ in case parseEither parseBlockDelta obj of+ Left err ->+ Left $+ llmError+ ("Failed to parse Anthropic content_block_delta: " <> T.pack err)+ (Just "Anthropic")+ Nothing+ Right (idx, dTyp, mbTxt, mbJsonFrag) -> case dTyp of+ "text_delta" ->+ let txt = fromMaybe "" mbTxt+ st2 = st1 {sseAccText = txt : sseAccText st1}+ in Right (startEvs ++ [LLMChunk (sseRunId st2) txt Nothing], st2)+ "input_json_delta" ->+ let frag = fromMaybe "" mbJsonFrag+ upd (ActiveTool tid tname fs) = ActiveTool tid tname (frag : fs)+ upd other = other+ st2 = st1 {sseActiveBlocks = Map.adjust upd idx (sseActiveBlocks st1)}+ in Right (startEvs, st2)+ _ -> Right (startEvs, st1)+ "content_block_stop" ->+ let (startEvs, st1) = ensureStarted st inputMsgs+ in case parseEither (.: "index") obj of+ Left err ->+ Left $+ llmError ("Failed to parse Anthropic content_block_stop: " <> T.pack err) (Just "Anthropic") Nothing+ Right idx -> case Map.lookup idx (sseActiveBlocks st1) of+ Just (ActiveTool tid tname frags) ->+ let rawToolJson = T.concat (reverse frags)+ parsedArgs+ | T.null (T.strip rawToolJson) = Right (object [])+ | otherwise = Aeson.eitherDecodeStrict (TE.encodeUtf8 rawToolJson)+ in case parsedArgs of+ Left jErr ->+ Left $+ llmError+ ( redactApiKey apiKey $+ "Invalid JSON in completed tool_use block '" <> tname <> "': " <> T.pack jErr+ )+ (Just "Anthropic")+ Nothing+ Right argsVal ->+ let tc = ToolCall tid "function" tname argsVal+ st2 =+ st1+ { sseActiveBlocks = Map.delete idx (sseActiveBlocks st1)+ , sseTools = tc : sseTools st1+ }+ in Right (startEvs ++ [LLMChunk (sseRunId st2) "" (Just tc)], st2)+ _ ->+ let st2 = st1 {sseActiveBlocks = Map.delete idx (sseActiveBlocks st1)}+ in Right (startEvs, st2)+ "message_delta" ->+ case parseEither parseMsgDelta obj of+ Left err ->+ Left $+ llmError ("Failed to parse Anthropic message_delta: " <> T.pack err) (Just "Anthropic") Nothing+ Right (mbInTok, mbOutTok) ->+ let st1 =+ st+ { sseInputTokens = fromMaybe (sseInputTokens st) mbInTok+ , sseOutputTokens = fromMaybe (sseOutputTokens st) mbOutTok+ }+ in Right ([], st1)+ "message_stop" ->+ let (startEvs, st1) = ensureStarted st inputMsgs+ in if sseEnded st1+ then Right (startEvs, st1)+ else+ let endEv = LLMEnd (sseRunId st1) (buildFinalStreamMessage st1) (Just (currentTokenUsage st1))+ in Right (startEvs ++ [endEv], st1 {sseEnded = True})+ _ -> Right ([], st)+ where+ parseMsgStart o = do+ msg <- o .: "message"+ mbId <- msg .:? "id"+ mbMdl <- msg .:? "model"+ mbUsage <- msg .:? "usage"+ mbInTok <- maybe (pure Nothing) (.:? "input_tokens") mbUsage+ pure (mbId, mbMdl, mbInTok)++ parseBlockStart o = do+ idx <- o .: "index"+ cb <- o .: "content_block"+ bTyp <- cb .: "type"+ mbText <- cb .:? "text"+ mbTool <- case (bTyp :: Text) of+ "tool_use" -> do+ tid <- cb .: "id"+ tname <- cb .: "name"+ pure (Just (tid, tname))+ _ -> pure Nothing+ pure (idx, bTyp, mbText, mbTool)++ parseBlockDelta o = do+ idx <- o .: "index"+ d <- o .: "delta"+ dTyp :: Text <- d .: "type"+ mbTxt <- d .:? "text"+ mbJson <- d .:? "partial_json"+ pure (idx, dTyp, mbTxt, mbJson)++ parseMsgDelta o = do+ mbUsage <- o .:? "usage"+ case mbUsage of+ Nothing -> pure (Nothing, Nothing)+ Just u -> (,) <$> u .:? "input_tokens" <*> u .:? "output_tokens"++streamSseBody ::+ Anthropic ->+ [Message] ->+ Response BodyReader ->+ ChatStream+streamSseBody provider inputMsgs resp = do+ if not (statusIsSuccessful (responseStatus resp))+ then do+ errBody <- LBS.fromChunks <$> liftIO (brConsume (responseBody resp))+ throwError $ httpFailureError provider resp errBody+ else loop BS.empty (initialSseState (anthropicModel provider))+ where+ apiKey = anthropicApiKey provider+ reader = responseBody resp++ emitFrames _ st [] = pure (Right st)+ emitFrames _ st (f : fs) =+ case processSseFrame apiKey inputMsgs st f of+ Left err -> pure (Left err)+ Right (evs, st') -> do+ mapM_ yield evs+ emitFrames apiKey st' fs++ loop buf st = do+ eChunk <- liftIO $ try @SomeException (brRead reader)+ case eChunk of+ Left ex ->+ throwError $+ llmError+ (redactApiKey apiKey $ "Anthropic stream transport error: " <> T.pack (displayException ex))+ (Just "Anthropic")+ Nothing+ Right chunk+ | BS.null chunk -> do+ let (finalFrames, _) = splitSseFrames (buf <> "\n\n")+ eFinalSt <- emitFrames apiKey st finalFrames+ case eFinalSt of+ Left err -> throwError err+ Right st' ->+ if sseEnded st'+ then pure ()+ else do+ let (startEvs, stStarted) = ensureStarted st' inputMsgs+ mapM_ yield startEvs+ yield $+ LLMEnd+ (sseRunId stStarted)+ (buildFinalStreamMessage stStarted)+ (Just (currentTokenUsage stStarted))+ | otherwise -> do+ let (frames, nextBuf) = splitSseFrames (buf <> chunk)+ eNextSt <- emitFrames apiKey st frames+ case eNextSt of+ Left err -> throwError err+ Right st' -> loop nextBuf st'++instance ChatModel Anthropic where+ type ModelConfig Anthropic = AnthropicConfig++ invoke provider msgs mbCfg = do+ let reqJson = buildAnthropicBody provider False mbCfg msgs+ apiKey = anthropicApiKey provider+ eReq <- liftIO $ mkHttpRequest provider False reqJson+ req <- case eReq of+ Left err -> throwError err+ Right r -> pure r+ eResp <- liftIO $ try @HttpException $ do+ mgr <- newManager tlsManagerSettings+ httpLbs req mgr+ case eResp of+ Left ex ->+ throwError $+ llmError+ (redactApiKey apiKey $ "Anthropic request error: " <> T.pack (displayException ex))+ (Just "Anthropic")+ Nothing+ Right resp ->+ if not (statusIsSuccessful (responseStatus resp))+ then throwError $ httpFailureError provider resp (responseBody resp)+ else case Aeson.eitherDecode (responseBody resp) of+ Left err ->+ throwError $+ llmError+ (redactApiKey apiKey $ "Invalid Anthropic response JSON: " <> T.pack err)+ (Just "Anthropic")+ Nothing+ Right val -> case parseAnthropicResponse val of+ Left err ->+ throwError $+ llmError+ (redactApiKey apiKey $ "Failed to decode Anthropic message: " <> err)+ (Just "Anthropic")+ Nothing+ Right msg -> pure msg++ stream provider msgs mbCfg = do+ let reqJson = buildAnthropicBody provider True mbCfg msgs+ eReq <- liftIO $ mkHttpRequest provider True reqJson+ req <- case eReq of+ Left err -> throwError err+ Right r -> pure r+ mgr <- liftIO $ newManager tlsManagerSettings+ bracketP (responseOpen req mgr) responseClose (streamSseBody provider msgs)++instance ToolBinder Anthropic m where+ bindToolsConfig tools mbCfg =+ let cfg = fromMaybe defaultAnthropicConfig mbCfg+ converted =+ [ AnthropicTool+ { anthropicToolName = CoreTool.toolName t+ , anthropicToolDescription = CoreTool.toolDescription t+ , anthropicToolInputSchema = CoreTool.toolSchema t+ }+ | t <- tools+ ]+ existing = fromMaybe [] (configTools cfg)+ existingNames = map anthropicToolName existing+ merged = existing ++ filter (\t -> anthropicToolName t `notElem` existingNames) converted+ in Just $+ if null merged+ then cfg+ else cfg {configTools = Just merged}++instance CacheableChatModel Anthropic where+ cacheModelIdentity Anthropic {..} mbCfg =+ let cfg = fromMaybe defaultAnthropicConfig mbCfg+ in object+ [ "provider" .= ("anthropic" :: Text)+ , "model" .= anthropicModel+ , "baseUrl" .= anthropicBaseUrl+ , "version" .= anthropicVersion+ , "maxTokens" .= configMaxTokens cfg+ , "system" .= configSystem cfg+ , "temperature" .= configTemperature cfg+ , "thinkingBudget" .= configThinkingBudget cfg+ , "promptCaching" .= configPromptCaching cfg+ , "tools" .= configTools cfg+ ]
+ test/Spec.hs view
@@ -0,0 +1,15 @@+module Main (main) where++import Test.Tasty++import qualified Test.Langchain.Provider.AnthropicSpec as AnthropicSpec++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests =+ testGroup+ "langchain-hs-anthropic"+ [ AnthropicSpec.tests+ ]
+ test/Test/Langchain/Provider/AnthropicSpec.hs view
@@ -0,0 +1,385 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module Test.Langchain.Provider.AnthropicSpec (tests) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, wait)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, tryReadMVar)+import Control.Concurrent.STM (atomically, modifyTVar', newTVarIO, readTVarIO)+import Control.Exception (SomeException, catch)+import Control.Monad.Except (runExceptT)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Trans.Resource (runResourceT)+import Data.Aeson (ToJSON (..), Value (..), decodeStrict, encode, object, (.=))+import qualified Data.Aeson.KeyMap as KM+import qualified Data.ByteString as BS+import qualified Data.ByteString.Builder as Builder+import qualified Data.ByteString.Lazy as LBS+import Data.Conduit (runConduit, (.|))+import qualified Data.Conduit.List as CL+import Data.List.NonEmpty (NonEmpty (..))+import qualified Data.Map.Strict as Map+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Network.HTTP.Types (HeaderName, hContentType, status200, status401)+import Network.Wai (Application, requestHeaders, responseLBS, responseStream, strictRequestBody)+import Network.Wai.Handler.Warp (testWithApplication)+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck++import Langchain.Cache.Core (CacheableChatModel (..))+import Langchain.Error (errorMessage)+import Langchain.Model+import Langchain.Stream (StreamEvent (..), TokenUsage (..), collectEvents, liftChatStreamResource)+import Langchain.Tool (Tool, createTool)+import Langchain.Tool.Binding (ToolBinder (..))++import Langchain.Provider.Anthropic++withTestServer :: Application -> (Text -> IO a) -> IO a+withTestServer app action =+ testWithApplication (pure app) $ \port ->+ action $ "http://127.0.0.1:" <> T.pack (show port)++sseEvent :: BS.ByteString -> LBS.ByteString -> LBS.ByteString+sseEvent evName payload =+ "event: " <> LBS.fromStrict evName <> "\ndata: " <> payload <> "\n\n"++findHeader :: HeaderName -> [(HeaderName, BS.ByteString)] -> Maybe BS.ByteString+findHeader = lookup++tests :: TestTree+tests =+ testGroup+ "Langchain.Provider.Anthropic"+ [ testCase "newAnthropic sets default base URL and version" $ do+ let p = newAnthropic "sk-ant-secret" "claude-sonnet-4-5"+ anthropicApiKey p @?= "sk-ant-secret"+ anthropicModel p @?= "claude-sonnet-4-5"+ anthropicBaseUrl p @?= "https://api.anthropic.com"+ anthropicVersion p @?= "2023-06-01"+ , testCase "ToolBinder merges tools as Anthropic input_schema definitions" $ do+ let weatherTool :: Tool IO+ weatherTool =+ createTool+ "get_weather"+ "Fetch weather"+ (object ["type" .= ("object" :: Text), "properties" .= object []])+ (const $ pure $ Right "sunny")+ cfg = bindToolsConfig @Anthropic [weatherTool] (Just defaultAnthropicConfig)+ fmap (length . fromMaybe [] . configTools) cfg @?= Just 1+ case cfg >>= configTools of+ Just [t] -> do+ anthropicToolName t @?= "get_weather"+ anthropicToolDescription t @?= "Fetch weather"+ other -> assertFailure $ "Expected 1 tool, got: " ++ show other+ , testCase "cacheModelIdentity distinguishes configs cleanly" $ do+ let p = newAnthropic "sk-ant-1" "claude-sonnet-4-5"+ id1 = cacheModelIdentity p (Just defaultAnthropicConfig)+ id2 = cacheModelIdentity p (Just $ withThinkingBudget 2048 defaultAnthropicConfig)+ id3 = cacheModelIdentity p (Just $ withPromptCaching True defaultAnthropicConfig)+ assertBool "Thinking budget changes identity" (id1 /= id2)+ assertBool "Prompt caching changes identity" (id1 /= id3)+ , testCase "invoke sends expected Anthropic JSON, headers, system, tool_use -> tool_result" $ do+ capturedReq <- newEmptyMVar+ let respJson =+ object+ [ "id" .= ("msg_123" :: Text)+ , "type" .= ("message" :: Text)+ , "role" .= ("assistant" :: Text)+ , "model" .= ("claude-sonnet-4-5" :: Text)+ , "content"+ .= [ object ["type" .= ("text" :: Text), "text" .= ("Checking..." :: Text)]+ , object+ [ "type" .= ("tool_use" :: Text)+ , "id" .= ("toolu_1" :: Text)+ , "name" .= ("get_weather" :: Text)+ , "input" .= object ["city" .= ("Paris" :: Text)]+ ]+ ]+ , "usage" .= object ["input_tokens" .= (11 :: Int), "output_tokens" .= (7 :: Int)]+ ]+ validApp :: Application+ validApp req respond = do+ body <- strictRequestBody req+ putMVar capturedReq (requestHeaders req, body)+ respond $+ responseLBS+ status200+ [(hContentType, "application/json")]+ (encode respJson)+ let tc = ToolCall "toolu_0" "function" "lookup" (object ["q" .= ("hs" :: Text)])+ assistantTurn = (assistantMessage "Searching") {messageToolCalls = Just [tc]}+ toolTurn = (textMessage Tool "found") {messageToolId = Just "toolu_0"}+ cfg =+ withPromptCaching True $+ withThinkingBudget 1024 $+ withMaxTokens 2048 defaultAnthropicConfig+ res <- withTestServer validApp $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-test-key" "claude-sonnet-4-5" base+ runExceptT $+ invoke provider [systemMessage "Be concise.", userMessage "Hi", assistantTurn, toolTurn] (Just cfg)+ (hdrs, rawBody) <- takeMVar capturedReq+ findHeader "x-api-key" hdrs @?= Just "sk-ant-test-key"+ findHeader "anthropic-version" hdrs @?= Just "2023-06-01"+ case decodeStrict (LBS.toStrict rawBody) of+ Just (Object o) -> do+ KM.lookup "max_tokens" o @?= Just (toJSON (2048 :: Int))+ KM.lookup "thinking" o+ @?= Just (object ["type" .= ("enabled" :: Text), "budget_tokens" .= (1024 :: Int)])+ KM.lookup "system" o+ @?= Just+ ( toJSON+ [ object+ [ "type" .= ("text" :: Text)+ , "text" .= ("Be concise." :: Text)+ , "cache_control" .= object ["type" .= ("ephemeral" :: Text)]+ ]+ ]+ )+ case KM.lookup "messages" o of+ Just (Array msgs) -> length msgs @?= 3+ other -> assertFailure $ "Expected 3 messages, got: " ++ show other+ other -> assertFailure $ "Expected JSON object body, got: " ++ show other+ case res of+ Left err -> assertFailure $ "Expected invoke success, got: " ++ show err+ Right msg -> do+ extractMessageText msg @?= "Checking..."+ messageToolCalls msg+ @?= Just [ToolCall "toolu_1" "function" "get_weather" (object ["city" .= ("Paris" :: Text)])]+ , testCase "invoke redacts x-api-key on HTTP 4xx errors" $ do+ let secret = "sk-ant-very-secret-token"+ errApp :: Application+ errApp _ respond =+ respond $+ responseLBS+ status401+ [(hContentType, "application/json")]+ (LBS.fromStrict $ TE.encodeUtf8 $ "Authentication failed for key " <> secret)+ res <- withTestServer errApp $ \base -> do+ let provider = newAnthropicWithBaseUrl secret "claude-sonnet-4-5" base+ runExceptT $ invoke provider [userMessage "Hi"] Nothing+ case res of+ Left err -> do+ assertBool "API key must be redacted" (not (secret `T.isInfixOf` errorMessage err))+ assertBool "Contains REDACTED marker" ("[REDACTED]" `T.isInfixOf` errorMessage err)+ Right msg -> assertFailure $ "Expected 401 error, got: " ++ show msg+ , testCase "messagesToAnthropic encodes ImageBase64 and ImageUrl blocks" $ do+ let imgB64 = imageMessage User "image/jpeg" "AQID"+ imgUrlMsg =+ Message+ User+ (ImageBlock (ImageContent (ImageUrl "https://example.com/a.png") Nothing Nothing) :| [])+ Nothing+ Nothing+ Nothing+ Map.empty+ (sys, msgs) = messagesToAnthropic Nothing [imgB64, imgUrlMsg]+ sys @?= Nothing+ length msgs @?= 1+ , testCase "stream parses SSE text_delta, fragmented input_json_delta, and TokenUsage" $ do+ let frames =+ [ sseEvent+ "message_start"+ "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_s1\",\"model\":\"claude-sonnet-4-5\",\"usage\":{\"input_tokens\":10,\"output_tokens\":0}}}"+ , sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}"+ , sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Checking \"}}"+ , sseEvent "content_block_stop" "{\"type\":\"content_block_stop\",\"index\":0}"+ , sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_9\",\"name\":\"calc\"}}"+ , sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"x\\\":\"}}"+ , sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"42}\"}}"+ , sseEvent "content_block_stop" "{\"type\":\"content_block_stop\",\"index\":1}"+ , sseEvent+ "message_delta"+ "{\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":6}}"+ , sseEvent "message_stop" "{\"type\":\"message_stop\"}"+ ]+ app :: Application+ app _ respond =+ respond $+ responseStream status200 [(hContentType, "text/event-stream")] $ \write flush ->+ mapM_ (\f -> write (Builder.lazyByteString f) >> flush) frames+ res <- withTestServer app $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-1" "claude-sonnet-4-5" base+ runResourceT $ runExceptT $ collectEvents (stream provider [userMessage "Compute"] Nothing)+ case res of+ Left err -> assertFailure $ "Expected stream success, got: " ++ show err+ Right events -> do+ let expectedTc = ToolCall "toolu_9" "function" "calc" (object ["x" .= (42 :: Int)])+ events+ @?= [ LLMStart "msg_s1" "claude-sonnet-4-5" [userMessage "Compute"]+ , LLMChunk "msg_s1" "Checking " Nothing+ , LLMChunk "msg_s1" "" (Just expectedTc)+ , LLMEnd+ "msg_s1"+ ((assistantMessage "Checking ") {messageToolCalls = Just [expectedTc]})+ (Just (TokenUsage 10 6 16))+ ]+ , testCase "stream handles 0-arg tool_use block with empty input_json_delta" $ do+ let frames =+ [ sseEvent+ "message_start"+ "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_z\",\"usage\":{\"input_tokens\":3}}}"+ , sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_z\",\"name\":\"ping\"}}"+ , sseEvent "content_block_stop" "{\"type\":\"content_block_stop\",\"index\":0}"+ , sseEvent "message_delta" "{\"type\":\"message_delta\",\"usage\":{\"output_tokens\":2}}"+ , sseEvent "message_stop" "{\"type\":\"message_stop\"}"+ ]+ app :: Application+ app _ respond =+ respond $+ responseStream status200 [(hContentType, "text/event-stream")] $ \write flush ->+ mapM_ (\f -> write (Builder.lazyByteString f) >> flush) frames+ res <- withTestServer app $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-1" "claude-sonnet-4-5" base+ runResourceT $ runExceptT $ collectEvents (stream provider [userMessage "Ping"] Nothing)+ case res of+ Left err -> assertFailure $ "Expected stream success, got: " ++ show err+ Right events -> do+ let deltas = [tc | LLMChunk _ _ (Just tc) <- events]+ deltas @?= [ToolCall "toolu_z" "function" "ping" (object [])]+ , testCase "stream delivers first LLMChunk before gated SSE response finishes" $ do+ releaseTail <- newEmptyMVar+ seenChunks <- newTVarIO ([] :: [Text])+ let headFrame =+ sseEvent "message_start" "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_gate\"}}"+ <> sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\"}}"+ <> sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"first \"}}"+ tailFrames =+ [ sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"second\"}}"+ , sseEvent "message_stop" "{\"type\":\"message_stop\"}"+ ]+ app :: Application+ app _ respond =+ respond $+ responseStream status200 [(hContentType, "text/event-stream")] $ \write flush -> do+ write (Builder.lazyByteString headFrame)+ flush+ takeMVar releaseTail+ mapM_ (\f -> write (Builder.lazyByteString f) >> flush) tailFrames+ res <- withTestServer app $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-1" "claude-sonnet-4-5" base+ runner <-+ async $+ runResourceT $+ runExceptT $+ runConduit $+ stream provider [userMessage "Gated"] Nothing+ .| CL.mapM_+ ( \case+ LLMChunk _ txt Nothing+ | not (T.null txt) ->+ liftIO $ atomically $ modifyTVar' seenChunks (++ [txt])+ _ -> pure ()+ )+ let waitFirst :: Int -> IO ()+ waitFirst 0 = assertFailure "Timed out waiting for first gated chunk"+ waitFirst k = do+ xs <- readTVarIO seenChunks+ if xs == ["first "] then pure () else threadDelay 10000 >> waitFirst (k - 1)+ waitFirst 100+ putMVar releaseTail ()+ wait runner+ case res of+ Left err -> assertFailure $ "Expected stream success, got: " ++ show err+ Right () -> pure ()+ finalChunks <- readTVarIO seenChunks+ finalChunks @?= ["first ", "second"]+ , testCase "stream raises LLMError on invalid tool_use JSON" $ do+ let frames =+ [ sseEvent "message_start" "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_bad\"}}"+ , sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_bad\",\"name\":\"broken\"}}"+ , sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{not valid json\"}}"+ , sseEvent "content_block_stop" "{\"type\":\"content_block_stop\",\"index\":0}"+ ]+ app :: Application+ app _ respond =+ respond $+ responseStream status200 [(hContentType, "text/event-stream")] $ \write flush ->+ mapM_ (\f -> write (Builder.lazyByteString f) >> flush) frames+ res <- withTestServer app $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-1" "claude-sonnet-4-5" base+ runResourceT $ runExceptT $ collectEvents (stream provider [userMessage "Broken"] Nothing)+ case res of+ Left _ -> pure ()+ Right evs -> assertFailure $ "Expected error on invalid tool JSON, got: " ++ show evs+ , testCase "early consumer close tears down SSE HTTP connection" $ do+ disconnected <- newEmptyMVar+ let firstFrame =+ sseEvent "message_start" "{\"type\":\"message_start\",\"message\":{\"id\":\"msg_disc\"}}"+ <> sseEvent+ "content_block_start"+ "{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\"}}"+ <> sseEvent+ "content_block_delta"+ "{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}"+ app :: Application+ app _ respond =+ respond $+ responseStream status200 [(hContentType, "text/event-stream")] $ \write flush -> do+ let loop = do+ write ": keepalive\n\n"+ flush+ threadDelay 5000+ loop+ write (Builder.lazyByteString firstFrame)+ flush+ loop `catch` (\(_ :: SomeException) -> putMVar disconnected ())+ _ <- withTestServer app $ \base -> do+ let provider = newAnthropicWithBaseUrl "sk-ant-1" "claude-sonnet-4-5" base+ runResourceT $+ runExceptT $+ runConduit $+ liftChatStreamResource (stream provider [userMessage "Disconnect"] Nothing)+ .| CL.take 2+ threadDelay 100000+ mbDisc <- tryReadMVar disconnected+ mbDisc @?= Just ()+ , testProperty "splitSseFrames preserves frames under arbitrary chunk splits" $+ \(Positive chunkLenRaw) ->+ let frame1 = "event: ping\ndata: {\"a\":1}"+ frame2 = "event: message_stop\ndata: {\"b\":2}"+ full = frame1 <> "\n\n" <> frame2 <> "\n\n"+ stepSize = max 1 (chunkLenRaw `mod` 13)+ chunks = chunkEvery stepSize full+ (parsed, remBuf) = feedAll chunks+ in parsed === [frame1, frame2] .&&. remBuf === BS.empty+ ]+ where+ chunkEvery n bs+ | BS.null bs = []+ | otherwise = let (h, t) = BS.splitAt n bs in h : chunkEvery n t+ feedAll = go BS.empty []+ where+ go buf acc [] = (concat (reverse acc), buf)+ go buf acc (c : cs) =+ let (fs, buf') = splitSseFrames (buf <> c)+ in go buf' (fs : acc) cs