Spaces:
Running
Running
File size: 13,237 Bytes
8c5bdf2 151d030 8c5bdf2 eafc773 8c5bdf2 eafc773 8c5bdf2 151d030 8c5bdf2 eafc773 8c5bdf2 eafc773 151d030 8c5bdf2 eafc773 8c5bdf2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 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 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 | """OpenAPI component implementations: Tool, Resource, and ResourceTemplate classes."""
import json
import re
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import httpx
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
# Import from our new utilities
from fastmcp.experimental.utilities.openapi import HTTPRoute
from fastmcp.experimental.utilities.openapi.director import RequestDirector
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.dependencies import get_http_headers
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
from fastmcp.server import Context
logger = get_logger(__name__)
class OpenAPITool(Tool):
"""Tool implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
name: str,
description: str,
parameters: dict[str, Any],
output_schema: dict[str, Any] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
):
super().__init__(
name=name,
description=description,
parameters=parameters,
output_schema=output_schema,
tags=tags or set(),
annotations=annotations,
serializer=serializer,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPITool(name={self.name!r}, method={self._route.method}, path={self._route.path})"
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Execute the HTTP request using RequestDirector for simplified parameter handling."""
try:
# Get base URL from client
base_url = (
str(self._client.base_url)
if hasattr(self._client, "base_url") and self._client.base_url
else "http://localhost"
)
# Get Headers from client
cli_headers = (
self._client.headers
if hasattr(self._client, "headers") and self._client.headers
else {}
)
# Build the request using RequestDirector
request = self._director.build(self._route, arguments, base_url)
# First add server headers (lowest precedence)
if cli_headers:
# Merge with existing headers, _client headers as base
if request.headers:
# Start with request headers, then add client headers
for key, value in cli_headers.items():
if key not in request.headers:
request.headers[key] = value
else:
# Create new headers from cli_headers
for key, value in cli_headers.items():
request.headers[key] = value
# Then add MCP client transport headers (highest precedence)
mcp_headers = get_http_headers()
if mcp_headers:
# Merge with existing headers, MCP headers take precedence over all
if request.headers:
request.headers.update(mcp_headers)
else:
# Create new headers from mcp_headers
for key, value in mcp_headers.items():
request.headers[key] = value
# print logger
logger.debug(f"run - sending request; headers: {request.headers}")
# Execute the request
# Note: httpx.AsyncClient.send() doesn't accept timeout parameter
# The timeout should be configured on the client itself
response = await self._client.send(request)
# Raise for 4xx/5xx responses
response.raise_for_status()
# Try to parse as JSON first
try:
result = response.json()
# Handle structured content based on output schema, if any
structured_output = None
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
# Schema says wrap - always wrap in result key
structured_output = {"result": result}
else:
structured_output = result
# If no output schema, use fallback logic for backward compatibility
elif not isinstance(result, dict):
structured_output = {"result": result}
else:
structured_output = result
return ToolResult(structured_content=structured_output)
except json.JSONDecodeError:
return ToolResult(content=response.text)
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message)
except httpx.RequestError as e:
# Handle request errors (connection, timeout, etc.)
raise ValueError(f"Request error: {str(e)}")
class OpenAPIResource(Resource):
"""Resource implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri: str,
name: str,
description: str,
mime_type: str = "application/json",
tags: set[str] = set(),
timeout: float | None = None,
):
super().__init__(
uri=AnyUrl(uri), # Convert string to AnyUrl
name=name,
description=description,
mime_type=mime_type,
tags=tags,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> str | bytes:
"""Fetch the resource data by making an HTTP request."""
try:
# Extract path parameters from the URI if present
path = self._route.path
resource_uri = str(self.uri)
# If this is a templated resource, extract path parameters from the URI
if "{" in path and "}" in path:
# Extract the resource ID from the URI (the last part after the last slash)
parts = resource_uri.split("/")
if len(parts) > 1:
# Find all path parameters in the route path
path_params = {}
# Find the path parameter names from the route path
param_matches = re.findall(r"\{([^}]+)\}", path)
if param_matches:
# Reverse sorting from creation order (traversal is backwards)
param_matches.sort(reverse=True)
# Number of sent parameters is number of parts -1 (assuming first part is resource identifier)
expected_param_count = len(parts) - 1
# Map parameters from the end of the URI to the parameters in the path
# Last parameter in URI (parts[-1]) maps to last parameter in path, and so on
for i, param_name in enumerate(param_matches):
# Ensure we don't use resource identifier as parameter
if i < expected_param_count:
# Get values from the end of parts
param_value = parts[-1 - i]
path_params[param_name] = param_value
# Replace path parameters with their values
for param_name, param_value in path_params.items():
path = path.replace(f"{{{param_name}}}", str(param_value))
# Filter any query parameters - get query parameters and filter out None/empty values
query_params = {}
for param in self._route.parameters:
if param.location == "query" and hasattr(self, f"_{param.name}"):
value = getattr(self, f"_{param.name}")
if value is not None and value != "":
query_params[param.name] = value
# Prepare headers with correct precedence: server < client transport
headers = {}
# Start with server headers (lowest precedence)
cli_headers = (
self._client.headers
if hasattr(self._client, "headers") and self._client.headers
else {}
)
headers.update(cli_headers)
# Add MCP client transport headers (highest precedence)
mcp_headers = get_http_headers()
headers.update(mcp_headers)
response = await self._client.request(
method=self._route.method,
url=path,
params=query_params,
headers=headers,
timeout=self._timeout,
)
# Raise for 4xx/5xx responses
response.raise_for_status()
# Determine content type and return appropriate format
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
result = response.json()
return json.dumps(result)
elif any(ct in content_type for ct in ["text/", "application/xml"]):
return response.text
else:
return response.content
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message)
except httpx.RequestError as e:
# Handle request errors (connection, timeout, etc.)
raise ValueError(f"Request error: {str(e)}")
class OpenAPIResourceTemplate(ResourceTemplate):
"""Resource template implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri_template: str,
name: str,
description: str,
parameters: dict[str, Any],
tags: set[str] = set(),
timeout: float | None = None,
):
super().__init__(
uri_template=uri_template,
name=name,
description=description,
parameters=parameters,
tags=tags,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResourceTemplate(name={self.name!r}, uri_template={self.uri_template!r}, path={self._route.path})"
async def create_resource(
self,
uri: str,
params: dict[str, Any],
context: "Context | None" = None,
) -> Resource:
"""Create a resource with the given parameters."""
# Generate a URI for this resource instance
uri_parts = []
for key, value in params.items():
uri_parts.append(f"{key}={value}")
# Create and return a resource
return OpenAPIResource(
client=self._client,
route=self._route,
director=self._director,
uri=uri,
name=f"{self.name}-{'-'.join(uri_parts)}",
description=self.description or f"Resource for {self._route.path}",
mime_type="application/json",
tags=set(self._route.tags or []),
timeout=self._timeout,
)
# Export public symbols
__all__ = [
"OpenAPITool",
"OpenAPIResource",
"OpenAPIResourceTemplate",
]
|