feat: introduce anp tool

.gitignore

@@ -194,3 +194,5 @@ cython_debug/
 
 # Built Visual Studio Code Extensions
 *.vsix
+
+config.toml

app/agent/manus.py

@@ -7,6 +7,7 @@
 from app.logger import logger
 from app.prompt.manus import NEXT_STEP_PROMPT, SYSTEM_PROMPT
 from app.tool import Terminate, ToolCollection
+from app.tool.anp_tool import ANPTool
 from app.tool.browser_use_tool import BrowserUseTool
 from app.tool.file_saver import FileSaver
 from app.tool.python_execute import PythonExecute
@@ -35,7 +36,11 @@ class Manus(ToolCallAgent):
     # Add general-purpose tools to the tool collection
     available_tools: ToolCollection = Field(
         default_factory=lambda: ToolCollection(
-            PythonExecute(), BrowserUseTool(), FileSaver(), Terminate()
+            PythonExecute(),
+            BrowserUseTool(),
+            FileSaver(),
+            ANPTool(),
+            Terminate(),
         )
     )
 

app/config.py

@@ -64,6 +64,15 @@ class BrowserSettings(BaseModel):
     )
 
 
+class ANPSettings(BaseModel):
+    did_document_path: Optional[str] = Field(
+        None, description="Path to DID document file"
+    )
+    private_key_path: Optional[str] = Field(
+        None, description="Path to private key file"
+    )
+
+
 class AppConfig(BaseModel):
     llm: Dict[str, LLMSettings]
     browser_config: Optional[BrowserSettings] = Field(
@@ -72,6 +81,9 @@ class AppConfig(BaseModel):
     search_config: Optional[SearchSettings] = Field(
         None, description="Search configuration"
     )
+    anp_config: Optional[ANPSettings] = Field(
+        None, description="ANP tool configuration"
+    )
 
     class Config:
         arbitrary_types_allowed = True
@@ -169,6 +181,18 @@ def _load_initial_config(self):
         if search_config:
             search_settings = SearchSettings(**search_config)
 
+        # handle ANP config
+        anp_config = raw_config.get("anp", {})
+        anp_settings = None
+        if anp_config:
+            anp_settings = ANPSettings(
+                **{
+                    k: v
+                    for k, v in anp_config.items()
+                    if k in ANPSettings.__annotations__ and v is not None
+                }
+            )
+
         config_dict = {
             "llm": {
                 "default": default_settings,
@@ -179,6 +203,7 @@ def _load_initial_config(self):
             },
             "browser_config": browser_settings,
             "search_config": search_settings,
+            "anp_config": anp_settings,
         }
 
         self._config = AppConfig(**config_dict)
@@ -195,5 +220,9 @@ def browser_config(self) -> Optional[BrowserSettings]:
     def search_config(self) -> Optional[SearchSettings]:
         return self._config.search_config
 
+    @property
+    def anp_config(self) -> Optional[ANPSettings]:
+        return self._config.anp_config
+
 
 config = Config()

app/prompt/manus.py

@@ -1,6 +1,6 @@
 SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user. You have various tools at your disposal that you can call upon to efficiently complete complex requests. Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."
 
-NEXT_STEP_PROMPT = """You can interact with the computer using PythonExecute, save important content and information files through FileSaver, open browsers with BrowserUseTool, and retrieve information using GoogleSearch.
+NEXT_STEP_PROMPT = """You can interact with the computer using PythonExecute, save important content and information files through FileSaver, open browsers with BrowserUseTool, and retrieve information using GoogleSearch, and interact with other agents using ANPTool.
 
 PythonExecute: Execute Python code to interact with the computer system, data processing, automation tasks, etc.
 
@@ -10,6 +10,8 @@
 
 Terminate: End the current interaction when the task is complete or when you need additional information from the user. Use this tool to signal that you've finished addressing the user's request or need clarification before proceeding further.
 
+ANPTool: Interact with other agents using Agent Network Protocol (ANP).ANP provides services such as hotel and scenic spot ticket query and booking.
+
 Based on user needs, proactively select the most appropriate tool or combination of tools. For complex tasks, you can break down the problem and use different tools step by step to solve it. After using each tool, clearly explain the execution results and suggest the next steps.
 
 Always maintain a helpful, informative tone throughout the interaction. If you encounter any limitations or need more details, clearly communicate this to the user before terminating.

app/tool/anp_tool.py

@@ -0,0 +1,230 @@
+import asyncio
+import json
+import yaml
+import aiohttp
+from pathlib import Path
+from typing import Dict, Any, Optional
+
+from app.logger import logger
+from app.tool.base import BaseTool
+from agent_connect.authentication import DIDWbaAuthHeader
+from app.config import config, PROJECT_ROOT
+
+
+class ANPTool(BaseTool):
+    name: str = "anp_tool"
+    description: str = """Use Agent Network Protocol (ANP) to interact with other agents.
+1. For the first use, please enter the URL: https://agent-search.ai/ad.json, which is an agent search service. You can use the interfaces inside to query agents that can provide hotels, tickets, and attractions.
+2. After receiving the agent's description document, you can crawl data based on the data link URL in the agent's description document.
+3. During the process, you can call the API to complete the service until you think the task is completed.
+4. Note, any URL obtained using ANPTool must be called using ANPTool, do not call it directly yourself.
+"""
+    parameters: dict = {
+        "type": "object",
+        "properties": {
+            "url": {
+                "type": "string",
+                "description": "(required) URL of the agent description file or API endpoint",
+            },
+            "method": {
+                "type": "string",
+                "description": "(optional) HTTP method, such as GET, POST, PUT, etc., default is GET",
+                "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"],
+                "default": "GET",
+            },
+            "headers": {
+                "type": "object",
+                "description": "(optional) HTTP request headers",
+                "default": {},
+            },
+            "params": {
+                "type": "object",
+                "description": "(optional) URL query parameters",
+                "default": {},
+            },
+            "body": {
+                "type": "object",
+                "description": "(optional) Request body for POST/PUT requests",
+            },
+        },
+        "required": ["url"],
+    }
+
+    # Declare auth_client field
+    auth_client: Optional[DIDWbaAuthHeader] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+
+        # Get default paths relative to project root
+        default_did_path = str(PROJECT_ROOT / "config/did_test_public_doc/did.json")
+        default_key_path = str(
+            PROJECT_ROOT / "config/did_test_public_doc/key-1_private.pem"
+        )
+
+        # Use paths from configuration if available, otherwise use defaults
+        did_path = default_did_path
+        key_path = default_key_path
+
+        if config.anp_config:
+            if config.anp_config.did_document_path:
+                did_path = config.anp_config.did_document_path
+            if config.anp_config.private_key_path:
+                key_path = config.anp_config.private_key_path
+
+        logger.info(
+            f"ANPTool initialized - DID path: {did_path}, private key path: {key_path}"
+        )
+
+        self.auth_client = DIDWbaAuthHeader(
+            did_document_path=did_path, private_key_path=key_path
+        )
+
+    async def execute(
+        self,
+        url: str,
+        method: str = "GET",
+        headers: Dict[str, str] = None,
+        params: Dict[str, Any] = None,
+        body: Dict[str, Any] = None,
+    ) -> Dict[str, Any]:
+        """
+        Execute HTTP request to interact with other agents
+
+        Args:
+            url (str): URL of the agent description file or API endpoint
+            method (str, optional): HTTP method, default is "GET"
+            headers (Dict[str, str], optional): HTTP request headers
+            params (Dict[str, Any], optional): URL query parameters
+            body (Dict[str, Any], optional): Request body for POST/PUT requests
+
+        Returns:
+            Dict[str, Any]: Response content
+        """
+
+        if headers is None:
+            headers = {}
+        if params is None:
+            params = {}
+
+        logger.info(f"ANP request: {method} {url}")
+
+        # Add basic request headers
+        if "Content-Type" not in headers and method in ["POST", "PUT", "PATCH"]:
+            headers["Content-Type"] = "application/json"
+
+        # Add DID authentication
+        if self.auth_client:
+            try:
+                auth_headers = self.auth_client.get_auth_header(url)
+                headers.update(auth_headers)
+            except Exception as e:
+                logger.error(f"Failed to get authentication header: {str(e)}")
+
+        async with aiohttp.ClientSession() as session:
+            # Prepare request parameters
+            request_kwargs = {
+                "url": url,
+                "headers": headers,
+                "params": params,
+            }
+
+            # If there is a request body and the method supports it, add the request body
+            if body is not None and method in ["POST", "PUT", "PATCH"]:
+                request_kwargs["json"] = body
+
+            # Execute request
+            http_method = getattr(session, method.lower())
+
+            try:
+                async with http_method(**request_kwargs) as response:
+                    logger.info(f"ANP response: status code {response.status}")
+
+                    # Check response status
+                    if (
+                        response.status == 401
+                        and "Authorization" in headers
+                        and self.auth_client
+                    ):
+                        logger.warning(
+                            "Authentication failed (401), trying to get authentication again"
+                        )
+                        # If authentication fails and a token was used, clear the token and retry
+                        self.auth_client.clear_token(url)
+                        # Get authentication header again
+                        headers.update(
+                            self.auth_client.get_auth_header(url, force_new=True)
+                        )
+                        # Execute request again
+                        request_kwargs["headers"] = headers
+                        async with http_method(**request_kwargs) as retry_response:
+                            logger.info(
+                                f"ANP retry response: status code {retry_response.status}"
+                            )
+                            return await self._process_response(retry_response, url)
+
+                    return await self._process_response(response, url)
+            except aiohttp.ClientError as e:
+                logger.error(f"HTTP request failed: {str(e)}")
+                return {"error": f"HTTP request failed: {str(e)}", "status_code": 500}
+
+    async def _process_response(self, response, url):
+        """Process HTTP response"""
+        # If authentication is successful, update the token
+        if response.status == 200 and self.auth_client:
+            try:
+                self.auth_client.update_token(url, dict(response.headers))
+            except Exception as e:
+                logger.error(f"Failed to update token: {str(e)}")
+
+        # Get response content type
+        content_type = response.headers.get("Content-Type", "").lower()
+
+        # Get response text
+        text = await response.text()
+
+        # Process response based on content type
+        if "application/json" in content_type:
+            # Process JSON response
+            try:
+                result = json.loads(text)
+                logger.info("Successfully parsed JSON response")
+            except json.JSONDecodeError:
+                logger.warning(
+                    "Content-Type declared as JSON but parsing failed, returning raw text"
+                )
+                result = {"text": text, "format": "text", "content_type": content_type}
+        elif "application/yaml" in content_type or "application/x-yaml" in content_type:
+            # Process YAML response
+            try:
+                result = yaml.safe_load(text)
+                logger.info("Successfully parsed YAML response")
+                result = {
+                    "data": result,
+                    "format": "yaml",
+                    "content_type": content_type,
+                }
+            except yaml.YAMLError:
+                logger.warning(
+                    "Content-Type declared as YAML but parsing failed, returning raw text"
+                )
+                result = {"text": text, "format": "text", "content_type": content_type}
+        else:
+            # Default to text
+            result = {"text": text, "format": "text", "content_type": content_type}
+
+        # Add status code to result
+        if isinstance(result, dict):
+            result["status_code"] = response.status
+        else:
+            result = {
+                "data": result,
+                "status_code": response.status,
+                "format": "unknown",
+                "content_type": content_type,
+            }
+
+        # Add URL to result for tracking
+        result["url"] = str(url)
+
+        return result

config/config.example.toml

@@ -65,3 +65,10 @@ temperature = 0.0                           # Controls randomness for vision mod
 # [search]
 # Search engine for agent to use. Default is "Google", can be set to "Baidu" or "DuckDuckGo".
 #engine = "Google"
+
+# Optional configuration for ANP tool
+# [anp]
+# Path to the DID document file
+# did_document_path = "/path/to/your/did.json"
+# Path to the private key file
+# private_key_path = "/path/to/your/private-key.pem"

config/did_test_public_doc/did.json

@@ -0,0 +1,25 @@
+{
+  "@context": [
+    "https://www.w3.org/ns/did/v1",
+    "https://w3id.org/security/suites/jws-2020/v1",
+    "https://w3id.org/security/suites/secp256k1-2019/v1"
+  ],
+  "id": "did:wba:agent-did.com:test:public",
+  "verificationMethod": [
+    {
+      "id": "did:wba:agent-did.com:test:public#key-1",
+      "type": "EcdsaSecp256k1VerificationKey2019",
+      "controller": "did:wba:agent-did.com:test:public",
+      "publicKeyJwk": {
+        "kty": "EC",
+        "crv": "secp256k1",
+        "x": "kNcfVufb4qxUpTC7kT6V56zSFEXbyo3nTDUFxqaZbKs",
+        "y": "upHpNoIm8h6cDRZqNWjb4VXbaniq2zz43yQoiR8Zfqs",
+        "kid": "4QOVubQtyJL_fzKreUfKDTOjrYAFRsq6XZ4Itqn2jLg"
+      }
+    }
+  ],
+  "authentication": [
+    "did:wba:agent-did.com:test:public#key-1"
+  ]
+}

config/did_test_public_doc/key-1_private.pem

@@ -0,0 +1,5 @@
+-----BEGIN PRIVATE KEY-----
+MIGEAgEAMBAGByqGSM49AgEGBSuBBAAKBG0wawIBAQQgvt5dMwF4pTOFv1BBD1IV
+ezSrGc0X/aDMesV57FeDvpKhRANCAASQ1x9W59virFSlMLuRPpXnrNIURdvKjedM
+NQXGpplsq7qR6TaCJvIenA0WajVo2+FV22p4qts8+N8kKIkfGX6r
+-----END PRIVATE KEY-----

config/did_test_public_doc/private_keys.json

@@ -0,0 +1,10 @@
+{
+  "did": "did:wba:agent-did.com:test:public",
+  "created_at": "2025-03-04T22:14:16.233463",
+  "keys": {
+    "key-1": {
+      "path": "key-1_private.pem",
+      "type": "EcdsaSecp256k1"
+    }
+  }
+}

requirements.txt

@@ -23,3 +23,4 @@ aiofiles~=24.1.0
 pydantic_core~=2.27.2
 colorama~=0.4.6
 playwright~=1.50.0
+agent-connect~=0.3.5