feat: add ANP tool configuration and remove unused HotelBook import

Author: [email protected]

app/agent/manus.py

@@ -10,7 +10,6 @@
 from app.tool.python_execute import PythonExecute
 from app.tool.anp_tool import ANPTool
 from app.tool.web_search import WebSearch
-from app.tool.hotel_book import HotelBook
 
 
 class Manus(ToolCallAgent):
@@ -40,7 +39,6 @@ class Manus(ToolCallAgent):
             WebSearch(),
             BrowserUseTool(),
             FileSaver(),
-            # HotelBook(),
             ANPTool(),
             Terminate(),
         )

app/config.py

@@ -61,6 +61,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(
@@ -69,6 +78,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
@@ -166,6 +178,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,
@@ -176,6 +200,7 @@ def _load_initial_config(self):
             },
             "browser_config": browser_settings,
             "search_config": search_settings,
+            "anp_config": anp_settings,
         }
 
         self._config = AppConfig(**config_dict)
@@ -192,5 +217,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/tool/anp_tool.py

@@ -8,6 +8,8 @@
 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"
@@ -47,47 +49,55 @@ class ANPTool(BaseTool):
         },
         "required": ["url"],
     }
-    
+
     # Declare auth_client field
     auth_client: Optional[DIDWbaAuthHeader] = None
 
     def __init__(self, **data):
         super().__init__(**data)
 
-        # Get current script directory
-        current_dir = Path(__file__).parent
-        # Get project root directory
-        base_dir = current_dir.parent.parent
-        
-        # Initialize DID authentication client
-        did_path = str(base_dir / "config/did_test_public_doc/did.json")
-        key_path = str(base_dir / "config/did_test_public_doc/key-1_private.pem")
-        
-        logger.info(f"ANPTool initialized - DID path: {did_path}, private key path: {key_path}")
-        
+        # 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
+            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
+        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
         """
@@ -96,21 +106,20 @@ async def execute(
             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
@@ -119,40 +128,46 @@ async def execute(
                 "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 (
+                        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))
+                        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}")
+                            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
-                }
+                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
@@ -161,42 +176,55 @@ async def _process_response(self, response, url):
                 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()
-        
+        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:
+        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")
+                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:
+        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}
+                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")
+                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}
-        
+            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