anp.fetchDoc
Fetch and parse ANP documents to extract content, type information, and followable links. This is the authorized method for accessing URLs within the ANP ecosystem.
Instructions
抓取并解析 ANP 文档，提取可跟进的链接。这是访问 ANP 生态系统中 URL 的唯一允许方法。返回文档内容、类型信息和发现的链接。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| url | Yes | 要抓取的 ANP 文档的 URL |
Implementation Reference
- mcp2anp/core/handlers.py:22-85 (handler)Core handler function for executing the 'anp.fetchDoc' tool. It validates input, uses ANPCrawler to fetch and parse ANP documents, handles JSON content, and returns structured results with error handling.async def handle_fetch_doc(self, arguments: dict[str, Any]) -> dict[str, Any]: """处理 fetchDoc 工具调用。""" try: # 验证参数 request = models.FetchDocRequest(**arguments) logger.info("Fetching document", url=request.url) # 检查 ANPCrawler 是否已初始化 if self.anp_crawler is None: return { "ok": False, "error": { "code": "ANP_NOT_INITIALIZED", "message": "ANPCrawler not initialized. Please check DID credentials.", }, } # 使用 ANPCrawler 获取文档 content_result, interfaces = await self.anp_crawler.fetch_text(request.url) # 构建链接列表 links = [] for interface in interfaces: func_info = interface.get("function", {}) links.append({ "rel": "interface", "url": request.url, # ANPCrawler 已经处理了 URL 解析 "title": func_info.get("name", ""), "description": func_info.get("description", ""), }) result = { "ok": True, "contentType": content_result.get("content_type", "application/json"), "text": content_result.get("content", ""), # "interfaces": links, } # 如果内容是 JSON，尝试解析 try: if content_result.get("content"): json_data = json.loads(content_result["content"]) result["json"] = json_data # 删除 text字段，只保留一个 result.pop("text") except json.JSONDecodeError: pass # 不是 JSON 内容，跳过 logger.info("Document fetched successfully", url=request.url, links_count=len(links)) return result except Exception as e: logger.error("Failed to fetch document", url=arguments.get("url"), error=str(e)) return { "ok": False, "error": { "code": "ANP_FETCH_ERROR", "message": str(e), }, }
- mcp2anp/server.py:38-94 (registration)MCP server registration of the 'anp.fetchDoc' tool via the list_tools() handler, including name, description, and input schema.@server.list_tools() async def list_tools() -> list[Tool]: """返回可用工具列表。""" return [ Tool( name="anp.fetchDoc", description=( "抓取并解析 ANP 文档，提取可跟进的链接。这是访问 ANP 生态系统中 URL 的唯一允许方法。返回文档内容、类型信息和发现的链接。" "ANP生态系统目前提供包括酒店、景点、地图规划、天气、快递等查询预订能力。" "ANP生态的入口URL：https://agent-navigation.com/ad.json" "你可以从这个URL开始，逐步跟进并调用工具，访问ANP生态的资源和能力。" ), inputSchema={ "type": "object", "properties": { "url": { "type": "string", "description": "要抓取的 ANP 文档的 URL", "format": "uri", }, }, "required": ["url"], }, ), Tool( name="anp.invokeOpenRPC", description=( "使用 JSON-RPC 2.0 协议调用 OpenRPC 端点上的方法。" "此工具处理与暴露 OpenRPC 接口的 ANP 智能体的结构化交互。" ), inputSchema={ "type": "object", "properties": { "endpoint": { "type": "string", "description": "OpenRPC 端点 URL", "format": "uri", }, "method": { "type": "string", "description": "要调用的 RPC 方法名称", }, "params": { "description": "传递给方法的参数", }, "id": { "type": "string", "description": "用于跟踪的可选请求 ID", }, }, "required": ["endpoint", "method"], }, ), ]
- mcp2anp/server.py:104-108 (handler)Dispatch logic in MCP call_tool() handler that routes 'anp.fetchDoc' calls to the core ANPHandler.handle_fetch_doc.if name == "anp.fetchDoc": result = await anp_handler.handle_fetch_doc(arguments) elif name == "anp.invokeOpenRPC": result = await anp_handler.handle_invoke_openrpc(arguments) else:
- mcp2anp/utils/models.py:31-35 (schema)Internal Pydantic schema (FetchDocRequest) used for input validation within the tool handler.class FetchDocRequest(ConfigMixin, BaseModel): """fetchDoc 工具请求模型。""" url: str = Field(..., description="要获取的 URL")
- mcp2anp/server_http.py:94-98 (schema)Pydantic input schema for anp.fetchDoc in the HTTP server exposure.class FetchDocIn(BaseModel): """Input model for anp.fetchDoc.""" url: HttpUrl = Field(..., description="ANP doc url to fetch")