direct_api_call
Call raw Blockscout API endpoints to retrieve blockchain data for specific chains, supporting pagination and query parameters for advanced data access.
Instructions
Call a raw Blockscout API endpoint for advanced or chain-specific data.
Do not include query strings in ``endpoint_path``; pass all query parameters via
``query_params`` to avoid double-encoding.
**SUPPORTS PAGINATION**: If response includes 'pagination' field,
use the provided next_call to get additional pages.
Returns:
ToolResponse[Any]: Must return ToolResponse[Any] (not ToolResponse[BaseModel])
because specialized handlers can return lists or other types that don't inherit
from BaseModel. The dispatcher system supports flexible data structures.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| chain_id | Yes | The ID of the blockchain | |
| endpoint_path | Yes | The Blockscout API path to call (e.g., '/api/v2/stats'); do not include query strings. | |
| query_params | No | Optional query parameters forwarded to the Blockscout API. | |
| cursor | No | The pagination cursor from a previous response to get the next page of results. |
Implementation Reference
- Main execution logic for the 'direct_api_call' tool: resolves Blockscout URL, makes API request with pagination support, dispatches to specialized handlers if applicable, and returns structured ToolResponse.@log_tool_invocation async def direct_api_call( chain_id: Annotated[str, Field(description="The ID of the blockchain")], endpoint_path: Annotated[ str, Field( description="The Blockscout API path to call (e.g., '/api/v2/stats'); do not include query strings.", ), ], ctx: Context, query_params: Annotated[ dict[str, Any] | None, Field(description="Optional query parameters forwarded to the Blockscout API."), ] = None, cursor: Annotated[ str | None, Field(description="The pagination cursor from a previous response to get the next page of results."), ] = None, ) -> ToolResponse[Any]: """Call a raw Blockscout API endpoint for advanced or chain-specific data. Do not include query strings in ``endpoint_path``; pass all query parameters via ``query_params`` to avoid double-encoding. **SUPPORTS PAGINATION**: If response includes 'pagination' field, use the provided next_call to get additional pages. Returns: ToolResponse[Any]: Must return ToolResponse[Any] (not ToolResponse[BaseModel]) because specialized handlers can return lists or other types that don't inherit from BaseModel. The dispatcher system supports flexible data structures. """ await report_and_log_progress( ctx, progress=0.0, total=2.0, message=f"Resolving Blockscout URL for chain {chain_id}...", ) base_url = await get_blockscout_base_url(chain_id) if endpoint_path != "/" and endpoint_path.endswith("/"): endpoint_path = endpoint_path.rstrip("/") if "?" in endpoint_path: raise ValueError("Do not include query parameters in endpoint_path. Use query_params instead.") params = dict(query_params) if query_params else {} apply_cursor_to_params(cursor, params) await report_and_log_progress( ctx, progress=1.0, total=2.0, message="Fetching data from Blockscout API...", ) response_json = await make_blockscout_request(base_url=base_url, api_path=endpoint_path, params=params) handler_response = await dispatcher.dispatch( endpoint_path=endpoint_path, query_params=query_params, response_json=response_json, chain_id=chain_id, base_url=base_url, ctx=ctx, ) if handler_response is not None: await report_and_log_progress( ctx, progress=2.0, total=2.0, message="Successfully fetched data.", ) return handler_response pagination = None next_page_params = response_json.get("next_page_params") if next_page_params: next_cursor = encode_cursor(next_page_params) next_call_params = { "chain_id": chain_id, "endpoint_path": endpoint_path, "cursor": next_cursor, } if query_params: next_call_params["query_params"] = query_params pagination = PaginationInfo(next_call=NextCallInfo(tool_name="direct_api_call", params=next_call_params)) await report_and_log_progress( ctx, progress=2.0, total=2.0, message="Successfully fetched data.", ) data = DirectApiData.model_validate(response_json) return build_tool_response(data=data, pagination=pagination)
- blockscout_mcp_server/server.py:213-216 (registration)Registers the 'direct_api_call' tool with the FastMCP server instance.mcp.tool( structured_output=False, annotations=create_tool_annotations("Direct Blockscout API Call"), )(direct_api_call)
- Dispatcher utility that routes API responses to registered specialized handlers based on endpoint path regex matching.async def dispatch( endpoint_path: str, **kwargs: Any, ) -> Any | None: """Find and execute the first matching handler for the given endpoint path. Args: endpoint_path: The API path that was requested. **kwargs: Additional context forwarded to the handler. Note: precedence follows registration order. Keep regex patterns disjoint or register the most specific handler first when overlap is unavoidable. Returns: Any | None: Must return Any (not ToolResponse[BaseModel]) because some handlers can return lists or other types that don't inherit from BaseModel. The handler is responsible for returning properly structured ToolResponse objects. """ for path_regex, handler in HANDLER_REGISTRY: match = path_regex.fullmatch(endpoint_path) if match: return await handler(match=match, **kwargs) return None
- Generic output schema used by direct_api_call and other tools for standardized responses including data, notes, instructions, and pagination.class ToolResponse(BaseModel, Generic[T]): """A standardized, structured response for all MCP tools, generic over the data payload type.""" data: T = Field(description="The main data payload of the tool's response.") data_description: list[str] | None = Field( None, description="A list of notes explaining the structure, fields, or conventions of the 'data' payload.", ) notes: list[str] | None = Field( None, description=( "A list of important contextual notes, such as warnings about data truncation or data quality issues." ), ) instructions: list[str] | None = Field( None, description="A list of suggested follow-up actions or instructions for the LLM to plan its next steps.", ) pagination: PaginationInfo | None = Field( None, description="Pagination information, present only if the 'data' is a single page of a larger result set.", )
- Pydantic model for validating and wrapping raw direct API response data, allowing extra fields.class DirectApiData(BaseModel): """Generic container for direct API responses.""" model_config = ConfigDict(extra="allow")