Agent Knowledge MCP

Instructions

Implementation Reference

• src/admin/admin_server.py:89-163 (handler)

  The primary handler implementation for the 'update_config' tool. It supports updating specific config sections/keys or replacing the entire config.json file. Includes validation for required sections, automatic reinitialization of security and Elasticsearch components, and comprehensive error handling with user-friendly messages.

  @app.tool(
      description="Update configuration with section-specific changes or full configuration replacement",
      tags={"admin", "config", "update", "modify", "settings"}
  )
  async def update_config(
      config_section: Annotated[Optional[str], Field(description="The top-level section of the config to update (e.g., 'security')")] = None,
      config_key: Annotated[Optional[str], Field(description="The key within the section to update (e.g., 'allowed_base_directory')")] = None,
      config_value: Annotated[Optional[str], Field(description="The new value for the specified key")] = None,
      full_config: Annotated[Optional[Dict[str, Any]], Field(description="Full configuration object to save. Replaces the entire config")] = None
  ) -> str:
      """Update configuration with comprehensive validation and automatic component reinitialization."""
      try:
          config_path = Path(__file__).parent.parent / "config.json"
  
          if full_config:
              # Full configuration replacement
              new_config = full_config
  
              # Validate new config structure
              required_sections = ["elasticsearch", "security", "document_validation", "server"]
              for section in required_sections:
                  if section not in new_config:
                      return f"❌ Error: Missing required config section '{section}'\n💡 Required sections: {', '.join(required_sections)}"
  
              # Write new config
              with open(config_path, 'w', encoding='utf-8') as f:
                  json.dump(new_config, f, indent=2, ensure_ascii=False)
  
              message = "✅ Full configuration updated successfully!"
  
          elif config_section and config_key is not None:
              # Section-specific key update
              config = load_config()
  
              if config_section not in config:
                  return f"❌ Error: Config section '{config_section}' not found\n💡 Available sections: {', '.join(config.keys())}"
  
              # Store old value for comparison
              old_value = config[config_section].get(config_key, "<not set>")
  
              # Update the value
              config[config_section][config_key] = config_value
  
              # Write updated config
              with open(config_path, 'w', encoding='utf-8') as f:
                  json.dump(config, f, indent=2, ensure_ascii=False)
  
              message = f"✅ Configuration updated successfully!\n\n"
              message += f"📁 Section: {config_section}\n"
              message += f"🔑 Key: {config_key}\n"
              message += f"📤 Old value: {old_value}\n"
              message += f"📥 New value: {config_value}"
  
          else:
              return "❌ Error: Must provide either 'full_config' or both 'config_section' and 'config_key'\n💡 Choose: Section-specific update (config_section + config_key + config_value) OR Full replacement (full_config)"
  
          # Reload configuration in current session
          new_config = load_config()
  
          # Reinitialize security if security section was updated
          if (config_section == "security" and config_key == "allowed_base_directory") or full_config:
              init_security(new_config["security"]["allowed_base_directory"])
  
          # Reinitialize Elasticsearch if elasticsearch section was updated
          if (config_section == "elasticsearch") or full_config:
              init_elasticsearch(new_config)
              reset_es_client()
  
          return message + f"\n\n💡 Configuration reloaded automatically - all components reinitialized"
  
      except json.JSONDecodeError as e:
          return f"❌ JSON Error: Invalid JSON format in full_config\n🔍 Details: {str(e)}\n💡 Check JSON syntax and structure"
      except Exception as e:
          return _format_admin_error(e, "update configuration", f"config_section={config_section}, config_key={config_key}")

• src/main_server.py:78-82 (registration)

  Registers the admin_server_app (containing update_config and other admin tools) by mounting it into the main FastMCP application. This makes the tools available server-wide, potentially with 'admin_' prefix as per comments.

  # Mount Administrative operations server with 'admin' prefix
  # This provides: admin_get_config, admin_update_config, admin_server_status, etc.
  app.mount(admin_server_app)
  
  # Mount Prompt server for AgentKnowledgeMCP guidance

• src/confirmation/confirmation.py:91-95 (registration)

  Configuration rule in confirmation system that requires user confirmation for 'update_config' tool execution, with 60-minute timeout.

  "admin_tools": {
      "tools": ["update_config", "reset_config", "setup_elasticsearch"],
      "require_confirmation": True,
      "timeout_minutes": 60
  },

• src/admin/admin_server.py:93-98 (schema)

  Input schema definition using Pydantic Annotated fields for config_section, config_key, config_value (for partial updates) or full_config (for complete replacement).

  async def update_config(
      config_section: Annotated[Optional[str], Field(description="The top-level section of the config to update (e.g., 'security')")] = None,
      config_key: Annotated[Optional[str], Field(description="The key within the section to update (e.g., 'allowed_base_directory')")] = None,
      config_value: Annotated[Optional[str], Field(description="The new value for the specified key")] = None,
      full_config: Annotated[Optional[Dict[str, Any]], Field(description="Full configuration object to save. Replaces the entire config")] = None
  ) -> str: