Skip to content

anthropic.calls

A module for calling Anthropic's Claude API.

AnthropicCall

Bases: BaseCall[AnthropicCallResponse, AnthropicCallResponseChunk, AnthropicTool]

A base class for calling Anthropic's Claude models.

Example:

from mirascope.anthropic import AnthropicCall


class BookRecommender(AnthropicCall):
    prompt_template = "Please recommend a {genre} book."

    genre: str


response = BookRecommender(genre="fantasy").call()
print(response.content)
#> There are many great books to read, it ultimately depends...
Source code in mirascope/anthropic/calls.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class AnthropicCall(
    BaseCall[AnthropicCallResponse, AnthropicCallResponseChunk, AnthropicTool]
):
    """A base class for calling Anthropic's Claude models.

    Example:

    ```python
    from mirascope.anthropic import AnthropicCall


    class BookRecommender(AnthropicCall):
        prompt_template = "Please recommend a {genre} book."

        genre: str


    response = BookRecommender(genre="fantasy").call()
    print(response.content)
    #> There are many great books to read, it ultimately depends...
    ```
    """

    call_params: ClassVar[AnthropicCallParams] = AnthropicCallParams()

    def messages(self) -> list[MessageParam]:
        """Returns the template as a formatted list of messages."""
        return self._parse_messages(
            [MessageRole.SYSTEM, MessageRole.USER, MessageRole.ASSISTANT]
        )  # type: ignore

    @retry
    def call(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> AnthropicCallResponse:
        """Makes a call to the model using this `AnthropicCall` instance.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `AnthropicCallResponse` instance.
        """
        messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
        client = Anthropic(api_key=self.api_key, base_url=self.base_url)
        if self.call_params.wrapper is not None:
            client = self.call_params.wrapper(client)

        create = client.messages.create
        if tool_types:
            create = client.beta.tools.messages.create  # type: ignore
        if self.call_params.weave is not None:
            create = self.call_params.weave(create)  # pragma: no cover
        if self.call_params.logfire:
            create = self.call_params.logfire(
                create,
                "anthropic",
                response_type=AnthropicCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover
        start_time = datetime.datetime.now().timestamp() * 1000
        message = create(
            messages=messages,
            stream=False,
            **kwargs,
        )
        return AnthropicCallResponse(
            response=message,
            tool_types=tool_types,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            cost=anthropic_api_calculate_cost(message.usage, message.model),
            response_format=self.call_params.response_format,
        )

    @retry
    async def call_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> AnthropicCallResponse:
        """Makes an asynchronous call to the model using this `AnthropicCall` instance.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Returns:
            A `AnthropicCallResponse` instance.
        """
        messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
        client = AsyncAnthropic(api_key=self.api_key, base_url=self.base_url)
        if self.call_params.wrapper_async is not None:
            client = self.call_params.wrapper_async(client)
        create = client.messages.create
        if tool_types:
            create = client.beta.tools.messages.create  # type: ignore
        if self.call_params.weave is not None:
            create = self.call_params.weave(create)  # pragma: no cover
        if self.call_params.logfire_async:
            create = self.call_params.logfire_async(
                create,
                "anthropic",
                response_type=AnthropicCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover
        start_time = datetime.datetime.now().timestamp() * 1000
        message = await create(
            messages=messages,
            stream=False,
            **kwargs,
        )
        return AnthropicCallResponse(
            response=message,
            tool_types=tool_types,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            cost=anthropic_api_calculate_cost(message.usage, message.model),
            response_format=self.call_params.response_format,
        )

    @retry
    def stream(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> Generator[AnthropicCallResponseChunk, None, None]:
        """Streams the response for a call using this `AnthropicCall`.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            An `AnthropicCallResponseChunk` for each chunk of the response.
        """
        messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
        client = Anthropic(api_key=self.api_key, base_url=self.base_url)
        if self.call_params.wrapper is not None:
            client = self.call_params.wrapper(client)
        if self.call_params.logfire:  # pragma: no cover
            logfire_stream = self.call_params.logfire(
                client.messages.stream,
                "anthropic",
                response_chunk_type=AnthropicCallResponseChunk,
            )
            stream = logfire_stream(  # type: ignore
                messages=messages,
                **kwargs,
            )
            for chunk in stream:  # type:ignore
                yield AnthropicCallResponseChunk(
                    chunk=chunk,
                    tool_types=tool_types,
                    response_format=self.call_params.response_format,
                )
        else:
            with client.messages.stream(messages=messages, **kwargs) as stream:
                for chunk in stream:
                    yield AnthropicCallResponseChunk(
                        chunk=chunk,
                        tool_types=tool_types,
                        response_format=self.call_params.response_format,
                    )

    @retry
    async def stream_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> AsyncGenerator[AnthropicCallResponseChunk, None]:
        """Streams the response for an asynchronous call using this `AnthropicCall`.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            An `AnthropicCallResponseChunk` for each chunk of the response.
        """
        messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
        client = AsyncAnthropic(api_key=self.api_key, base_url=self.base_url)
        if self.call_params.wrapper_async is not None:
            client = self.call_params.wrapper_async(client)
        if self.call_params.logfire_async:  # pragma: no cover
            stream = self.call_params.logfire_async(
                client.messages.stream,
                "anthropic",
                response_chunk_type=AnthropicCallResponseChunk,
            )
            async for chunk in stream(
                messages=messages,
                **kwargs,
            ):
                yield AnthropicCallResponseChunk(
                    chunk=chunk,
                    tool_types=tool_types,
                    response_format=self.call_params.response_format,
                )
        else:
            async with client.messages.stream(messages=messages, **kwargs) as stream:
                async for chunk in stream:  # type: ignore
                    yield AnthropicCallResponseChunk(
                        chunk=chunk,
                        tool_types=tool_types,
                        response_format=self.call_params.response_format,
                    )

    ############################## PRIVATE METHODS ###################################

    def _setup_anthropic_kwargs(
        self,
        kwargs: dict[str, Any],
    ) -> tuple[
        list[MessageParam],
        dict[str, Any],
        Optional[list[Type[AnthropicTool]]],
    ]:
        """Overrides the `BaseCall._setup` for Anthropic specific setup."""
        kwargs, tool_types = self._setup(kwargs, AnthropicTool)
        messages = self.messages()
        system_message = ""
        if "system" in kwargs and kwargs["system"] is not None:
            system_message += f'{kwargs.pop("system")}'
        if messages[0]["role"] == "system":
            if system_message:
                system_message += "\n"
            system_message += messages.pop(0)["content"]
        if self.call_params.response_format == "json":
            if system_message:
                system_message += "\n\n"
            system_message += "Response format: JSON."
            messages.append(
                {
                    "role": "assistant",
                    "content": "Here is the JSON requested with only the fields "
                    "defined in the schema you provided:\n{",
                }
            )
            if "tools" in kwargs:
                tools = kwargs.pop("tools")
                messages[-1]["content"] = (
                    "For each JSON you output, output ONLY the fields defined by these "
                    "schemas. Include a `tool_name` field that EXACTLY MATCHES the "
                    "tool name found in the schema matching this tool:"
                    "\n{schemas}\n{json_msg}".format(
                        schemas="\n\n".join([str(tool) for tool in tools]),
                        json_msg=messages[-1]["content"],
                    )
                )
        if system_message:
            kwargs["system"] = system_message

        return messages, kwargs, tool_types

call(retries=1, **kwargs)

Makes a call to the model using this AnthropicCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
AnthropicCallResponse

A AnthropicCallResponse instance.
Source code in mirascope/anthropic/calls.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@retry
def call(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> AnthropicCallResponse:
    """Makes a call to the model using this `AnthropicCall` instance.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `AnthropicCallResponse` instance.
    """
    messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
    client = Anthropic(api_key=self.api_key, base_url=self.base_url)
    if self.call_params.wrapper is not None:
        client = self.call_params.wrapper(client)

    create = client.messages.create
    if tool_types:
        create = client.beta.tools.messages.create  # type: ignore
    if self.call_params.weave is not None:
        create = self.call_params.weave(create)  # pragma: no cover
    if self.call_params.logfire:
        create = self.call_params.logfire(
            create,
            "anthropic",
            response_type=AnthropicCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover
    start_time = datetime.datetime.now().timestamp() * 1000
    message = create(
        messages=messages,
        stream=False,
        **kwargs,
    )
    return AnthropicCallResponse(
        response=message,
        tool_types=tool_types,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        cost=anthropic_api_calculate_cost(message.usage, message.model),
        response_format=self.call_params.response_format,
    )

call_async(retries=1, **kwargs) async

Makes an asynchronous call to the model using this AnthropicCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Returns:

Type Description
AnthropicCallResponse

A AnthropicCallResponse instance.
Source code in mirascope/anthropic/calls.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@retry
async def call_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> AnthropicCallResponse:
    """Makes an asynchronous call to the model using this `AnthropicCall` instance.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Returns:
        A `AnthropicCallResponse` instance.
    """
    messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
    client = AsyncAnthropic(api_key=self.api_key, base_url=self.base_url)
    if self.call_params.wrapper_async is not None:
        client = self.call_params.wrapper_async(client)
    create = client.messages.create
    if tool_types:
        create = client.beta.tools.messages.create  # type: ignore
    if self.call_params.weave is not None:
        create = self.call_params.weave(create)  # pragma: no cover
    if self.call_params.logfire_async:
        create = self.call_params.logfire_async(
            create,
            "anthropic",
            response_type=AnthropicCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover
    start_time = datetime.datetime.now().timestamp() * 1000
    message = await create(
        messages=messages,
        stream=False,
        **kwargs,
    )
    return AnthropicCallResponse(
        response=message,
        tool_types=tool_types,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        cost=anthropic_api_calculate_cost(message.usage, message.model),
        response_format=self.call_params.response_format,
    )

messages()

Returns the template as a formatted list of messages.

Source code in mirascope/anthropic/calls.py 
45
46
47
48
49
def messages(self) -> list[MessageParam]:
    """Returns the template as a formatted list of messages."""
    return self._parse_messages(
        [MessageRole.SYSTEM, MessageRole.USER, MessageRole.ASSISTANT]
    )  # type: ignore

stream(retries=1, **kwargs)

Streams the response for a call using this AnthropicCall.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
AnthropicCallResponseChunk

An AnthropicCallResponseChunk for each chunk of the response.
Source code in mirascope/anthropic/calls.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@retry
def stream(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> Generator[AnthropicCallResponseChunk, None, None]:
    """Streams the response for a call using this `AnthropicCall`.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        An `AnthropicCallResponseChunk` for each chunk of the response.
    """
    messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
    client = Anthropic(api_key=self.api_key, base_url=self.base_url)
    if self.call_params.wrapper is not None:
        client = self.call_params.wrapper(client)
    if self.call_params.logfire:  # pragma: no cover
        logfire_stream = self.call_params.logfire(
            client.messages.stream,
            "anthropic",
            response_chunk_type=AnthropicCallResponseChunk,
        )
        stream = logfire_stream(  # type: ignore
            messages=messages,
            **kwargs,
        )
        for chunk in stream:  # type:ignore
            yield AnthropicCallResponseChunk(
                chunk=chunk,
                tool_types=tool_types,
                response_format=self.call_params.response_format,
            )
    else:
        with client.messages.stream(messages=messages, **kwargs) as stream:
            for chunk in stream:
                yield AnthropicCallResponseChunk(
                    chunk=chunk,
                    tool_types=tool_types,
                    response_format=self.call_params.response_format,
                )

stream_async(retries=1, **kwargs) async

Streams the response for an asynchronous call using this AnthropicCall.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
AsyncGenerator[AnthropicCallResponseChunk, None]

An AnthropicCallResponseChunk for each chunk of the response.
Source code in mirascope/anthropic/calls.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@retry
async def stream_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> AsyncGenerator[AnthropicCallResponseChunk, None]:
    """Streams the response for an asynchronous call using this `AnthropicCall`.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        An `AnthropicCallResponseChunk` for each chunk of the response.
    """
    messages, kwargs, tool_types = self._setup_anthropic_kwargs(kwargs)
    client = AsyncAnthropic(api_key=self.api_key, base_url=self.base_url)
    if self.call_params.wrapper_async is not None:
        client = self.call_params.wrapper_async(client)
    if self.call_params.logfire_async:  # pragma: no cover
        stream = self.call_params.logfire_async(
            client.messages.stream,
            "anthropic",
            response_chunk_type=AnthropicCallResponseChunk,
        )
        async for chunk in stream(
            messages=messages,
            **kwargs,
        ):
            yield AnthropicCallResponseChunk(
                chunk=chunk,
                tool_types=tool_types,
                response_format=self.call_params.response_format,
            )
    else:
        async with client.messages.stream(messages=messages, **kwargs) as stream:
            async for chunk in stream:  # type: ignore
                yield AnthropicCallResponseChunk(
                    chunk=chunk,
                    tool_types=tool_types,
                    response_format=self.call_params.response_format,
                )