Jeremiah Lowin commited on
Commit
6e546c4
·
1 Parent(s): 8dc263e

Add Audio helper

Browse files
src/fastmcp/tools/tool.py CHANGED
@@ -17,6 +17,7 @@ from fastmcp.utilities.components import FastMCPComponent
17
  from fastmcp.utilities.json_schema import compress_schema
18
  from fastmcp.utilities.logging import get_logger
19
  from fastmcp.utilities.types import (
 
20
  Image,
21
  MCPContent,
22
  find_kwarg_by_type,
@@ -273,6 +274,9 @@ def _convert_to_content(
273
  if isinstance(result, Image):
274
  return [result.to_image_content()]
275
 
 
 
 
276
  if isinstance(result, list | tuple) and not _process_as_single_item:
277
  # if the result is a list, then it could either be a list of MCP types,
278
  # or a "regular" list that the tool is returning, or a mix of both.
@@ -284,7 +288,7 @@ def _convert_to_content(
284
  other_content = []
285
 
286
  for item in result:
287
- if isinstance(item, MCPContent | Image):
288
  mcp_types.append(_convert_to_content(item)[0])
289
  else:
290
  other_content.append(item)
 
17
  from fastmcp.utilities.json_schema import compress_schema
18
  from fastmcp.utilities.logging import get_logger
19
  from fastmcp.utilities.types import (
20
+ Audio,
21
  Image,
22
  MCPContent,
23
  find_kwarg_by_type,
 
274
  if isinstance(result, Image):
275
  return [result.to_image_content()]
276
 
277
+ elif isinstance(result, Audio):
278
+ return [result.to_audio_content()]
279
+
280
  if isinstance(result, list | tuple) and not _process_as_single_item:
281
  # if the result is a list, then it could either be a list of MCP types,
282
  # or a "regular" list that the tool is returning, or a mix of both.
 
288
  other_content = []
289
 
290
  for item in result:
291
+ if isinstance(item, MCPContent | Image | Audio):
292
  mcp_types.append(_convert_to_content(item)[0])
293
  else:
294
  other_content.append(item)
src/fastmcp/utilities/types.py CHANGED
@@ -8,7 +8,13 @@ from pathlib import Path
8
  from types import UnionType
9
  from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
10
 
11
- from mcp.types import AudioContent, EmbeddedResource, ImageContent, TextContent
 
 
 
 
 
 
12
  from pydantic import BaseModel, ConfigDict, TypeAdapter
13
 
14
  T = TypeVar("T")
@@ -90,6 +96,7 @@ class Image:
90
  path: str | Path | None = None,
91
  data: bytes | None = None,
92
  format: str | None = None,
 
93
  ):
94
  if path is None and data is None:
95
  raise ValueError("Either path or data must be provided")
@@ -100,6 +107,7 @@ class Image:
100
  self.data = data
101
  self._format = format
102
  self._mime_type = self._get_mime_type()
 
103
 
104
  def _get_mime_type(self) -> str:
105
  """Get MIME type from format or guess from file extension."""
@@ -117,7 +125,11 @@ class Image:
117
  }.get(suffix, "application/octet-stream")
118
  return "image/png" # default for raw binary data
119
 
120
- def to_image_content(self) -> ImageContent:
 
 
 
 
121
  """Convert to MCP ImageContent."""
122
  if self.path:
123
  with open(self.path, "rb") as f:
@@ -127,4 +139,67 @@ class Image:
127
  else:
128
  raise ValueError("No image data available")
129
 
130
- return ImageContent(type="image", data=data, mimeType=self._mime_type)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8
  from types import UnionType
9
  from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
10
 
11
+ from mcp.types import (
12
+ Annotations,
13
+ AudioContent,
14
+ EmbeddedResource,
15
+ ImageContent,
16
+ TextContent,
17
+ )
18
  from pydantic import BaseModel, ConfigDict, TypeAdapter
19
 
20
  T = TypeVar("T")
 
96
  path: str | Path | None = None,
97
  data: bytes | None = None,
98
  format: str | None = None,
99
+ annotations: Annotations | None = None,
100
  ):
101
  if path is None and data is None:
102
  raise ValueError("Either path or data must be provided")
 
107
  self.data = data
108
  self._format = format
109
  self._mime_type = self._get_mime_type()
110
+ self.annotations = annotations
111
 
112
  def _get_mime_type(self) -> str:
113
  """Get MIME type from format or guess from file extension."""
 
125
  }.get(suffix, "application/octet-stream")
126
  return "image/png" # default for raw binary data
127
 
128
+ def to_image_content(
129
+ self,
130
+ mime_type: str | None = None,
131
+ annotations: Annotations | None = None,
132
+ ) -> ImageContent:
133
  """Convert to MCP ImageContent."""
134
  if self.path:
135
  with open(self.path, "rb") as f:
 
139
  else:
140
  raise ValueError("No image data available")
141
 
142
+ return ImageContent(
143
+ type="image",
144
+ data=data,
145
+ mimeType=mime_type or self._mime_type,
146
+ annotations=annotations or self.annotations,
147
+ )
148
+
149
+
150
+ class Audio:
151
+ """Helper class for returning audio from tools."""
152
+
153
+ def __init__(
154
+ self,
155
+ path: str | Path | None = None,
156
+ data: bytes | None = None,
157
+ format: str | None = None,
158
+ annotations: Annotations | None = None,
159
+ ):
160
+ if path is None and data is None:
161
+ raise ValueError("Either path or data must be provided")
162
+ if path is not None and data is not None:
163
+ raise ValueError("Only one of path or data can be provided")
164
+
165
+ self.path = Path(path) if path else None
166
+ self.data = data
167
+ self._format = format
168
+ self._mime_type = self._get_mime_type()
169
+ self.annotations = annotations
170
+
171
+ def _get_mime_type(self) -> str:
172
+ """Get MIME type from format or guess from file extension."""
173
+ if self._format:
174
+ return f"audio/{self._format.lower()}"
175
+
176
+ if self.path:
177
+ suffix = self.path.suffix.lower()
178
+ return {
179
+ ".wav": "audio/wav",
180
+ ".mp3": "audio/mpeg",
181
+ ".ogg": "audio/ogg",
182
+ ".m4a": "audio/mp4",
183
+ ".flac": "audio/flac",
184
+ }.get(suffix, "application/octet-stream")
185
+ return "audio/wav" # default for raw binary data
186
+
187
+ def to_audio_content(
188
+ self,
189
+ mime_type: str | None = None,
190
+ annotations: Annotations | None = None,
191
+ ) -> AudioContent:
192
+ if self.path:
193
+ with open(self.path, "rb") as f:
194
+ data = base64.b64encode(f.read()).decode()
195
+ elif self.data is not None:
196
+ data = base64.b64encode(self.data).decode()
197
+ else:
198
+ raise ValueError("No audio data available")
199
+
200
+ return AudioContent(
201
+ type="audio",
202
+ data=data,
203
+ mimeType=mime_type or self._mime_type,
204
+ annotations=annotations or self.annotations,
205
+ )